[DE-627] batch writes (#23)

* batch size config

* batch writes

* test fixes

* test fixes

* enabled snapshots-repo

* tests

* disabled snapshot repo

* fixed endpoints rotation

Author: rashtao

ChangeLog.md

@@ -5,6 +5,8 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) a
 
 ## [Unreleased]
 
+- batch writes (DE-627, #23)
+
 ## [1.0.0] - 2023-05-26
 
 - initial release

demo/README.md

@@ -5,7 +5,7 @@
 Set package version:
 
 ```shell
-export PACKAGE_VERSION=1.0.0
+export PACKAGE_VERSION=1.1.0
 ```
 
 Create the Docker network:
@@ -69,7 +69,7 @@ The messages produced can be checked at [http://172.28.0.1:8080/topics/orders](h
 Create db collection:
 
 ```shell
-curl -u root:test http://172.28.0.1:8529/_api/collection -d '{"name": "orders"}'
+curl -u root:test http://172.28.0.1:8529/_api/collection -d '{"name": "orders", "numberOfShards": 3}'
 ```
 
 Explore configuration options in the console at [http://172.28.0.1:8080/connect-clusters/kafka-connect/create-connector](http://172.28.0.1:8080/connect-clusters/kafka-connect/create-connector)

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.arangodb</groupId>
     <artifactId>kafka-connect-arangodb</artifactId>
-    <version>1.0.0</version>
+    <version>1.1.0</version>
     <packaging>jar</packaging>
     <inceptionYear>2023</inceptionYear>
 
@@ -62,7 +62,7 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <kafka.version>3.4.1</kafka.version>
         <confluent.version>7.4.0</confluent.version>
-        <arangodb.version>7.1.0</arangodb.version>
+        <arangodb.version>7.3.0</arangodb.version>
         <jackson.version>2.13.5</jackson.version>
         <slf4j.version>1.7.36</slf4j.version>
     </properties>
@@ -78,6 +78,16 @@
             <name>Confluent</name>
             <url>https://packages.confluent.io/maven/</url>
         </repository>
+        <repository>
+            <id>snapshots-repo</id>
+            <url>https://oss.sonatype.org/content/repositories/snapshots</url>
+            <releases>
+                <enabled>false</enabled>
+            </releases>
+            <snapshots>
+                <enabled>false</enabled>
+            </snapshots>
+        </repository>
     </repositories>
 
     <dependencies>

src/main/java/com/arangodb/kafka/ArangoSinkConnector.java

@@ -34,6 +34,8 @@
 public class ArangoSinkConnector extends SinkConnector {
     private static final Logger LOG = LoggerFactory.getLogger(SinkConnector.class);
     private Map<String, String> config;
+    private boolean acquireHostList;
+    private List<HostDescription> initialEndpoints;
     private HostListMonitor hostListMonitor;
 
     @Override
@@ -54,8 +56,11 @@ public void start(Map<String, String> props) {
             throw new ConnectException(e);
         }
 
-        hostListMonitor = new HostListMonitor(sinkConfig, context);
-        if (sinkConfig.isAcquireHostListEnabled()) {
+        acquireHostList = sinkConfig.isAcquireHostListEnabled();
+        initialEndpoints = sinkConfig.getEndpoints();
+
+        if (acquireHostList) {
+            hostListMonitor = new HostListMonitor(sinkConfig, context);
             hostListMonitor.start();
         }
     }
@@ -67,7 +72,7 @@ public Class<? extends Task> taskClass() {
 
     @Override
     public List<Map<String, String>> taskConfigs(int maxTasks) {
-        List<HostDescription> endpoints = new ArrayList<>(hostListMonitor.getEndpoints());
+        List<HostDescription> endpoints = new ArrayList<>(acquireHostList ? hostListMonitor.getEndpoints() : initialEndpoints);
         int rotationDistance = endpoints.size() / maxTasks;
         if (rotationDistance == 0) {
             rotationDistance = 1;
@@ -89,8 +94,10 @@ public List<Map<String, String>> taskConfigs(int maxTasks) {
 
     @Override
     public void stop() {
-        LOG.info("stopping ArangoSinkConnector");
-        hostListMonitor.stop();
+        if (acquireHostList) {
+            LOG.info("stopping ArangoSinkConnector");
+            hostListMonitor.stop();
+        }
     }
 
     @Override