[improve][io] Support update subscription position for sink connector (…

…apache#23538)

(cherry picked from commit 3f12269)
(cherry picked from commit eca6af3)

pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/SinkConfigUtils.java

@@ -709,7 +709,9 @@ public static SinkConfig validateUpdate(SinkConfig existingConfig, SinkConfig ne
         if (newConfig.getTransformFunctionConfig() != null) {
             mergedConfig.setTransformFunctionConfig(newConfig.getTransformFunctionConfig());
         }
-
+        if (newConfig.getSourceSubscriptionPosition() != null) {
+            mergedConfig.setSourceSubscriptionPosition(newConfig.getSourceSubscriptionPosition());
+        }
         return mergedConfig;
     }
 

pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/SinkConfigUtilsTest.java

@@ -38,6 +38,7 @@
 import lombok.Data;
 import lombok.NoArgsConstructor;
 import lombok.experimental.Accessors;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
 import org.apache.pulsar.common.functions.ConsumerConfig;
 import org.apache.pulsar.common.functions.FunctionConfig;
 import org.apache.pulsar.common.functions.Resources;
@@ -223,6 +224,18 @@ public void testCleanSubscriptionField() throws IOException {
         }
     }
 
+    @Test
+    public void testUpdateSubscriptionPosition() {
+        SinkConfig sinkConfig = createSinkConfig();
+        SinkConfig newSinkConfig = createSinkConfig();
+        newSinkConfig.setSourceSubscriptionPosition(SubscriptionInitialPosition.Earliest);
+        SinkConfig mergedConfig = SinkConfigUtils.validateUpdate(sinkConfig, newSinkConfig);
+        assertEquals(
+                new Gson().toJson(newSinkConfig),
+                new Gson().toJson(mergedConfig)
+        );
+    }
+
     @Test
     public void testMergeEqual() {
         SinkConfig sinkConfig = createSinkConfig();
@@ -548,6 +561,7 @@ private SinkConfig createSinkConfig() {
         inputSpecs.put("test-input", ConsumerConfig.builder().isRegexPattern(true).serdeClassName("test-serde").build());
         sinkConfig.setInputSpecs(inputSpecs);
         sinkConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE);
+        sinkConfig.setSourceSubscriptionPosition(SubscriptionInitialPosition.Latest);
         sinkConfig.setRetainOrdering(false);
         sinkConfig.setRetainKeyOrdering(false);
         sinkConfig.setConfigs(new HashMap<>());