Make commit after every batch of inserts optional, with the default not to commit.

For at least once semantics, it is not necessary to commit after every call to put(), as Kafka
Connect will commit the kafka Offsets periodically. Doing it periodically will lead to less
writes to the __consumer_offsets topic, and hence, less load on the Kafka brokers.

The default is false, i.e not to commit offset after every set of INSERT operations. This
is a change in behaviour of Kafka Connect ScyllaDB, but it leads to a significant decrease
in number of produce requests on the Kafka brokers, which in production environments
with high load is quite visible in Broker CPU consumption metrics.

Author: forsberg

config/scylladb-sink-quickstart.properties

@@ -57,6 +57,7 @@ scylladb.keyspace=<keyspace-name>
 #scylladb.execute.timeout.ms=30000
 #scylladb.ttl=null
 #scylladb.offset.storage.table.enable=true
+#offset.commit.after.every.insert=false
 
 ### Converter configs(AVRO):
 #key.converter=io.confluent.connect.avro.AvroConverter

src/main/java/io/connect/scylladb/ScyllaDbSinkConnectorConfig.java

@@ -63,6 +63,7 @@ public class ScyllaDbSinkConnectorConfig extends AbstractConfig {
   public final List<String> cipherSuites;
   public final File certFilePath;
   public final File privateKeyPath;
+  public final boolean requestCommitAfterEveryInsert;
 
   private static final Pattern TOPIC_KS_TABLE_SETTING_PATTERN =
           Pattern.compile("topic\\.([a-zA-Z0-9._-]+)\\.([^.]+|\"[\"]+\")\\.([^.]+|\"[\"]+\")\\.(mapping|consistencyLevel|ttlSeconds|deletesEnabled)$");
@@ -160,6 +161,8 @@ public ScyllaDbSinkConnectorConfig(Map<?, ?> originals) {
       TopicConfigs topicConfigs = new TopicConfigs(topicWiseConfig.getValue(), this);
       topicWiseConfigs.put(topicWiseConfig.getKey(), topicConfigs);
     }
+
+    this.requestCommitAfterEveryInsert = getBoolean(OFFSET_COMMIT_AFTER_EVERY_INSERT);
   }
 
   public static final String PORT_CONFIG = "scylladb.port";
@@ -303,6 +306,13 @@ public ScyllaDbSinkConnectorConfig(Map<?, ?> originals) {
           + "inserting records into ScyllDB and continues to process next set of records, "
           + "available in the kafka topics.";
 
+  public static final String OFFSET_COMMIT_AFTER_EVERY_INSERT = "offset.commit.after.every.insert";
+  public static final boolean OFFSET_COMMIT_AFTER_EVERY_INSERT_DEFAULT = false;
+  public static final String OFFSET_COMMIT_AFTER_EVERY_INSERT_DOC = "Configure code to request Kafka Offset Commit " 
+          + "After every batch of insertion to Scylla. This will mean less repeated inserts after recovery " 
+          + "but will increase load on Kafka cluster. If set to false, Kafka Connect will commit offsets "
+          + "periodically instead.";
+
   public static final String SCYLLADB_GROUP = "ScyllaDB";
   public static final String CONNECTION_GROUP = "Connection";
   public static final String SSL_GROUP = "SSL";
@@ -626,6 +636,17 @@ public static ConfigDef config() {
                     ConfigDef.Width.NONE,
                     BEHAVIOR_ON_ERROR_DISPLAY,
                     new ListRecommender(Arrays.asList(toStringArray(BehaviorOnError.values())))
+            )
+            .define(
+                OFFSET_COMMIT_AFTER_EVERY_INSERT,
+                ConfigDef.Type.BOOLEAN,
+                OFFSET_COMMIT_AFTER_EVERY_INSERT_DEFAULT,
+                ConfigDef.Importance.MEDIUM,
+                OFFSET_COMMIT_AFTER_EVERY_INSERT_DOC,
+                WRITE_GROUP,
+                5,
+                ConfigDef.Width.SHORT,
+                "Request Kafka Offset Commit after every Scylla Write"
             );
   }
 

src/main/java/io/connect/scylladb/ScyllaDbSinkTask.java

@@ -128,7 +128,9 @@ public void put(Collection<SinkRecord> records) {
           ResultSet resultSet =
                   future.getUninterruptibly(this.config.statementTimeoutMs, TimeUnit.MILLISECONDS);
         }
