fix config name to scan.startup.sub-start-offset

Author: 高章敏

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

@@ -189,7 +189,7 @@ public DynamicTableSource createDynamicTableSource(Context context) {
         Properties properties = getPulsarProperties(context.getCatalogTable().toProperties());
 
         final PulsarTableOptions.StartupOptions startupOptions = PulsarTableOptions
-                .getStartupOptions(tableOptions, topics);
+                .getStartupOptions(tableOptions);
 
         final DataType physicalDataType = context.getCatalogTable().getSchema().toPhysicalRowDataType();
 

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

@@ -173,10 +173,10 @@ private PulsarTableOptions() {
             .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")
+            .key("scan.startup.sub-start-offset")
             .stringType()
             .defaultValue("latest")
-            .withDescription("Optional sub-startOffset used in case of \"external-subscription\" startup mode");
+            .withDescription("Optional sub-start-offset used in case of \"external-subscription\" startup mode");
 
     public static final ConfigOption<Long> SCAN_STARTUP_TIMESTAMP_MILLIS = ConfigOptions
             .key("scan.startup.timestamp-millis")
@@ -446,58 +446,46 @@ private static boolean hasPulsarClientProperties(Map<String, String> tableOption
         return tableOptions.keySet().stream().anyMatch(k -> k.startsWith(PROPERTIES_PREFIX));
     }
 
-    public static StartupOptions getStartupOptions(
-            ReadableConfig tableOptions,
-            List<String> topics) {
-
-        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) {
-                        case PulsarValidator.CONNECTOR_STARTUP_MODE_VALUE_EARLIEST:
-                            return StartupMode.EARLIEST;
-
-                        case PulsarValidator.CONNECTOR_STARTUP_MODE_VALUE_LATEST:
-                            return StartupMode.LATEST;
-
-                        case PulsarValidator.CONNECTOR_STARTUP_MODE_VALUE_SPECIFIC_OFFSETS:
-                            String specificOffsetsStrOpt = tableOptions.get(SCAN_STARTUP_SPECIFIC_OFFSETS);
-
-                            final Map<Integer, String> offsetList = parseSpecificOffsets(
-                                    specificOffsetsStrOpt,
-                                    SCAN_STARTUP_SPECIFIC_OFFSETS.key());
-                            offsetList.forEach((partition, offset) -> {
-                                try {
-
-                                    final MessageIdImpl messageId = parseMessageId(offset);
-                                    specificOffsets.put(partition.toString(), messageId);
-                                } catch (Exception e) {
-                                    log.error("Failed to decode message id from properties {}",
-                                            ExceptionUtils.stringifyException(e));
-                                    throw new RuntimeException(e);
-                                }
-                            });
-                            return StartupMode.SPECIFIC_OFFSETS;
-
-                        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:
-                            throw new TableException("Unsupported startup mode. Validator should have checked that.");
-                    }
-                }).orElse(StartupMode.LATEST);
+    public static StartupOptions getStartupOptions(ReadableConfig tableOptions) {
         final StartupOptions options = new StartupOptions();
-        options.startupMode = startupMode;
-        options.specificOffsets = specificOffsets;
-        if (subName.size() != 0) {
-            options.externalSubscriptionName = subName.get(0);
-        }
-        if (subStartOffset.size() != 0) {
-            options.externalSubStartOffset = subStartOffset.get(0);
+        final Map<String, MessageId> specificOffsets = new HashMap<>();
+        Optional<String> modeString = tableOptions.getOptional(SCAN_STARTUP_MODE);
+        if(!modeString.isPresent()) {
+            options.startupMode = StartupMode.LATEST;
+        } else {
+            switch (modeString.get()) {
+                case PulsarValidator.CONNECTOR_STARTUP_MODE_VALUE_EARLIEST:
+                    options.startupMode = StartupMode.EARLIEST;
+
+                case PulsarValidator.CONNECTOR_STARTUP_MODE_VALUE_LATEST:
+                    options.startupMode = StartupMode.LATEST;
+
+                case PulsarValidator.CONNECTOR_STARTUP_MODE_VALUE_SPECIFIC_OFFSETS:
+                    String specificOffsetsStrOpt = tableOptions.get(SCAN_STARTUP_SPECIFIC_OFFSETS);
+
+                    final Map<Integer, String> offsetList = parseSpecificOffsets(
+                            specificOffsetsStrOpt,
+                            SCAN_STARTUP_SPECIFIC_OFFSETS.key());
+                    offsetList.forEach((partition, offset) -> {
+                        try {
+                            final MessageIdImpl messageId = parseMessageId(offset);
+                            specificOffsets.put(partition.toString(), messageId);
+                        } catch (Exception e) {
+                            log.error("Failed to decode message id from properties {}",
+                                    ExceptionUtils.stringifyException(e));
+                            throw new RuntimeException(e);
+                        }
+                    });
+                    options.startupMode = StartupMode.SPECIFIC_OFFSETS;
+                    options.specificOffsets = specificOffsets;
+
+                case PulsarValidator.CONNECTOR_STARTUP_MODE_VALUE_EXTERNAL_SUB:
+                    options.externalSubscriptionName = tableOptions.get(SCAN_STARTUP_SUB_NAME);
+                    options.externalSubStartOffset = tableOptions.get(SCAN_STARTUP_SUB_START_OFFSET);
+                    options.startupMode = StartupMode.EXTERNAL_SUBSCRIPTION;
+                default:
+                    throw new TableException("Unsupported startup mode. Validator should have checked that.");
+            }
         }
         return options;