feat: support start consume from timestamp (#466)

syhily · zouyunhe · web-flow · commit fbb8d1760dec · 2021-11-16T19:33:40.000+08:00
Co-authored-by: zouyunhe &lt;zouyunhe@bigo.sg&gt;
diff --git a/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSource.java b/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSource.java
@@ -102,6 +102,7 @@
 @Slf4j
 public class FlinkPulsarSource<T> extends RichParallelSourceFunction<T>
         implements ResultTypeQueryable<T>, CheckpointListener, CheckpointedFunction {
+    private static final long serialVersionUID = -6080350107046202906L;
 
     /** The maximum number of pending non-committed checkpoints to track, to avoid memory leaks. */
     public static final int MAX_NUM_PENDING_CHECKPOINTS = 100;
@@ -235,6 +236,8 @@ public class FlinkPulsarSource<T> extends RichParallelSourceFunction<T>
 
     private transient int numParallelTasks;
 
+    private long startupOffsetsTimestamp = -1L;
+
     public FlinkPulsarSource(
             String adminUrl,
             ClientConfigurationData clientConf,
@@ -448,6 +451,24 @@ public FlinkPulsarSource<T> setStartFromSubscription(
         return this;
     }
 
+    public FlinkPulsarSource<T> setStartFromTimestamp(long startupOffsetsTimestamp) {
+        checkArgument(
+                startupOffsetsTimestamp >= 0,
+                "The provided value for the startup offsets timestamp is invalid.");
+
+        long currentTimestamp = System.currentTimeMillis();
+        checkArgument(
+                startupOffsetsTimestamp <= currentTimestamp,
+                "Startup time[%s] must be before current time[%s].",
+                startupOffsetsTimestamp,
+                currentTimestamp);
+
+        this.startupMode = StartupMode.TIMESTAMP;
+        this.startupOffsetsTimestamp = startupOffsetsTimestamp;
+        this.specificStartupOffsets = null;
+        return this;
+    }
+
     // ------------------------------------------------------------------------
     //  Work methods
     // ------------------------------------------------------------------------
@@ -631,9 +652,6 @@ protected PulsarFetcher<T> createFetcher(
             Set<TopicRange> excludeStartMessageIds)
             throws Exception {
 
-        // readerConf.putIfAbsent(PulsarOptions.SUBSCRIPTION_ROLE_OPTION_KEY,
-        // getSubscriptionName());
-
         return new PulsarFetcher<>(
                 sourceContext,
                 seedTopicsWithInitialOffsets,
@@ -650,7 +668,8 @@ protected PulsarFetcher<T> createFetcher(
                 deserializer,
                 metadataReader,
                 streamingRuntime.getMetricGroup().addGroup(PULSAR_SOURCE_METRICS_GROUP),
-                useMetrics);
+                useMetrics,
+                startupOffsetsTimestamp);
     }
 
     public void joinDiscoveryLoopThread() throws InterruptedException {
@@ -988,6 +1007,7 @@ public Map<TopicRange, MessageId> offsetForEachTopic(
 
         switch (mode) {
             case LATEST:
+            case TIMESTAMP:
                 return topics.stream().collect(Collectors.toMap(k -> k, k -> MessageId.latest));
             case EARLIEST:
                 return topics.stream().collect(Collectors.toMap(k -> k, k -> MessageId.earliest));
diff --git a/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/config/StartupMode.java b/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/config/StartupMode.java
@@ -26,5 +26,7 @@ public enum StartupMode {
 
     SPECIFIC_OFFSETS,
 
-    EXTERNAL_SUBSCRIPTION
+    EXTERNAL_SUBSCRIPTION,
+
+    TIMESTAMP
 }
diff --git a/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/internal/PulsarFetcher.java b/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/internal/PulsarFetcher.java
@@ -148,6 +148,8 @@ public class PulsarFetcher<T> {
     /** The metric group which all metrics for the source should be registered to. */
     private final MetricGroup consumerMetricGroup;
 
+    private final long startConsumeTimestamp;
+
     public PulsarFetcher(
             SourceContext<T> sourceContext,
             Map<TopicRange, MessageId> seedTopicsWithInitialOffsets,
@@ -180,7 +182,8 @@ public PulsarFetcher(
                 deserializer,
                 metadataReader,
                 consumerMetricGroup,
-                useMetrics);
+                useMetrics,
+                -1);
     }
 
     public PulsarFetcher(
@@ -199,7 +202,8 @@ public PulsarFetcher(
             PulsarDeserializationSchema<T> deserializer,
             PulsarMetadataReader metadataReader,
             MetricGroup consumerMetricGroup,
-            boolean useMetrics)
+            boolean useMetrics,
+            long startConsumeTimestamp)
             throws Exception {
 
         this.sourceContext = sourceContext;
@@ -272,6 +276,8 @@ public PulsarFetcher(
 
             periodicEmitter.start();
         }
+
+        this.startConsumeTimestamp = startConsumeTimestamp;
     }
 
     public void runFetchLoop() throws Exception {
@@ -567,7 +573,8 @@ protected ReaderThread<T> createReaderThread(
                 exceptionProxy,
                 failOnDataLoss,
                 useEarliestWhenDataLoss,
-                excludeStartMessageIds.contains(state.getTopicRange()));
+                excludeStartMessageIds.contains(state.getTopicRange()),
+                startConsumeTimestamp);
     }
 
     /**
diff --git a/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/internal/ReaderThread.java b/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/internal/ReaderThread.java
@@ -62,14 +62,17 @@ public class ReaderThread<T> extends Thread {
 
     protected volatile Reader<T> reader = null;
 
+    private final long startConsumeTimestamp;
+
     public ReaderThread(
             PulsarFetcher<T> owner,
             PulsarTopicState state,
             ClientConfigurationData clientConf,
             Map<String, Object> readerConf,
             PulsarDeserializationSchema<T> deserializer,
             int pollTimeoutMs,
-            ExceptionProxy exceptionProxy) {
+            ExceptionProxy exceptionProxy,
+            long startConsumeTimestamp) {
         this.owner = owner;
         this.state = state;
         this.clientConf = clientConf;
@@ -80,6 +83,7 @@ public ReaderThread(
 
         this.topicRange = state.getTopicRange();
         this.startMessageId = state.getOffset();
+        this.startConsumeTimestamp = startConsumeTimestamp;
     }
 
     public ReaderThread(
@@ -92,8 +96,17 @@ public ReaderThread(
             ExceptionProxy exceptionProxy,
             boolean failOnDataLoss,
             boolean useEarliestWhenDataLoss,
-            boolean excludeMessageId) {
-        this(owner, state, clientConf, readerConf, deserializer, pollTimeoutMs, exceptionProxy);
+            boolean excludeMessageId,
+            long startConsumeTimestamp) {
+        this(
+                owner,
+                state,
+                clientConf,
+                readerConf,
+                deserializer,
+                pollTimeoutMs,
+                exceptionProxy,
+                startConsumeTimestamp);
         this.failOnDataLoss = failOnDataLoss;
         this.useEarliestWhenDataLoss = useEarliestWhenDataLoss;
         this.excludeMessageId = excludeMessageId;
@@ -111,6 +124,10 @@ public void run() {
             handleTooLargeCursor();
             createActualReader();
             log.info("Starting to read {} with reader thread {}", topicRange, getName());
+            if (startConsumeTimestamp >= 0) {
+                log.info("Reader seek to timestamp:{}", startConsumeTimestamp);
+                reader.seek(startConsumeTimestamp);
+            }
 
             while (running) {
                 Message<T> message = reader.readNext(pollTimeoutMs, TimeUnit.MILLISECONDS);
diff --git a/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/table/PulsarDynamicTableFactory.java b/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/table/PulsarDynamicTableFactory.java
@@ -62,6 +62,7 @@
 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.SCAN_STARTUP_TIMESTAMP_MILLIS;
 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;
@@ -262,6 +263,7 @@ public Set<ConfigOption<?>> optionalOptions() {
         options.add(SCAN_STARTUP_SPECIFIC_OFFSETS);
         options.add(SCAN_STARTUP_SUB_NAME);
         options.add(SCAN_STARTUP_SUB_START_OFFSET);
+        options.add(SCAN_STARTUP_TIMESTAMP_MILLIS);
         options.add(GENERIC);
 
         options.add(PARTITION_DISCOVERY_INTERVAL_MILLIS);
diff --git a/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/table/PulsarDynamicTableSource.java b/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/table/PulsarDynamicTableSource.java
@@ -236,6 +236,9 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext context) {
             case SPECIFIC_OFFSETS:
                 source.setStartFromSpecificOffsets(startupOptions.specificOffsets);
                 break;
+            case TIMESTAMP:
+                source.setStartFromTimestamp(startupOptions.startupTimestampMills);
+                break;
             case EXTERNAL_SUBSCRIPTION:
                 MessageId subscriptionPosition = MessageId.latest;
                 if (CONNECTOR_STARTUP_MODE_VALUE_EARLIEST.equals(
@@ -244,6 +247,7 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext context) {
                 }
                 source.setStartFromSubscription(
                         startupOptions.externalSubscriptionName, subscriptionPosition);
+                break;
         }
         return SourceFunctionProvider.of(source, false);
     }
diff --git a/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/table/PulsarTableOptions.java b/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/table/PulsarTableOptions.java
@@ -255,14 +255,16 @@ public class PulsarTableOptions {
     public static final String SCAN_STARTUP_MODE_VALUE_EXTERNAL_SUBSCRIPTION =
             "external-subscription";
     public static final String SCAN_STARTUP_MODE_VALUE_SPECIFIC_OFFSETS = "specific-offsets";
+    public static final String SCAN_STARTUP_MODE_TIMESTAMP = "timestamp";
 
     private static final Set<String> SCAN_STARTUP_MODE_ENUMS =
             new HashSet<>(
                     Arrays.asList(
                             SCAN_STARTUP_MODE_VALUE_EARLIEST,
                             SCAN_STARTUP_MODE_VALUE_LATEST,
                             SCAN_STARTUP_MODE_VALUE_EXTERNAL_SUBSCRIPTION,
-                            SCAN_STARTUP_MODE_VALUE_SPECIFIC_OFFSETS));
+                            SCAN_STARTUP_MODE_VALUE_SPECIFIC_OFFSETS,
+                            SCAN_STARTUP_MODE_TIMESTAMP));
 
     // Sink partitioner.
     public static final String SINK_MESSAGE_ROUTER_VALUE_KEY_HASH = "key-hash";
@@ -555,6 +557,11 @@ public static StartupOptions getStartupOptions(ReadableConfig tableOptions) {
                     options.startupMode = StartupMode.EXTERNAL_SUBSCRIPTION;
                     break;
 
+                case PulsarValidator.CONNECTOR_STARTUP_MODE_VALUE_TIMESTAMP:
+                    options.startupMode = StartupMode.TIMESTAMP;
+                    options.startupTimestampMills = tableOptions.get(SCAN_STARTUP_TIMESTAMP_MILLIS);
+                    break;
+
                 default:
                     throw new TableException(
                             "Unsupported startup mode. Validator should have checked that.");
@@ -734,6 +741,7 @@ public static class StartupOptions {
         public Map<String, MessageId> specificOffsets = new HashMap<>();
         public String externalSubscriptionName;
         public String externalSubStartOffset;
+        public long startupTimestampMills;
     }
 
     /** Strategies to derive the data type of a value format by considering a key format. */
diff --git a/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/table/descriptors/PulsarValidator.java b/pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/table/descriptors/PulsarValidator.java
@@ -44,6 +44,7 @@ public class PulsarValidator extends ConnectorDescriptorValidator {
     public static final String CONNECTOR_STARTUP_MODE_VALUE_LATEST = "latest";
     public static final String CONNECTOR_STARTUP_MODE_VALUE_SPECIFIC_OFFSETS = "specific-offsets";
     public static final String CONNECTOR_STARTUP_MODE_VALUE_EXTERNAL_SUB = "external-subscription";
+    public static final String CONNECTOR_STARTUP_MODE_VALUE_TIMESTAMP = "timestamp";
     public static final String CONNECTOR_SPECIFIC_OFFSETS = "connector.specific-offsets";
     public static final String CONNECTOR_SPECIFIC_OFFSETS_PARTITION = "partition";
     public static final String CONNECTOR_SPECIFIC_OFFSETS_OFFSET = "offset";

Original file line number	Diff line number	Diff line change
`@@ -26,5 +26,7 @@ public enum StartupMode {`
`26`	`26`
`27`	`27`	`SPECIFIC_OFFSETS,`
`28`	`28`
`29`		`- EXTERNAL_SUBSCRIPTION`
	`29`	`+ EXTERNAL_SUBSCRIPTION,`
	`30`	`+`
	`31`	`+ TIMESTAMP`
`30`	`32`	`}`
Original file line number	Diff line number	Diff line change
`@@ -236,6 +236,9 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext context) {`
`236`	`236`	`case SPECIFIC_OFFSETS:`
`237`	`237`	`source.setStartFromSpecificOffsets(startupOptions.specificOffsets);`
`238`	`238`	`break;`
	`239`	`+ case TIMESTAMP:`
	`240`	`+ source.setStartFromTimestamp(startupOptions.startupTimestampMills);`
	`241`	`+ break;`
`239`	`242`	`case EXTERNAL_SUBSCRIPTION:`
`240`	`243`	`MessageId subscriptionPosition = MessageId.latest;`
`241`	`244`	`if (CONNECTOR_STARTUP_MODE_VALUE_EARLIEST.equals(`
`@@ -244,6 +247,7 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext context) {`
`244`	`247`	`}`
`245`	`248`	`source.setStartFromSubscription(`
`246`	`249`	`startupOptions.externalSubscriptionName, subscriptionPosition);`
	`250`	`+ break;`
`247`	`251`	`}`
`248`	`252`	`return SourceFunctionProvider.of(source, false);`
`249`	`253`	`}`