Repository: microsoft/kafka-connect-cosmosdb (java)

Issue:
Option to start the changefeed at current time
**## Problem Statement**
The startFromBeginning option is hard coded today, making it impossible to start the change feed ignoring previous changes

**## Proposed Solution**
Create a new setting to make it possible to start the change feed with the default behaviour of ignoring previous changes

**## Additional context**
https://github.com/microsoft/kafka-connect-cosmosdb/blob/bf072b6032248a6ed1101e9ce0a67304f65cb768/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/CosmosDBSourceTask.java#L147

Pull request:
Create new setting to the change feed
## Type of PR

- [x] Documentation changes
- [x] Code changes
- [x] Test changes
- [ ] CI-CD changes
- [ ] GitHub Template changes

## Purpose of PR
Today the option to start from beginning on the change feed is hard coded to true. A new setting make it possible to set it to false on demand was created

## Review notes

## Issues Closed or Referenced
- Closes #164


Patch:
diff --git a/doc/README_Source.md b/doc/README_Source.md
index 5a8ad05..356c9a6 100644
--- a/doc/README_Source.md
+++ b/doc/README_Source.md
@@ -18,6 +18,7 @@ At the moment the following settings can be configured by means of the *connecto
 | 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
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/source/CosmosDBSourceTask.java b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/CosmosDBSourceTask.java
index 5f1889b..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
@@ -144,7 +144,7 @@ public class CosmosDBSourceTask extends SourceTask {
         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)
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/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());