[flink-connector-pulsar] 01/16: [FLINK-28853][pulsar] Implement pausing Pulsar splits in case of watermark drift

[ma...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git

commit d21147e8bc3dab7d42ecd7b1fa48d2736ff7b918
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Fri Mar 25 09:33:33 2022 +0100

    [FLINK-28853][pulsar] Implement pausing Pulsar splits in case of watermark drift
---
 .../source/reader/source/PulsarSourceReaderBase.java     |  8 ++++++++
 .../reader/split/PulsarPartitionSplitReaderBase.java     | 16 ++++++++++++++++
 2 files changed, 24 insertions(+)

diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java
index 0122021..c910e94 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/source/PulsarSourceReaderBase.java
@@ -32,6 +32,8 @@ import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplitState;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.api.PulsarClient;
 
+import java.util.Collection;
+
 /**
  * The common pulsar source reader for both ordered & unordered message consuming.
  *
@@ -75,6 +77,12 @@ abstract class PulsarSourceReaderBase<OUT>
         return splitState.toPulsarPartitionSplit();
     }
 
+    @Override
+    public void pauseOrResumeSplits(
+            Collection<String> splitsToPause, Collection<String> splitsToResume) {
+        splitFetcherManager.pauseOrResumeSplits(splitsToPause, splitsToResume);
+    }
+
     @Override
     public void close() throws Exception {
         // Close the all the consumers first.
diff --git a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java
index 9537969..c1459a6 100644
--- a/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java
+++ b/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java
@@ -50,6 +50,7 @@ import javax.annotation.Nullable;
 
 import java.io.IOException;
 import java.time.Duration;
+import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeoutException;
@@ -179,6 +180,21 @@ abstract class PulsarPartitionSplitReaderBase<OUT>
         LOG.info("Register split {} consumer for current reader.", registeredSplit);
     }
 
+    @Override
+    public void pauseOrResumeSplits(
+            Collection<PulsarPartitionSplit> splitsToPause,
+            Collection<PulsarPartitionSplit> splitsToResume) {
+        if (splitsToPause.size() > 1 || splitsToResume.size() > 1) {
+            throw new IllegalStateException("This pulsar split reader only support one split.");
+        }
+
+        if (!splitsToPause.isEmpty()) {
+            pulsarConsumer.pause();
+        } else if (!splitsToResume.isEmpty()) {
+            pulsarConsumer.resume();
+        }
+    }
+
     @Override
     public void wakeUp() {
         // Nothing to do on this method.