external-sub support consuming from earliest offset

Author: 高章敏

pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/table/PulsarDynamicTableFactory.java

@@ -57,6 +57,7 @@
 import static org.apache.flink.streaming.connectors.pulsar.table.PulsarTableOptions.SCAN_STARTUP_MODE;
 import static org.apache.flink.streaming.connectors.pulsar.table.PulsarTableOptions.SCAN_STARTUP_SPECIFIC_OFFSETS;
 import static org.apache.flink.streaming.connectors.pulsar.table.PulsarTableOptions.SCAN_STARTUP_SUB_NAME;
+import static org.apache.flink.streaming.connectors.pulsar.table.PulsarTableOptions.SCAN_STARTUP_SUB_START_OFFSET;
 import static org.apache.flink.streaming.connectors.pulsar.table.PulsarTableOptions.SERVICE_URL;
 import static org.apache.flink.streaming.connectors.pulsar.table.PulsarTableOptions.SINK_MESSAGE_ROUTER;
 import static org.apache.flink.streaming.connectors.pulsar.table.PulsarTableOptions.SINK_SEMANTIC;
@@ -240,6 +241,7 @@ public Set<ConfigOption<?>> optionalOptions() {
         options.add(SCAN_STARTUP_MODE);
         options.add(SCAN_STARTUP_SPECIFIC_OFFSETS);
         options.add(SCAN_STARTUP_SUB_NAME);
+        options.add(SCAN_STARTUP_SUB_START_OFFSET);
 
         options.add(PARTITION_DISCOVERY_INTERVAL_MILLIS);
         options.add(SINK_SEMANTIC);

pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/table/PulsarDynamicTableSource.java

@@ -249,8 +249,7 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext context) {
                 break;
             case EXTERNAL_SUBSCRIPTION:
                 MessageId subscriptionPosition = MessageId.latest;
-                if (CONNECTOR_STARTUP_MODE_VALUE_EARLIEST
-                        .equals(properties.get(CONNECTOR_EXTERNAL_SUB_DEFAULT_OFFSET))) {
+                if (CONNECTOR_STARTUP_MODE_VALUE_EARLIEST.equals(startupOptions.externalSubStartOffset)) {
                     subscriptionPosition = MessageId.earliest;
                 }
                 source.setStartFromSubscription(startupOptions.externalSubscriptionName, subscriptionPosition);

pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/table/PulsarTableOptions.java

@@ -170,7 +170,13 @@ private PulsarTableOptions() {
             .key("scan.startup.sub-name")
             .stringType()
             .noDefaultValue()
-            .withDescription("Optional sub-name used in case of \"specific-offsets\" startup mode");
+            .withDescription("Optional sub-name used in case of \"external-subscription\" startup mode");
+
+    public static final ConfigOption<String> SCAN_STARTUP_SUB_START_OFFSET = ConfigOptions
+            .key("scan.startup.sub-startOffset")
+            .stringType()
+            .defaultValue("latest")
+            .withDescription("Optional sub-startOffset used in case of \"external-subscription\" startup mode");
 
     public static final ConfigOption<Long> SCAN_STARTUP_TIMESTAMP_MILLIS = ConfigOptions
             .key("scan.startup.timestamp-millis")
@@ -446,6 +452,7 @@ public static StartupOptions getStartupOptions(
 
         final Map<String, MessageId> specificOffsets = new HashMap<>();
         final List<String> subName = new ArrayList<>(1);
+        final List<String> subStartOffset = new ArrayList<>(1);
         final StartupMode startupMode = tableOptions.getOptional(SCAN_STARTUP_MODE)
                 .map(modeString -> {
                     switch (modeString) {
@@ -476,6 +483,7 @@ public static StartupOptions getStartupOptions(
 
                         case PulsarValidator.CONNECTOR_STARTUP_MODE_VALUE_EXTERNAL_SUB:
                             subName.add(tableOptions.get(SCAN_STARTUP_SUB_NAME));
+                            subStartOffset.add(tableOptions.get(SCAN_STARTUP_SUB_START_OFFSET));
                             return StartupMode.EXTERNAL_SUBSCRIPTION;
 
                         default:
@@ -488,6 +496,9 @@ public static StartupOptions getStartupOptions(
         if (subName.size() != 0) {
             options.externalSubscriptionName = subName.get(0);
         }
+        if (subStartOffset.size() != 0) {
+            options.externalSubStartOffset = subStartOffset.get(0);
+        }
         return options;
 
     }
@@ -680,6 +691,7 @@ public static class StartupOptions {
         public StartupMode startupMode;
         public Map<String, MessageId> specificOffsets;
         public String externalSubscriptionName;
+        public String externalSubStartOffset;
     }
 
     /**