Repository: microsoft/kafka-connect-cosmosdb
Pull request #171 (2020-12-01 22:09:00)

beta-1 release
merging latest dev changes in to main ahead of beta-1 release


Patch:
diff --git a/.devcontainer/Dockerfile b/.devcontainer/Dockerfile
deleted file mode 100644
index 8b13789..0000000
--- a/.devcontainer/Dockerfile
+++ /dev/null
@@ -1,1 +0,0 @@
-
diff --git a/doc/README_Sink.md b/doc/README_Sink.md
index 96df75f..77f3d28 100644
--- a/doc/README_Sink.md
+++ b/doc/README_Sink.md
@@ -6,6 +6,8 @@ The connector polls data from Kafka to write to collection(s) in the database ba
 
 ## Installation
 
+### Install from the [Confluent Hub Client](https://docs.confluent.io/current/connect/managing/confluent-hub/client.html#confluent-hub-client)
+
 ### Install Connector Manually
 Download and extract the ZIP file for your connector and follow the manual connector installation [instructions](https://docs.confluent.io/current/connect/managing/install.html#install-connector-manually)
 
@@ -236,4 +238,4 @@ confluent local stop
 
 Delete the created Azure Cosmos DB service and its resource group using Azure CLI.
 ```bash
-```
+```
\ No newline at end of file
diff --git a/doc/README_Source.md b/doc/README_Source.md
index db6d40f..68adb0b 100644
--- a/doc/README_Source.md
+++ b/doc/README_Source.md
@@ -11,15 +11,15 @@ Download and extract the ZIP file for your connector and follow the manual conne
 
 At the moment the following settings can be configured by means of the *connector.properties* file. For a config file containing default settings see [this example](../src/integration-test/resources/source.config.json).
 
-All configuation properties for the source connector are prefixed with *connect.cosmosdb. e.g. connect.cosmosdb.databasename*
-
-
 | Name                                           | Description                                                                                          | Type    | Default                                                                       | Valid Values                                                                                                     | Importance |
 |------------------------------------------------|------------------------------------------------------------------------------------------------------|---------|-------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------|------------|
-| databasename                             | name of the database to write to                                                              | string  |
-| master.key | the configured master key for Cosmos DB | string |
-| endpoint | the endpoint for the Cosmos DB Account | uri | 
-| task.poll.interval |  | int
+| connect.cosmosdb.databasename                             | name of the database to write to                                                              | string  |
+| connect.cosmosdb.master.key | the configured master key for Cosmos DB | string |
+| connect.cosmosdb.connection.endpoint | the endpoint for the Cosmos DB Account | uri |
+| connect.cosmosdb.containers.topicmap | comma separeted topic to collection mapping, eg. topic1#coll1,topic2#coll2 | string
+| connect.cosmosdb.containers | list of collections to monitor | string
+| connect.cosmosdb.changefeed.startFromBeginning | set if the change feed should start from beginning | boolean | true 
+| connect.cosmosdb.task.poll.interval | interval to poll the changefeedcontainer for changes  | int
 
 ### Kafka Connect Converter Configuration
 
@@ -47,8 +47,7 @@ It is possible to have the Source connector output CSV string by using StringCon
 ## Quick Start
 
 ### Prerequisites
-* [Confluent Platform](https://docs.confluent.io/current/installation/index.html#installation-overview)
-* [Confluent CLI](https://docs.confluent.io/current/cli/installing.html#cli-install) (requires separate installation)
+* [Event Hub](https://docs.microsoft.com/pt-br/azure/event-hubs/event-hubs-for-kafka-ecosystem-overview)
 * [Azure CLI](https://docs.microsoft.com/cli/azure/install-azure-cli) (requires separate installation)
 
 ### Create Azure Cosmos DB Instance, Database and Collection
@@ -64,67 +63,81 @@ Create a new Azure Resource Group for this quickstart, then add to it a Cosmos D
 
 ```
 
-### Install connector
-```bash
-# install the connector (run from your CP installation directory)
+### Build the connector
+Download the repository and build it
+```shell script
+mvn clean package
+```
 
-# start conluent platform
-confluent local start
+### Using docker
+Create variables.env file
+```properties
+CONNECT_BOOTSTRAP_SERVERS={YOUR.EVENTHUBS.FQDN}:9093 # e.g. namespace.servicebus.windows.net:9093
+CONNECT_REST_PORT=8083
+CONNECT_GROUP_ID=connect-cluster-group
 
-```
+CONNECT_CONFIG_STORAGE_TOPIC=connect-cluster-configs
+CONNECT_OFFSET_STORAGE_TOPIC=connect-cluster-offsets
+CONNECT_STATUS_STORAGE_TOPIC=connect-cluster-status
 
-### Insert document in to Cosmos DB
+CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR=1
+CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR=1
+CONNECT_STATUS_STORAGE_REPLICATION_FACTOR=1
 
-Insert a new document in to Cosmos DB using the Azure CLI
-```bash
-```
+CONNECT_KEY_CONVERTER=org.apache.kafka.connect.json.JsonConverter
+CONNECT_VALUE_CONVERTER=org.apache.kafka.connect.json.JsonConverter
+CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE=false
 
-Verify the record is in Cosmos DB
-```bash
+CONNECT_INTERNAL_KEY_CONVERTER=org.apache.kafka.connect.json.JsonConverter
+CONNECT_INTERNAL_VALUE_CONVERTER=org.apache.kafka.connect.json.JsonConverter
+
+CONNECT_REST_ADVERTISED_HOST_NAME=connect
+
+CONNECT_SECURITY_PROTOCOL=SASL_SSL
+CONNECT_SASL_MECHANISM=PLAIN
+CONNECT_SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{YOUR.EVENTHUBS.CONNECTION.STRING}";
+
+CONNECT_PRODUCER_SECURITY_PROTOCOL=SASL_SSL
+CONNECT_PRODUCER_SASL_MECHANISM=PLAIN
+CONNECT_PRODUCER_SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{YOUR.EVENTHUBS.CONNECTION.STRING}";
+
+CONNECT_CONSUMER_SECURITY_PROTOCOL=SASL_SSL
+CONNECT_CONSUMER_SASL_MECHANISM=PLAIN
+CONNECT_CONSUMER_SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{YOUR.EVENTHUBS.CONNECTION.STRING}";
+
+CONNECT_PLUGIN_PATH=/usr/share/java,/etc/kafka-connect/jars
+CLASSPATH=/usr/share/java/monitoring-interceptors/monitoring-interceptors-5.5.0.jar
 ```
 
-### Load the connector
-Create *azure-cosmosdb.json* file with the following contents: 
+```shell script
+mkdir -p /tmp/quickstart/jars
+cp target/cosmosdb.kafka.connect-1.0-SNAPSHOT-jar-with-dependencies.jar /tmp/quickstart/jars
 
-```javascript
-{
-  "name": "azure-cosmosdb",
-  "config": {
-    "tasks.max": "1",
-    "connector.class": "com.microsoft.azure.cosmosdb.kafka.connect.source.CosmosDBSourceConnector",
-    "key.converter": "io.confluent.connect.avro.AvroConverter",
-    "key.converter.schema.registry.url": "http://localhost:8081",
-    "value.converter": "io.confluent.connect.avro.AvroConverter",
-    "value.converter.schema.registry.url": "http://localhost:8081",
-    "confluent.topic.bootstrap.servers": "localhost:9092",
-    "confluent.topic.replication.factor": "1"
-  }
-}
+docker run --name=kafka-connect-cosmosdb --net=host --env-file variables.env -v /tmp/quickstart/jars:/etc/kafka-connect/jars confluentinc/cp-kafka-connect:6.0.0
 ```
 
-Load the Azure Cosmos DB Source Connector
-```bash
-confluent local load azure-cosmosdb -- -d path/to/azure-cosmosdb.json
+### Create the connector
+```shell script
+curl -X POST \
+  -H "Content-Type: application/json" \
+  --data '{ "name": "quickstart-cosmosdb-source", "config": { "connector.class": "com.microsoft.azure.cosmosdb.kafka.connect.source.CosmosDBSourceConnector", "tasks.max": 1, "connect.cosmosdb.databasename": "<DB_NAME>", "connect.cosmosdb.master.key": "<KEY>", "connect.cosmosdb.connection.endpoint": "<URL_COSMOS>", "connect.cosmosdb.task.poll.interval": "10000", "connect.cosmosdb.containers.topicmap": "<TOPIC_MAPPING>", "connect.cosmosdb.containers": "<COLLECTIONS>" } }' \
+  http://localhost:8083/connectors
 ```
+### Insert document in to Cosmos DB
 
-Confirm that the connector is in a RUNNING state.
+Insert a new document in to Cosmos DB using the Azure CLI
 ```bash
-confluent local status azure-cosmosdb
 ```
 
-Confirm that the messages were delivered to the result topic in Kafka
+Verify the record is in Cosmos DB
 ```bash
 ```
 
 ### Cleanup
-Delete the connector
-```bash
-confluent local unload azure-cosmosdb
-```
 
-Stop Confluent Platform
+Stop the container
 ```bash
-confluent local stop
+docker stop kafka-connect-cosmosdb
 ```
 
 Delete the created Azure Cosmos DB service and its resource group using Azure CLI.
diff --git a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/BooleanSetting.java b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/BooleanSetting.java
new file mode 100644
index 0000000..5047b1f
--- /dev/null
+++ b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/BooleanSetting.java
@@ -0,0 +1,37 @@
+package com.microsoft.azure.cosmosdb.kafka.connect;
+
+import org.apache.commons.lang3.BooleanUtils;
+import org.apache.kafka.common.config.ConfigDef;
+
+import java.util.Optional;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+/**
+ * Represents a setting that must be a boolean
+ */
+public class BooleanSetting extends Setting {
+
+    public BooleanSetting(String name, String documentation, String displayName, Boolean defaultValue, Consumer<Boolean> modifier, Supplier<Boolean> accessor) {
+        super(name, documentation, displayName, defaultValue, booleanModifier(modifier), booleanAccessor(accessor));
+    }
+
+    protected static Supplier<String> booleanAccessor(Supplier<Boolean> baseMethod) {
+        return () -> Optional.ofNullable(baseMethod.get()).map(b -> Boolean.toString(b)).orElse(null);
+    }
+
+    protected static Consumer<String> booleanModifier(Consumer<Boolean> baseMethod) {
+        return (String s) -> Optional.ofNullable(s).map(BooleanUtils::toBoolean).ifPresent(baseMethod);
+    }
+
+    @Override
+    protected ConfigDef.Type getKafkaConfigType() {
+        return ConfigDef.Type.BOOLEAN;
+    }
+
+    @Override
+    public boolean isValid(Object value) {
+        //Null (unset) is ok
+        return value == null || super.isValid(value);
+    }
+}
diff --git a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/Settings.java b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/Settings.java
index dfaaee2..52c64b7 100644
--- a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/Settings.java
+++ b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/Settings.java
@@ -34,8 +34,8 @@ public class Settings {
                     "Task poll interval", SettingDefaults.TASK_POLL_INTERVAL, this::setTaskPollInterval, this::getTaskPollInterval),
             new Setting(PREFIX+".connection.endpoint", "The Cosmos DB endpoint.", "CosmosDB Endpoint", this::setEndpoint, this::getEndpoint),
             new Setting(PREFIX+".master.key", "The connection master key.", "Master Key", this::setKey, this::getKey),
-            new Setting(PREFIX+".cosmosdb.databasename", "The Cosmos DB target database.", "CosmosDB Database Name", this::setDatabaseName, this::getDatabaseName),
-            new Setting(PREFIX+".containers.topicmap", "A comma delimited list of collections mapped to their partitions. Formatted coll1#topic1,coll2#topic2.",
+            new Setting(PREFIX+".databasename", "The Cosmos DB target database.", "CosmosDB Database Name", this::setDatabaseName, this::getDatabaseName),
+            new Setting(PREFIX+".containers.topicmap", "A comma delimited list of collections mapped to their partitions. Formatted topic1#coll1,topic2#coll2.",
                     "Topic-Container map", value -> this.setTopicContainerMap(TopicContainerMap.deserialize(value)), ()->this.getTopicContainerMap().serialize()),
             new Setting(PREFIX+".containers", "A comma delimited list of source/target container names.",
                     "Collection Names List", this::setContainerList, this::getContainerList)
diff --git a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/CosmosDBSourceTask.java b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/CosmosDBSourceTask.java
index f53e191..3ab5675 100644
--- a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/CosmosDBSourceTask.java
+++ b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/CosmosDBSourceTask.java
@@ -103,9 +103,8 @@ public class CosmosDBSourceTask extends SourceTask {
             }
 
             if (records.size() > 0 || System.currentTimeMillis() > maxWaitTime) {
-                if(logger.isDebugEnabled()) {
-                    logger.debug(String.format("Sending %d documents.", records.size()));
-                }
+                logger.debug("Sending {} documents.", records.size());
+
                 break;
             }
         }
@@ -136,15 +135,16 @@ public class CosmosDBSourceTask extends SourceTask {
                 .endpoint(this.settings.getEndpoint())
                 .key(this.settings.getKey())
                 .consistencyLevel(ConsistencyLevel.SESSION)
+                .contentResponseOnWriteEnabled(true)
                 .buildAsyncClient();
     }
 
     private ChangeFeedProcessor getChangeFeedProcessor(String hostName, CosmosAsyncContainer feedContainer, CosmosAsyncContainer leaseContainer) {
-        logger.info("Creating Change Feed Processor for " + hostName + ".");
+        logger.info("Creating Change Feed Processor for {}.", hostName);
         ChangeFeedProcessorOptions changeFeedProcessorOptions = new ChangeFeedProcessorOptions();
         changeFeedProcessorOptions.setFeedPollDelay(Duration.ofMillis(this.settings.getTaskPollInterval()));
         changeFeedProcessorOptions.setMaxItemCount(this.settings.getTaskBatchSize().intValue());
-        changeFeedProcessorOptions.setStartFromBeginning(true);
+        changeFeedProcessorOptions.setStartFromBeginning(this.settings.isStartFromBeginning());
 
         return new ChangeFeedProcessorBuilder()
                 .options(changeFeedProcessorOptions)
@@ -161,9 +161,8 @@ public class CosmosDBSourceTask extends SourceTask {
             // Blocks for each transfer till it is processed by the poll method.
             // If we fail before checkpointing then the new worker starts again.
             try {
-                if(logger.isDebugEnabled()) {
-                    logger.debug("Queuing document : " + document.toString());
-                }
+                logger.debug("Queuing document : {}", document.toString());
+
                 this.queue.transfer(document);
             } catch (InterruptedException e) {
                 logger.error("Interrupted in changeFeedReader.", e);
@@ -181,23 +180,16 @@ public class CosmosDBSourceTask extends SourceTask {
         try {
             leaseContainerResponse = leaseCollection.read().block();
 
-        } catch (RuntimeException ex) {
-            // Swallowing exceptions when the type is CosmosClientException and statusCode is 404
-            if (ex instanceof CosmosException) {
-                CosmosException cosmosClientException = (CosmosException) ex;
-
-                if (cosmosClientException.getStatusCode() == 404) {
-                    throw ex;
-                }
-            } else {
+        } catch (CosmosException ex) {
+            // Swallowing exceptions when the type is CosmosException and statusCode is 404
+            if (ex.getStatusCode() != 404) {
                 throw ex;
             }
-
-            logger.info("Lease container does not exist" + ex.getMessage());
+            logger.info("Lease container does not exist {}", ex.getMessage());
         }
 
         if (leaseContainerResponse == null) {
-            logger.info(String.format("Creating the Lease container : %s", leaseCollectionName));
+            logger.info("Creating the Lease container : {}", leaseCollectionName);
             CosmosContainerProperties containerSettings = new CosmosContainerProperties(leaseCollectionName, "/id");
             ThroughputProperties throughputProperties = ThroughputProperties.createManualThroughput(400);
             CosmosContainerRequestOptions requestOptions = new CosmosContainerRequestOptions();
diff --git a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/SourceSettingDefaults.java b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/SourceSettingDefaults.java
new file mode 100644
index 0000000..afa6cd7
--- /dev/null
+++ b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/SourceSettingDefaults.java
@@ -0,0 +1,6 @@
+package com.microsoft.azure.cosmosdb.kafka.connect.source;
+
+public class SourceSettingDefaults {
+
+    public static final Boolean CHANGE_FEED_START_FROM_BEGINNING = true;
+}
diff --git a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/SourceSettings.java b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/SourceSettings.java
index 2265e96..252c17e 100644
--- a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/SourceSettings.java
+++ b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/SourceSettings.java
@@ -1,5 +1,6 @@
 package com.microsoft.azure.cosmosdb.kafka.connect.source;
 
+import com.microsoft.azure.cosmosdb.kafka.connect.BooleanSetting;
 import com.microsoft.azure.cosmosdb.kafka.connect.Setting;
 import com.microsoft.azure.cosmosdb.kafka.connect.Settings;
 import org.apache.commons.collections4.ListUtils;
@@ -11,15 +12,17 @@ import java.util.List;
  * Contains settings for the CosmosDB Kafka Source Connector
  */
 public class SourceSettings extends Settings {
-    private String assignedPartitions;
     private String postProcessor;
+    private boolean startFromBeginning;
     private final List<Setting> sourceSettings = Arrays.asList(
             new Setting(Settings.PREFIX + ".source.post-processor", "Comma-separated list of Source Post-Processor class names to use for post-processing",
                     "Source post-processor", this::setPostProcessor, this::getPostProcessor),
             new Setting(Settings.PREFIX + ".assigned.container", "The CosmosDB Feed Container assigned to the task.",
                     "Assigned Container", this::setAssignedContainer, this::getAssignedContainer),
-          new Setting(Settings.PREFIX + ".worker.name", "The CosmosDB worker name.",
-            "Worker name", this::setWorkerName, this::getWorkerName)
+            new Setting(Settings.PREFIX + ".worker.name", "The CosmosDB worker name.",
+                    "Worker name", this::setWorkerName, this::getWorkerName),
+            new BooleanSetting(Settings.PREFIX + ".changefeed.startFromBeginning", "If the change feed should start from beginning",
+                    "Change Feed start from beginning", SourceSettingDefaults.CHANGE_FEED_START_FROM_BEGINNING, this::setStartFromBeginning, this::isStartFromBeginning)
 
     );
 
@@ -52,6 +55,9 @@ public class SourceSettings extends Settings {
         this.postProcessor = postProcessor;
     }
 
+    public boolean isStartFromBeginning() { return this.startFromBeginning; }
+
+    public void setStartFromBeginning(boolean startFromBeginning) { this.startFromBeginning = startFromBeginning;  }
 
     @Override
     protected List<Setting> getAllSettings() {


Test patch:
diff --git a/src/integration-test/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/PostToKafkaTest.java b/src/integration-test/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/PostToKafkaTest.java
index cc7d342..7c32f61 100644
--- a/src/integration-test/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/PostToKafkaTest.java
+++ b/src/integration-test/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/PostToKafkaTest.java
@@ -61,7 +61,7 @@ public class PostToKafkaTest {
         topic = StringUtils.substringBefore(topicContainerMap, "#");
 
         //Get the Cosmos container for reading messages.
-        databaseName = config.get("connect.cosmosdb.cosmosdb.databasename").textValue();
+        databaseName = config.get("connect.cosmosdb.databasename").textValue();
 
         CosmosClient cosmosClient = new CosmosClientBuilder()
                 .endpoint(config.get("connect.cosmosdb.connection.endpoint").textValue())
diff --git a/src/integration-test/resources/sink.config.json b/src/integration-test/resources/sink.config.json
index 435c4df..979644a 100644
--- a/src/integration-test/resources/sink.config.json
+++ b/src/integration-test/resources/sink.config.json
@@ -7,7 +7,7 @@
     "value.converter": "JsonConverter",
     "value.converter.schemas.enable": "false",
     "errors.tolerance": "all",
-    "connect.cosmosdb.cosmosdb.databasename": "",
+    "connect.cosmosdb.databasename": "",
     "connect.cosmosdb.master.key": "",
     "connect.cosmosdb.connection.endpoint": "",
     "connect.cosmosdb.containers.topicmap": ""
diff --git a/src/integration-test/resources/source.config.json b/src/integration-test/resources/source.config.json
index fc71d0f..93f3aed 100644
--- a/src/integration-test/resources/source.config.json
+++ b/src/integration-test/resources/source.config.json
@@ -4,7 +4,7 @@
     "connector.class": "com.microsoft.azure.cosmosdb.kafka.connect.source.CosmosDBSourceConnector",
     "tasks.max": "3",
     "topics":"CartContainerByState",
-    "connect.cosmosdb.cosmosdb.databasename": "StoreDatabase",
+    "connect.cosmosdb.databasename": "StoreDatabase",
     "connect.cosmosdb.master.key": "",
     "connect.cosmosdb.connection.endpoint":"",
     "connect.cosmosdb.containers.topicmap" : "CartContainerByState#CartContainer",
diff --git a/src/test/java/com/microsoft/azure/cosmosdb/kafka/connect/source/SourceSettingsTest.java b/src/test/java/com/microsoft/azure/cosmosdb/kafka/connect/source/SourceSettingsTest.java
index 3a7dec9..c82029e 100644
--- a/src/test/java/com/microsoft/azure/cosmosdb/kafka/connect/source/SourceSettingsTest.java
+++ b/src/test/java/com/microsoft/azure/cosmosdb/kafka/connect/source/SourceSettingsTest.java
@@ -6,6 +6,7 @@ import org.junit.Test;
 import java.util.HashMap;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
 public class SourceSettingsTest {
     /**
@@ -20,12 +21,14 @@ public class SourceSettingsTest {
         source.put(Settings.PREFIX + ".task.timeout", "444");
         source.put(Settings.PREFIX + ".task.poll.interval", "787");
         source.put(Settings.PREFIX + ".containers.topicmap", "mytopic666#mycontainer555");
+        source.put(Settings.PREFIX + ".changefeed.startFromBeginning", "true");
         SourceSettings sourceSettings = new SourceSettings();
         sourceSettings.populate(source);
 
         assertEquals("foobar", sourceSettings.getPostProcessor());
         assertEquals(444L, (long) sourceSettings.getTaskTimeout());
         assertEquals(666L, (long) sourceSettings.getTaskBufferSize());
+        assertTrue(sourceSettings.isStartFromBeginning());
         assertEquals("mytopic666", sourceSettings.getTopicContainerMap().getTopicForContainer("mycontainer555").get());
         assertEquals("mycontainer555", sourceSettings.getTopicContainerMap().getContainerForTopic("mytopic666").get());
     }
@@ -35,6 +38,7 @@ public class SourceSettingsTest {
         HashMap<String, String> source = new HashMap<>();
         SourceSettings sourceSettings = new SourceSettings();
         sourceSettings.populate(source);
+        assertTrue(sourceSettings.isStartFromBeginning());
         assertEquals(5000L, (long) sourceSettings.getTaskTimeout());
         assertEquals(10000L, (long) sourceSettings.getTaskBufferSize());
         assertEquals(100L, (long) sourceSettings.getTaskBatchSize());