-        context.requestCommit();
+        if(config.requestCommitAfterEveryInsert) {
+          context.requestCommit();
+        }
         // TODO : Log the records that fail in Queue/Kafka Topic.
       } catch (TransportException ex) {
         log.debug("put() - Setting clusterValid = false", ex);

src/test/java/io/connect/scylladb/integration/ScyllaDbSinkConnectorIT.java

@@ -69,6 +69,7 @@ static Map<String, String> settings() {
     result.put(ScyllaDbSinkConnectorConfig.CONTACT_POINTS_CONFIG, SCYLLA_DB_CONTACT_POINT);
     result.put(ScyllaDbSinkConnectorConfig.PORT_CONFIG, String.valueOf(SCYLLA_DB_PORT));
     result.put(ScyllaDbSinkConnectorConfig.KEYSPACE_REPLICATION_FACTOR_CONFIG, "1");
+    result.put(ScyllaDbSinkConnectorConfig.OFFSET_COMMIT_AFTER_EVERY_INSERT, "true");
     return result;
   }
 
@@ -181,6 +182,61 @@ public void insert() {
     verify(this.sinkTaskContext, times(1)).assignment();
   }
 
+  @Test
+  public void insertWithoutRequestCommit() {
+    final Map<String, String> settings = settings();
+    settings.put(ScyllaDbSinkConnectorConfig.OFFSET_COMMIT_AFTER_EVERY_INSERT, "false");
+    connector = new ScyllaDbSinkConnector();
+    connector.start(settings);
+    final String topic = "insertTestingWORC";
+    when(this.sinkTaskContext.assignment()).thenReturn(ImmutableSet.of(new TopicPartition(topic, 3)));
+    this.task.start(settings);
+    List<SinkRecord> finalRecordsList = new ArrayList<>();
+    for (int i = 0; i< 1000; i++) {
+      List<SinkRecord> records = ImmutableList.of(
+              write(
+                      topic,
+                      struct("key",
+                              "id", Schema.Type.INT64, true, 12345L+ i
+                      ), struct("key",
+                              "id", Schema.Type.INT64, true, 12345L + i,
+                              "firstName", Schema.Type.STRING, true, "test",
+                              "lastName", Schema.Type.STRING, true, "user",
+                              "age", Schema.Type.INT64, true, 1234L + i
+                      )
+              ),
+              write(topic,
+                      null,
+                      asMap(
+                              struct("key",
+                                      "id", Schema.Type.INT64, true, 67890L + i
+                              )
+                      ),
+                      null,
+                      asMap(
+                              struct("key",
+                                      "id", Schema.Type.INT64, true, 67890L + i,
+                                      "firstName", Schema.Type.STRING, true, "another",
+                                      "lastName", Schema.Type.STRING, true, "user",
+                                      "age", Schema.Type.INT64, true, 10L + i
+                              )
+                      )
+              )
+      );
+      finalRecordsList.addAll(records);
+    }
+    this.validations = finalRecordsList.stream()
+            .map(RowValidator::of)
+            .collect(Collectors.toList());
+    this.task.put(finalRecordsList);
+    Boolean tableExists = IsOffsetStorageTableExists(SCYLLADB_OFFSET_TABLE);
+    assertEquals(true, tableExists);
+    verify(this.sinkTaskContext, times(0)).requestCommit();
+    verify(this.sinkTaskContext, times(1)).assignment();
+  }
+
+
+
   @Test
   @Disabled
   public void insertWithTopicMapping() {

src/test/java/io/connect/scylladb/integration/ScyllaNativeTypesIT.java

@@ -79,6 +79,7 @@ static Cluster.Builder clusterBuilder() {
             .put(ScyllaDbSinkConnectorConfig.CONTACT_POINTS_CONFIG, SCYLLA_DB_CONTACT_POINT)
             .put(ScyllaDbSinkConnectorConfig.PORT_CONFIG, String.valueOf(SCYLLA_DB_PORT))
             .put(ScyllaDbSinkConnectorConfig.KEYSPACE_REPLICATION_FACTOR_CONFIG, "1")
+            .put(ScyllaDbSinkConnectorConfig.OFFSET_COMMIT_AFTER_EVERY_INSERT, "true")
             .build());
 
     @BeforeAll