[flink] branch release-1.14 updated: [FLINK-27917][Connector/Pulsar] Drop Consumer.seek() in the testing method for fixing the race condition. (#20581)

[tison]

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

tison pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.14 by this push:
     new 6ce74dd8e67 [FLINK-27917][Connector/Pulsar] Drop Consumer.seek() in 
the testing method for fixing the race condition. (#20581)
6ce74dd8e67 is described below

commit 6ce74dd8e671688b79770dacf88fd31924cc2e33
Author: Yufan Sheng <yu...@streamnative.io>
AuthorDate: Tue Aug 23 09:55:05 2022 +0800

    [FLINK-27917][Connector/Pulsar] Drop Consumer.seek() in the testing method 
for fixing the race condition. (#20581)
---
 .../split/PulsarPartitionSplitReaderTestBase.java  | 37 +++++++++++++++-------
 1 file changed, 26 insertions(+), 11 deletions(-)

diff --git 
a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderTestBase.java
 
b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderTestBase.java
index e481d1fd793..170e9f3c3dd 100644
--- 
a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderTestBase.java
+++ 
b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderTestBase.java
@@ -32,9 +32,8 @@ import 
org.apache.flink.connector.pulsar.testutils.PulsarTestSuiteBase;
 import 
org.apache.flink.connector.pulsar.testutils.extension.TestOrderlinessExtension;
 import 
org.apache.flink.connectors.test.common.junit.extensions.TestLoggerExtension;
 
-import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.client.impl.MessageIdImpl;
 import org.junit.jupiter.api.TestTemplate;
@@ -61,7 +60,6 @@ import static java.time.Duration.ofSeconds;
 import static java.util.Collections.singletonList;
 import static org.apache.commons.lang3.RandomStringUtils.randomAlphabetic;
 import static 
org.apache.flink.connector.pulsar.common.utils.PulsarExceptionUtils.sneakyAdmin;
-import static 
org.apache.flink.connector.pulsar.common.utils.PulsarExceptionUtils.sneakyThrow;
 import static 
org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_ENABLE_AUTO_ACKNOWLEDGE_MESSAGE;
 import static 
org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_MAX_FETCH_RECORDS;
 import static 
org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_MAX_FETCH_TIME;
@@ -136,14 +134,31 @@ public abstract class PulsarPartitionSplitReaderTestBase 
extends PulsarTestSuite
                 new PulsarPartitionSplit(partition, StopCursor.never(), null, 
null);
         SplitsAddition<PulsarPartitionSplit> addition = new 
SplitsAddition<>(singletonList(split));
 
-        // create consumer and seek before split changes
-        try (Consumer<byte[]> consumer = 
reader.createPulsarConsumer(partition)) {
-            // inclusive messageId
-            consumer.seek(startPosition);
-        } catch (PulsarClientException e) {
-            sneakyThrow(e);
+        // Create the subscription and set the start position for this reader.
+        // Remember not to use Consumer.seek(startPosition)
+        SourceConfiguration sourceConfiguration = reader.sourceConfiguration;
+        PulsarAdmin pulsarAdmin = reader.pulsarAdmin;
+        String subscriptionName = sourceConfiguration.getSubscriptionName();
+        List<String> subscriptions =
+                sneakyAdmin(() -> 
pulsarAdmin.topics().getSubscriptions(topicName));
+        if (!subscriptions.contains(subscriptionName)) {
+            // If this subscription is not available. Just create it.
+            sneakyAdmin(
+                    () ->
+                            pulsarAdmin
+                                    .topics()
+                                    .createSubscription(
+                                            topicName, subscriptionName, 
startPosition));
+        } else {
+            // Reset the subscription if this is existed.
+            sneakyAdmin(
+                    () ->
+                            pulsarAdmin
+                                    .topics()
+                                    .resetCursor(topicName, subscriptionName, 
startPosition));
         }
 
+        // Accept the split and start consuming.
         reader.handleSplitsChanges(addition);
     }
 
@@ -200,7 +215,7 @@ public abstract class PulsarPartitionSplitReaderTestBase 
extends PulsarTestSuite
         String topicName = randomAlphabetic(10);
 
         // Add a split
-        seekStartPositionAndHandleSplit(splitReader, topicName, 0);
+        handleSplit(splitReader, topicName, 0, MessageId.latest);
 
         // Poll once with a null message
         PulsarMessage<String> message1 = fetchedMessage(splitReader);
@@ -224,7 +239,7 @@ public abstract class PulsarPartitionSplitReaderTestBase 
extends PulsarTestSuite
     void consumeMessageCreatedAfterHandleSplitChangesAndFetch(
             PulsarPartitionSplitReaderBase<String> splitReader) {
         String topicName = randomAlphabetic(10);
-        seekStartPositionAndHandleSplit(splitReader, topicName, 0);
+        handleSplit(splitReader, topicName, 0, MessageId.latest);
         operator().sendMessage(topicNameWithPartition(topicName, 0), STRING, 
randomAlphabetic(10));
         fetchedMessages(splitReader, 1, true);
     }