This is an automated email from the ASF dual-hosted git repository.
martijnvisser 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 10e6b3ef591 [FLINK-26645][Connector/pulsar] Support subscribe only one
topic partition.
10e6b3ef591 is described below
commit 10e6b3ef591a0c42155f9fd799bf5d53a5f7df4d
Author: Yufan Sheng <[email protected]>
AuthorDate: Wed Mar 30 20:41:55 2022 +0800
[FLINK-26645][Connector/pulsar] Support subscribe only one topic partition.
---
.../pulsar/source/PulsarSourceBuilder.java | 3 +-
.../subscriber/impl/BasePulsarSubscriber.java | 23 +++--
.../subscriber/impl/TopicListSubscriber.java | 56 +++++++++---
.../source/enumerator/topic/TopicNameUtils.java | 50 +++++++++++
.../subscriber/PulsarSubscriberTest.java | 99 +++++++++++++++++-----
5 files changed, 181 insertions(+), 50 deletions(-)
diff --git
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
index 6e9cfc0b7cd..19c93f4014d 100644
---
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
+++
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/PulsarSourceBuilder.java
@@ -57,6 +57,7 @@ import static
org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSA
import static
org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_SUBSCRIPTION_TYPE;
import static
org.apache.flink.connector.pulsar.source.PulsarSourceOptions.PULSAR_TRANSACTION_TIMEOUT_MILLIS;
import static
org.apache.flink.connector.pulsar.source.config.PulsarSourceConfigUtils.checkConfigurations;
+import static
org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.distinctTopics;
import static org.apache.flink.util.InstantiationUtil.isSerializable;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -209,7 +210,7 @@ public final class PulsarSourceBuilder<OUT> {
*/
public PulsarSourceBuilder<OUT> setTopics(List<String> topics) {
ensureSubscriberIsNull("topics");
- this.subscriber = PulsarSubscriber.getTopicListSubscriber(topics);
+ this.subscriber =
PulsarSubscriber.getTopicListSubscriber(distinctTopics(topics));
return this;
}
diff --git
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/BasePulsarSubscriber.java
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/BasePulsarSubscriber.java
index d266c98df42..d29f7a0d548 100644
---
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/BasePulsarSubscriber.java
+++
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/BasePulsarSubscriber.java
@@ -28,8 +28,8 @@ import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
+import java.util.ArrayList;
import java.util.List;
-import java.util.stream.IntStream;
import static java.util.stream.Collectors.toList;
@@ -63,18 +63,15 @@ public abstract class BasePulsarSubscriber implements
PulsarSubscriber {
.map(range -> new TopicPartition(metadata.getName(), -1,
range))
.collect(toList());
} else {
- return IntStream.range(0, metadata.getPartitionSize())
- .boxed()
- .flatMap(
- partitionId ->
- ranges.stream()
- .map(
- range ->
- new TopicPartition(
-
metadata.getName(),
-
partitionId,
- range)))
- .collect(toList());
+ List<TopicPartition> partitions = new ArrayList<>();
+ for (int i = 0; i < metadata.getPartitionSize(); i++) {
+ for (TopicRange range : ranges) {
+ TopicPartition partition = new
TopicPartition(metadata.getName(), i, range);
+ partitions.add(partition);
+ }
+ }
+
+ return partitions;
}
}
}
diff --git
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/TopicListSubscriber.java
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/TopicListSubscriber.java
index 5f5617ebe50..28f525b2d07 100644
---
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/TopicListSubscriber.java
+++
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/impl/TopicListSubscriber.java
@@ -18,40 +18,68 @@
package org.apache.flink.connector.pulsar.source.enumerator.subscriber.impl;
+import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicMetadata;
import
org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
import org.apache.flink.connector.pulsar.source.enumerator.topic.TopicRange;
import
org.apache.flink.connector.pulsar.source.enumerator.topic.range.RangeGenerator;
import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.common.naming.TopicName;
+import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
-import java.util.Objects;
import java.util.Set;
-import static java.util.stream.Collectors.toSet;
+import static
org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils.isPartition;
/** the implements of consuming multiple topics. */
public class TopicListSubscriber extends BasePulsarSubscriber {
private static final long serialVersionUID = 6473918213832993116L;
- private final List<String> topics;
+ private final List<String> partitions;
+ private final List<String> fullTopicNames;
- public TopicListSubscriber(List<String> topics) {
- this.topics = topics;
+ public TopicListSubscriber(List<String> fullTopicNameOrPartitions) {
+ this.partitions = new ArrayList<>();
+ this.fullTopicNames = new ArrayList<>();
+
+ for (String fullTopicNameOrPartition : fullTopicNameOrPartitions) {
+ if (isPartition(fullTopicNameOrPartition)) {
+ this.partitions.add(fullTopicNameOrPartition);
+ } else {
+ this.fullTopicNames.add(fullTopicNameOrPartition);
+ }
+ }
}
@Override
public Set<TopicPartition> getSubscribedTopicPartitions(
PulsarAdmin pulsarAdmin, RangeGenerator rangeGenerator, int
parallelism) {
+ Set<TopicPartition> results = new HashSet<>();
+
+ // Query topics from Pulsar.
+ for (String topic : fullTopicNames) {
+ TopicMetadata metadata = queryTopicMetadata(pulsarAdmin, topic);
+ List<TopicRange> ranges = rangeGenerator.range(metadata,
parallelism);
+ List<TopicPartition> list = toTopicPartitions(metadata, ranges);
+
+ results.addAll(list);
+ }
+
+ for (String partition : partitions) {
+ TopicName topicName = TopicName.get(partition);
+ String name = topicName.getPartitionedTopicName();
+ int index = topicName.getPartitionIndex();
+
+ TopicMetadata metadata = queryTopicMetadata(pulsarAdmin, name);
+ List<TopicRange> ranges = rangeGenerator.range(metadata,
parallelism);
+
+ for (TopicRange range : ranges) {
+ results.add(new TopicPartition(name, index, range));
+ }
+ }
- return topics.parallelStream()
- .map(topic -> queryTopicMetadata(pulsarAdmin, topic))
- .filter(Objects::nonNull)
- .flatMap(
- metadata -> {
- List<TopicRange> ranges =
rangeGenerator.range(metadata, parallelism);
- return toTopicPartitions(metadata,
ranges).stream();
- })
- .collect(toSet());
+ return results;
}
}
diff --git
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicNameUtils.java
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicNameUtils.java
index 446622c8ea2..41c54892838 100644
---
a/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicNameUtils.java
+++
b/flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/enumerator/topic/TopicNameUtils.java
@@ -20,8 +20,17 @@ package
org.apache.flink.connector.pulsar.source.enumerator.topic;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableList;
+
import org.apache.pulsar.common.naming.TopicName;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
import static org.apache.flink.util.Preconditions.checkArgument;
/** util for topic name. */
@@ -42,4 +51,45 @@ public final class TopicNameUtils {
checkArgument(partitionId >= 0, "Illegal partition id %s",
partitionId);
return TopicName.get(topic).getPartition(partitionId).toString();
}
+
+ /** Get a non-partitioned topic name that does not belong to any
partitioned topic. */
+ public static String topicNameWithoutPartition(String topic) {
+ return TopicName.get(topic).toString();
+ }
+
+ public static boolean isPartition(String topic) {
+ return TopicName.get(topic).isPartitioned();
+ }
+
+ /** Merge the same topics into one topic. */
+ public static List<String> distinctTopics(List<String> topics) {
+ Set<String> fullTopics = new HashSet<>();
+ Map<String, List<Integer>> partitionedTopics = new HashMap<>();
+
+ for (String topic : topics) {
+ TopicName topicName = TopicName.get(topic);
+ String partitionedTopicName = topicName.getPartitionedTopicName();
+
+ if (!topicName.isPartitioned()) {
+ fullTopics.add(partitionedTopicName);
+ partitionedTopics.remove(partitionedTopicName);
+ } else if (!fullTopics.contains(partitionedTopicName)) {
+ List<Integer> partitionIds =
+ partitionedTopics.computeIfAbsent(
+ partitionedTopicName, k -> new ArrayList<>());
+ partitionIds.add(topicName.getPartitionIndex());
+ }
+ }
+
+ ImmutableList.Builder<String> builder =
ImmutableList.<String>builder().addAll(fullTopics);
+
+ for (Map.Entry<String, List<Integer>> topicSet :
partitionedTopics.entrySet()) {
+ String topicName = topicSet.getKey();
+ for (Integer partitionId : topicSet.getValue()) {
+ builder.add(topicNameWithPartition(topicName, partitionId));
+ }
+ }
+
+ return builder.build();
+ }
}
diff --git
a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java
b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java
index bdfbb42ea31..eaae6ba6b33 100644
---
a/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java
+++
b/flink-connectors/flink-connector-pulsar/src/test/java/org/apache/flink/connector/pulsar/source/enumerator/subscriber/PulsarSubscriberTest.java
@@ -18,11 +18,12 @@
package org.apache.flink.connector.pulsar.source.enumerator.subscriber;
-import
org.apache.flink.connector.pulsar.source.enumerator.topic.TopicNameUtils;
import
org.apache.flink.connector.pulsar.source.enumerator.topic.TopicPartition;
import
org.apache.flink.connector.pulsar.source.enumerator.topic.range.FullRangeGenerator;
import org.apache.flink.connector.pulsar.testutils.PulsarTestSuiteBase;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
@@ -30,50 +31,108 @@ import java.util.HashSet;
import java.util.Set;
import java.util.regex.Pattern;
+import static java.util.Collections.singletonList;
+import static org.apache.commons.lang3.RandomStringUtils.randomAlphanumeric;
import static
org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber.getTopicListSubscriber;
import static
org.apache.flink.connector.pulsar.source.enumerator.subscriber.PulsarSubscriber.getTopicPatternSubscriber;
+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.connector.pulsar.source.enumerator.topic.TopicRange.createFullRange;
import static org.apache.pulsar.client.api.RegexSubscriptionMode.AllTopics;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
/** Unit tests for {@link PulsarSubscriber}. */
class PulsarSubscriberTest extends PulsarTestSuiteBase {
- private static final String TOPIC1 = TopicNameUtils.topicName("topic1");
- private static final String TOPIC2 =
TopicNameUtils.topicName("pattern-topic");
- private static final String TOPIC3 = TopicNameUtils.topicName("topic2");
+ private final String topic1 = topicName("topic-" + randomAlphanumeric(4));
+ private final String topic2 = topicName("pattern-topic-" +
randomAlphanumeric(4));
+ private final String topic3 = topicName("topic2-" + randomAlphanumeric(4));
+ private final String topic4 = topicName("non-partitioned-topic-" +
randomAlphanumeric(4));
+ private final String topic5 = topicName("non-partitioned-topic2-" +
randomAlphanumeric(4));
private static final int NUM_PARTITIONS_PER_TOPIC = 5;
private static final int NUM_PARALLELISM = 10;
+ @BeforeAll
+ void setUp() {
+ operator().createTopic(topic1, NUM_PARTITIONS_PER_TOPIC);
+ operator().createTopic(topic2, NUM_PARTITIONS_PER_TOPIC);
+ operator().createTopic(topic3, NUM_PARTITIONS_PER_TOPIC);
+ operator().createTopic(topic4, 0);
+ operator().createTopic(topic5, 0);
+ }
+
+ @AfterAll
+ void tearDown() {
+ operator().deleteTopic(topic1, true);
+ operator().deleteTopic(topic2, true);
+ operator().deleteTopic(topic3, true);
+ operator().deleteTopic(topic4, false);
+ operator().deleteTopic(topic5, false);
+ }
+
@Test
void topicListSubscriber() {
- operator().createTopic(TOPIC1, NUM_PARTITIONS_PER_TOPIC);
- operator().createTopic(TOPIC2, NUM_PARTITIONS_PER_TOPIC);
-
- PulsarSubscriber subscriber =
getTopicListSubscriber(Arrays.asList(TOPIC1, TOPIC2));
+ PulsarSubscriber subscriber =
getTopicListSubscriber(Arrays.asList(topic1, topic2));
Set<TopicPartition> topicPartitions =
subscriber.getSubscribedTopicPartitions(
operator().admin(), new FullRangeGenerator(),
NUM_PARALLELISM);
Set<TopicPartition> expectedPartitions = new HashSet<>();
for (int i = 0; i < NUM_PARTITIONS_PER_TOPIC; i++) {
- expectedPartitions.add(new TopicPartition(TOPIC1, i,
createFullRange()));
- expectedPartitions.add(new TopicPartition(TOPIC2, i,
createFullRange()));
+ expectedPartitions.add(new TopicPartition(topic1, i,
createFullRange()));
+ expectedPartitions.add(new TopicPartition(topic2, i,
createFullRange()));
}
assertEquals(expectedPartitions, topicPartitions);
+ }
+
+ @Test
+ void subscribeOnePartitionOfMultiplePartitionTopic() {
+ String partition = topicNameWithPartition(topic1, 2);
- operator().deleteTopic(TOPIC1, true);
- operator().deleteTopic(TOPIC2, true);
+ PulsarSubscriber subscriber =
getTopicListSubscriber(singletonList(partition));
+ Set<TopicPartition> partitions =
+ subscriber.getSubscribedTopicPartitions(
+ operator().admin(), new FullRangeGenerator(),
NUM_PARALLELISM);
+
+ TopicPartition desiredPartition = new TopicPartition(topic1, 2,
createFullRange());
+ assertThat(partitions).hasSize(1).containsExactly(desiredPartition);
}
@Test
- void topicPatternSubscriber() {
- operator().createTopic(TOPIC1, NUM_PARTITIONS_PER_TOPIC);
- operator().createTopic(TOPIC2, NUM_PARTITIONS_PER_TOPIC);
- operator().createTopic(TOPIC3, NUM_PARTITIONS_PER_TOPIC);
+ void subscribeNonPartitionedTopicList() {
+ PulsarSubscriber subscriber =
getTopicListSubscriber(singletonList(topic4));
+ Set<TopicPartition> partitions =
+ subscriber.getSubscribedTopicPartitions(
+ operator().admin(), new FullRangeGenerator(),
NUM_PARALLELISM);
+
+ TopicPartition desiredPartition = new TopicPartition(topic4, -1,
createFullRange());
+ assertThat(partitions).hasSize(1).containsExactly(desiredPartition);
+ }
+ @Test
+ void subscribeNonPartitionedTopicPattern() {
+ PulsarSubscriber subscriber =
+ getTopicPatternSubscriber(
+
Pattern.compile("persistent://public/default/non-partitioned-topic*?"),
+ AllTopics);
+
+ Set<TopicPartition> topicPartitions =
+ subscriber.getSubscribedTopicPartitions(
+ operator().admin(), new FullRangeGenerator(),
NUM_PARALLELISM);
+
+ Set<TopicPartition> expectedPartitions = new HashSet<>();
+
+ expectedPartitions.add(new TopicPartition(topic4, -1,
createFullRange()));
+ expectedPartitions.add(new TopicPartition(topic5, -1,
createFullRange()));
+
+ assertEquals(expectedPartitions, topicPartitions);
+ }
+
+ @Test
+ void topicPatternSubscriber() {
PulsarSubscriber subscriber =
getTopicPatternSubscriber(
Pattern.compile("persistent://public/default/topic*?"), AllTopics);
@@ -85,14 +144,10 @@ class PulsarSubscriberTest extends PulsarTestSuiteBase {
Set<TopicPartition> expectedPartitions = new HashSet<>();
for (int i = 0; i < NUM_PARTITIONS_PER_TOPIC; i++) {
- expectedPartitions.add(new TopicPartition(TOPIC1, i,
createFullRange()));
- expectedPartitions.add(new TopicPartition(TOPIC3, i,
createFullRange()));
+ expectedPartitions.add(new TopicPartition(topic1, i,
createFullRange()));
+ expectedPartitions.add(new TopicPartition(topic3, i,
createFullRange()));
}
assertEquals(expectedPartitions, topicPartitions);
-
- operator().deleteTopic(TOPIC1, true);
- operator().deleteTopic(TOPIC2, true);
- operator().deleteTopic(TOPIC3, true);
}
}
