[FLINK-27399][Connector/Pulsar] Drop Consumer.seek() for apache/pulsar#16171

Author: syhily

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarOrderedPartitionSplitReader.java

@@ -26,7 +26,7 @@
 import org.apache.flink.connector.pulsar.source.split.PulsarPartitionSplit;
 
 import org.apache.pulsar.client.admin.PulsarAdmin;
-import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.PulsarClient;
@@ -75,12 +75,12 @@ protected void finishedPollMessage(Message<byte[]> message) {
     }
 
     @Override
-    protected void startConsumer(PulsarPartitionSplit split, Consumer<byte[]> consumer) {
+    protected void beforeCreatingConsumer(PulsarPartitionSplit split) {
         MessageId latestConsumedId = split.getLatestConsumedId();
 
         // Reset the start position for ordered pulsar consumer.
         if (latestConsumedId != null) {
-            LOG.debug("Start seeking from the checkpoint {}", latestConsumedId);
+            LOG.info("Reset subscription position by the checkpoint {}", latestConsumedId);
             try {
                 MessageId initialPosition;
                 if (latestConsumedId == MessageId.latest
@@ -91,8 +91,15 @@ protected void startConsumer(PulsarPartitionSplit split, Consumer<byte[]> consum
                     initialPosition = nextMessageId(latestConsumedId);
                 }
 
-                consumer.seek(initialPosition);
-            } catch (PulsarClientException e) {
+                // Remove Consumer.seek() here for waiting for pulsar-client-all 2.12.0
+                // See https://github.com/apache/pulsar/issues/16757 for more details.
+                pulsarAdmin
+                        .topics()
+                        .resetCursor(
+                                split.getPartition().getFullTopicName(),
+                                sourceConfiguration.getSubscriptionName(),
+                                initialPosition);
+            } catch (PulsarAdminException e) {
                 if (sourceConfiguration.getVerifyInitialOffsets() == FAIL_ON_MISMATCH) {
                     throw new IllegalArgumentException(e);
                 } else {

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarPartitionSplitReaderBase.java

@@ -53,7 +53,6 @@
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.apache.flink.connector.pulsar.common.utils.PulsarExceptionUtils.sneakyClient;
 import static org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.createConsumerBuilder;
@@ -71,7 +70,6 @@ abstract class PulsarPartitionSplitReaderBase<OUT>
     protected final PulsarAdmin pulsarAdmin;
     protected final SourceConfiguration sourceConfiguration;
     protected final PulsarDeserializationSchema<OUT> deserializationSchema;
-    protected final AtomicBoolean wakeup;
 
     protected Consumer<byte[]> pulsarConsumer;
     protected PulsarPartitionSplit registeredSplit;
@@ -85,7 +83,6 @@ protected PulsarPartitionSplitReaderBase(
         this.pulsarAdmin = pulsarAdmin;
         this.sourceConfiguration = sourceConfiguration;
         this.deserializationSchema = deserializationSchema;
-        this.wakeup = new AtomicBoolean(false);
     }
 
     @Override
@@ -97,19 +94,14 @@ public RecordsWithSplitIds<PulsarMessage<OUT>> fetch() throws IOException {
             return builder.build();
         }
 
-        // Set wakeup to false for start consuming.
-        wakeup.compareAndSet(true, false);
-
         StopCursor stopCursor = registeredSplit.getStopCursor();
         String splitId = registeredSplit.splitId();
         PulsarMessageCollector<OUT> collector = new PulsarMessageCollector<>(splitId, builder);
         Deadline deadline = Deadline.fromNow(sourceConfiguration.getMaxFetchTime());
 
         // Consume message from pulsar until it was woke up by flink reader.
         for (int messageNum = 0;
-                messageNum < sourceConfiguration.getMaxFetchRecords()
-                        && deadline.hasTimeLeft()
-                        && isNotWakeup();
+                messageNum < sourceConfiguration.getMaxFetchRecords() && deadline.hasTimeLeft();
                 messageNum++) {
             try {
                 Duration timeout = deadline.timeLeftIfAny();
@@ -170,23 +162,27 @@ public void handleSplitsChanges(SplitsChange<PulsarPartitionSplit> splitsChanges
                 newSplits.size() == 1, "This pulsar split reader only support one split.");
         PulsarPartitionSplit newSplit = newSplits.get(0);
 
+        // Open stop cursor.
+        newSplit.open(pulsarAdmin);
+
+        // Before creating the consumer.
+        beforeCreatingConsumer(newSplit);
+
         // Create pulsar consumer.
         Consumer<byte[]> consumer = createPulsarConsumer(newSplit);
 
-        // Open start & stop cursor.
-        newSplit.open(pulsarAdmin);
-
-        // Start Consumer.
-        startConsumer(newSplit, consumer);
+        // After creating the consumer.
+        afterCreatingConsumer(newSplit, consumer);
 
         LOG.info("Register split {} consumer for current reader.", newSplit);
+
         this.registeredSplit = newSplit;
         this.pulsarConsumer = consumer;
     }
 
     @Override
     public void wakeUp() {
-        wakeup.compareAndSet(false, true);
+        // Nothing to do on this method.
     }
 
     @Override
@@ -202,14 +198,16 @@ protected abstract Message<byte[]> pollMessage(Duration timeout)
 
     protected abstract void finishedPollMessage(Message<byte[]> message);
 
-    protected abstract void startConsumer(PulsarPartitionSplit split, Consumer<byte[]> consumer);
-
-    // --------------------------- Helper Methods -----------------------------
+    protected void beforeCreatingConsumer(PulsarPartitionSplit split) {
+        // Nothing to do by default.
+    }
 
-    protected boolean isNotWakeup() {
-        return !wakeup.get();
+    protected void afterCreatingConsumer(PulsarPartitionSplit split, Consumer<byte[]> consumer) {
+        // Nothing to do by default.
     }
 
+    // --------------------------- Helper Methods -----------------------------
+
     /** Create a specified {@link Consumer} by the given split information. */
     protected Consumer<byte[]> createPulsarConsumer(PulsarPartitionSplit split) {
         return createPulsarConsumer(split.getPartition());

flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/split/PulsarUnorderedPartitionSplitReader.java

@@ -126,7 +126,7 @@ protected void finishedPollMessage(Message<byte[]> message) {
     }
 
     @Override
-    protected void startConsumer(PulsarPartitionSplit split, Consumer<byte[]> consumer) {
+    protected void afterCreatingConsumer(PulsarPartitionSplit split, Consumer<byte[]> consumer) {
         TxnID uncommittedTransactionId = split.getUncommittedTransactionId();
 
         // Abort the uncommitted pulsar transaction.

flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/PulsarTestContext.java

@@ -24,6 +24,9 @@
 import java.net.URL;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Random;
+
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
 
 /** Common test context for pulsar based test. */
 public abstract class PulsarTestContext<T> implements DataStreamSourceExternalContext<T> {
@@ -39,10 +42,13 @@ protected PulsarTestContext(PulsarTestEnvironment environment, List<URL> connect
     // Helper methods for generating data.
 
     protected List<String> generateStringTestData(int splitIndex, long seed) {
-        int recordNum = 300;
+        Random random = new Random(seed);
+        int recordNum = 300 + random.nextInt(200);
         List<String> records = new ArrayList<>(recordNum);
+
         for (int i = 0; i < recordNum; i++) {
-            records.add(splitIndex + "-" + i);
+            int length = random.nextInt(40) + 10;
+            records.add(splitIndex + "-" + i + "-" + randomAlphanumeric(length));
         }
 
         return records;

flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/testutils/runtime/PulsarRuntimeOperator.java

@@ -31,13 +31,14 @@
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException;
 import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
-import org.apache.pulsar.client.api.SubscriptionInitialPosition;
 import org.apache.pulsar.client.api.TypedMessageBuilder;
 import org.apache.pulsar.client.api.transaction.TransactionCoordinatorClient;
 import org.apache.pulsar.client.api.transaction.TxnID;
@@ -49,10 +50,8 @@
 import java.io.IOException;
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
-import java.util.Map;
 import java.util.Random;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
@@ -62,9 +61,7 @@
 import static java.util.Collections.emptyList;
 import static java.util.Collections.singletonList;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import static java.util.function.Function.identity;
 import static java.util.stream.Collectors.toList;
-import static java.util.stream.Collectors.toMap;
 import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
 import static org.apache.flink.connector.base.DeliveryGuarantee.EXACTLY_ONCE;
 import static org.apache.flink.connector.pulsar.common.config.PulsarOptions.PULSAR_ADMIN_URL;
@@ -79,8 +76,10 @@
 import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicName;
 import static org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.topicNameWithPartition;
 import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.pulsar.client.api.SubscriptionInitialPosition.Earliest;
 import static org.apache.pulsar.client.api.SubscriptionMode.Durable;
 import static org.apache.pulsar.client.api.SubscriptionType.Exclusive;
+import static org.apache.pulsar.common.partition.PartitionedTopicMetadata.NON_PARTITIONED;
 
 /**
  * A pulsar cluster operator used for operating pulsar instance. It's serializable for using in
@@ -178,7 +177,7 @@ public <T> void setupTopic(
      */
     public void createTopic(String topic, int numberOfPartitions) {
         checkArgument(numberOfPartitions >= 0);
-        if (numberOfPartitions <= 0) {
+        if (numberOfPartitions == 0) {
             createNonPartitionedTopic(topic);
         } else {
             createPartitionedTopic(topic, numberOfPartitions);
@@ -196,7 +195,7 @@ public void increaseTopicPartitions(String topic, int newPartitionsNum) {
                 sneakyAdmin(() -> admin().topics().getPartitionedTopicMetadata(topic));
         checkArgument(
                 metadata.partitions < newPartitionsNum,
-                "The new partition size which should exceed previous size.");
+                "The new partition size which should greater than previous size.");
 
         sneakyAdmin(() -> admin().topics().updatePartitionedTopic(topic, newPartitionsNum));
     }
@@ -220,9 +219,11 @@ public void deleteTopic(String topic) {
             return;
         }
 
+        // Close all the available consumers and producers.
         removeConsumers(topic);
         removeProducers(topic);
-        if (metadata.partitions <= 0) {
+
+        if (metadata.partitions == NON_PARTITIONED) {
             sneakyAdmin(() -> admin().topics().delete(topicName));
         } else {
             sneakyAdmin(() -> admin().topics().deletePartitionedTopic(topicName));
@@ -245,22 +246,6 @@ public List<TopicPartition> topicInfo(String topic) {
         }
     }
 
-    /**
-     * Query a list of topics. Convert the topic metadata into a list of topic partitions. Return a
-     * mapping for topic and its partitions.
-     */
-    public Map<String, List<TopicPartition>> topicsInfo(String... topics) {
-        return topicsInfo(Arrays.asList(topics));
-    }
-
-    /**
-     * Query a list of topics. Convert the topic metadata into a list of topic partitions. Return a
-     * mapping for topic and its partitions.
-     */
-    public Map<String, List<TopicPartition>> topicsInfo(Collection<String> topics) {
-        return topics.stream().collect(toMap(identity(), this::topicInfo));
-    }
-
     /**
      * Send a single message to Pulsar, return the message id after the ack from Pulsar.
      *
@@ -518,12 +503,13 @@ private <T> Producer<T> createProducer(String topic, Schema<T> schema)
                 topicProducers.computeIfAbsent(
                         index,
                         i -> {
-                            try {
-                                return client().newProducer(schema).topic(topic).create();
-                            } catch (PulsarClientException e) {
-                                sneakyThrow(e);
-                                return null;
-                            }
+                            ProducerBuilder<T> builder =
+                                    client().newProducer(schema)
+                                            .topic(topic)
+                                            .enableBatching(false)
+                                            .enableMultiSchema(true);
+
+                            return sneakyClient(builder::create);
                         });
     }
 
@@ -540,19 +526,15 @@ private <T> Consumer<T> createConsumer(String topic, Schema<T> schema)
                 topicConsumers.computeIfAbsent(
                         index,
                         i -> {
-                            try {
-                                return client().newConsumer(schema)
-                                        .topic(topic)
-                                        .subscriptionName(SUBSCRIPTION_NAME)
-                                        .subscriptionMode(Durable)
-                                        .subscriptionType(Exclusive)
-                                        .subscriptionInitialPosition(
-                                                SubscriptionInitialPosition.Earliest)
-                                        .subscribe();
-                            } catch (PulsarClientException e) {
-                                sneakyThrow(e);
-                                return null;
-                            }
+                            ConsumerBuilder<T> builder =
+                                    client().newConsumer(schema)
+                                            .topic(topic)
+                                            .subscriptionName(SUBSCRIPTION_NAME)
+                                            .subscriptionMode(Durable)
+                                            .subscriptionType(Exclusive)
+                                            .subscriptionInitialPosition(Earliest);
+
+                            return sneakyClient(builder::subscribe);
                         });
     }
 
@@ -561,11 +543,7 @@ private void removeProducers(String topic) {
         ConcurrentHashMap<Integer, Producer<?>> integerProducers = producers.remove(topicName);
         if (integerProducers != null) {
             for (Producer<?> producer : integerProducers.values()) {
-                try {
-                    producer.close();
-                } catch (PulsarClientException e) {
-                    sneakyThrow(e);
-                }
+                sneakyClient(producer::close);
             }
         }
     }
@@ -575,11 +553,7 @@ private void removeConsumers(String topic) {
         ConcurrentHashMap<Integer, Consumer<?>> integerConsumers = consumers.remove(topicName);
         if (integerConsumers != null) {
             for (Consumer<?> consumer : integerConsumers.values()) {
-                try {
-                    consumer.close();
-                } catch (PulsarClientException e) {
-                    sneakyThrow(e);
-                }
+                sneakyClient(consumer::close);
             }
         }
     }