PIP-13-1/3: Provide `TopicsConsumer` to consume from several topics under same namespace #1103

zhaijack · 2018-01-24T15:47:19Z

Motivation

This is a first sub-task for pip-13, which would like handle subscription to topics under same namespace.

Modifications

add subscribe methods in PulsarClient and PulsarClientImpl.
add TopicsConsumerImpl, TopicMessageImpl, TopicMessageIdImpl, add some test.
change parameter in UnAckedMessageTracker from MessageIdImpl to MessageId.

Result

old methods behaviour not changed,
user could use new method to subscribe to topics

zhaijack · 2018-01-26T04:17:55Z

retest this please

ivankelly · 2018-02-06T15:32:46Z

pulsar-client/src/main/java/org/apache/pulsar/client/api/MessageId.java

+     *
+     * @return the topic name
+     */
+    default String getTopicName() {


why is this in both message and messageid?

I understand that the implemenation needs to know which topic originated the message so that it can "acknowledge" on the right internal consumer, but I think this doesn't need to be exposed in the MessageId interface.

Yes. It is as @merlimat metioned. for MessageId, we also need topicName for redeliverUnacknowledgedMessages.
will remove it from MessageId

ivankelly · 2018-02-06T15:45:59Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedConsumerImpl.java

@@ -468,17 +468,20 @@ public void redeliverUnacknowledgedMessages() {
    }

    @Override
-    public void redeliverUnacknowledgedMessages(Set<MessageIdImpl> messageIds) {
+    public void redeliverUnacknowledgedMessages(Set<MessageId> messageIds) {


Why are you changing this to MessageId? You're casting to MessageIdImpl anyhow.

Thanks, As above reply, we also need this redeliverUnacknowledgedMessages method in Consumer.java handling TopicMessageIdImpl. would like to change this in Consumer.java and make the case in each child class.

In Consumer.java it doesn't know about the MessageIds. The call takes no parameters.

TopicsMessageIdImpl should be a specialization of MessageIdImpl (I assumed it was when I originally commented, until i saw Matteo's comment to the same effect). If it is a specialization of TopicsMessageIdImpl, then this signature doesn't need to change.

Oh, it is in ConsumerBase.java, sorry for the wrong reference of Consumer.java.
TopicsMessageIdImpl is more need to be a wrapper for MessageIdImpl, and as the reply below, may be better to implements MessageId, instead of extends MessageIdImpl.

ivankelly · 2018-02-06T15:50:39Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java

 import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

-import io.netty.util.Timeout;
-import io.netty.util.TimerTask;
-
 public class UnAckedMessageTracker implements Closeable {


UnackedMessageTracker is a member of ConsumerImpl. Since there is a new ConsumerImpl in this patch, instead of changing the sets to MessageId, you should create a generics parameter on this class, and let the creator of the class decide which message id type to track. This will also get rid of the instanceof stuff further down.

Thanks. UnackedMessageTracker is not only a member of ConsumerImpl, it should also a member of PartitionedConsumer and TopicsConsumer here. Most of the change is to leverage MessageId::compareTo to make the code clearer. and seems it not casts the type too often here.

For both PartitionedConsumer and TopicsConsumer you know what kind of MessageId you will be need.
For PartitionedConsumer you can create a UnAckedMessageTracker<MessageIdImpl>.
For ConsumerConsumer you can create an UnAckedMessageTracker<MessageIdImpl>.
For TopicsConsumer you can create an UnAckedMessageTracker<TopicMessageIdImpl>.

Any cast, at all, is code smell. Looking at this again, you should also have a specialization of UnAckedMessageTracker, and only this class should implement #removeTopicMessages.

Thanks, will make a class UnAckedTopicMessageTracker extends UnAckedMessageTracker

Github mauled my previous comment. Added some formatting to clarify.

ivankelly · 2018-02-06T16:01:08Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicsConsumerImpl.java

+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TopicsConsumerImpl extends ConsumerBase {


What bits of ConsumerBase are actually useful here? It looks like you've had to reimplement a large amount of it (consumerbase is ~350LOC, this is ~900LOC), while having to stub out all the HandlerBase stuff which doesn't apply in the case of this class.

The majority of ConsumerBase is just default implementations that call other implementation. Perhaps these could be moved up a layer in the class hierarchy so you could avoid pulling in the HandlerBase stuff completely.

HandlerBase seems only useful for ConsumerImpl and ProducerImpl, There was an issue opened, which plan to remove it from ConsumerBase.
Since ConsumerBase has did some of the work, would like to leverage it and not to do replicated work.

ivankelly · 2018-02-06T16:04:29Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicsConsumerImpl.java

+        if (topics.isEmpty()) {
+            this.namespaceName = null;
+            setState(State.Ready);
+            // We have successfully created N consumers, so we can start receiving messages now


N consumer? no consumers?

Thanks. This is an obsolete comments. will remove it. and Yes. here is no topics created.

ivankelly · 2018-02-06T16:22:46Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicsConsumerImpl.java

+    // Check topics are valid.
+    // - each topic is valid,
+    // - every topic has same namespace.
+    private static boolean topicsNameInvalid(Collection<String> topics) {


nit: topicNamesInvalid makes more sense. topicNamesValid would be even better IMO, but you'd have to invert the booleans.

Thanks, will change the name

ivankelly · 2018-02-06T16:41:56Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicsConsumerImpl.java

+    }
+
+    @Override
+    protected Message internalReceive() throws PulsarClientException {


why doesn't this and the next message just call internalReceiveAsync()? It would save a lot of duplicate code.

Thanks. Seems there is some different between sync and async call. and also ComsumerImpl and PartitionedConsumerImpl also doing like this.

ivankelly · 2018-02-06T16:45:07Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicsConsumerImpl.java

+
+        final String namespace = DestinationName.get(topics.stream().findFirst().get()).getNamespace();
+
+        Optional<String> result = topics.stream()


should probably check that topic names are unique also.

Thanks. will add it.

ivankelly · 2018-02-06T16:53:38Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicsConsumerImpl.java

+        AtomicInteger completed = new AtomicInteger(numberTopicPartitions.get());
+        CompletableFuture<Void> closeFuture = new CompletableFuture<>();
+
+        consumers.values().stream().forEach(consumer -> {


return CompletaleFuture.allOf( consumers.values().stream().map(c -> consumer.closeAsync())) .whenComplete((res,ex) -> { unAckedMessageTracker.close(); client.cleanupConsumer(TopicsConsumerImpl.this); // fail all pending-receive futures to notify application failPendingReceive(); })

Thanks. will change it.

ivankelly · 2018-02-06T17:04:57Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicsConsumerImpl.java

+     */
+    private void addConsumerForOneTopic(ConsumerImpl consumer,
+                                        String topicName,
+                                        AtomicInteger numPartitionsToCreate,


It'd be nicer to replace this atomicInteger + atomicReference pattern, with addConsumerForOneTopic returning a future, and then the caller of addConsumerForOneTopic using CompleteableFuture.allOf to do something when all futures complete and are successful, as in another comment I made above.

Thanks, will change it.

merlimat · 2018-02-09T01:31:56Z

pulsar-client/src/main/java/org/apache/pulsar/client/api/Message.java

+
+    /**
+     * Get the topic name of this message.
+     * This is mainly for TopicsConsumerImpl to identify a message belongs to which topic.


"TopicsConsumerImpl" we shouldn't be referring to implemenation classes from the API javadoc

Thanks. will change it

merlimat · 2018-02-09T01:45:43Z

pulsar-client/src/main/java/org/apache/pulsar/client/api/Message.java

+     *
+     * @return the topic name
+     */
+    default String getTopicName() {


We shouldn't need to use default here, we can always return null in the implementation if needed.

thanks. will change it.

merlimat · 2018-02-09T01:46:03Z

pulsar-client/src/main/java/org/apache/pulsar/client/api/Message.java

+     * Get the topic name of this message.
+     * This is mainly for TopicsConsumerImpl to identify a message belongs to which topic.
+     *
+     * @return the topic name


the name of the topic on which this message was published

merlimat · 2018-02-09T01:48:19Z

pulsar-client/src/main/java/org/apache/pulsar/client/api/MessageId.java

+     *
+     * @return the topic name
+     */
+    default String getTopicName() {


I understand that the implemenation needs to know which topic originated the message so that it can "acknowledge" on the right internal consumer, but I think this doesn't need to be exposed in the MessageId interface.

merlimat · 2018-02-09T01:49:36Z

pulsar-client/src/main/java/org/apache/pulsar/client/api/PulsarClient.java

+     * @return The {@code Consumer} object
+     * @throws PulsarClientException
+     */
+    Consumer subscribe(Collection<String> topics, String subscription) throws PulsarClientException;


These methods should be converted into builder API from #1089

merlimat · 2018-02-09T01:52:15Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicMessageIdImpl.java

+
+import org.apache.pulsar.client.api.MessageId;
+
+public class TopicMessageIdImpl implements MessageId {


This could be extending MessageIdImpl instead of composing

+1. This would reduce the amount of change needed elsewhere also.

Thanks. Here TopicMessageIdImpl is mainly a wrapper for MessageIdImpl, We need keep a reference of MessageIdImpl, because it will be used for one internal-sub-consumer.
If extending, the constructor and getInnerMessageIdInner may be not easy to handle.

merlimat · 2018-02-09T01:53:01Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicMessageImpl.java

+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+
+public class TopicMessageImpl implements Message {


Same here, could we just extend MessageImpl?

Thanks. Here it is also mainly a wrapper for original MessageImpl, the original MessageImpl will be used for one internal-sub-consumer. a extend may be not easy to handle constructor and get.

merlimat · 2018-02-09T01:55:03Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicsConsumerImpl.java

+    protected NamespaceName namespaceName;
+
+    // Map <topic+partition, consumer>, when get do ACK, consumer will by find by topic name
+    private final ConcurrentHashMap<String, ConsumerImpl> consumers;


Most of this logic is very similar to the PartitionedConsumerImpl, we should try to share and reuse as much as possible from that code, since there are a lot of nuances, especially in the consumer part.

Thanks. Yes. There is a plan to make this extends PartitionedConsumerImpl, that is also the reason that this change keep use a lot of same interface.
But it maybe good to make PartitionedConsumerImpl and TopicsConsumerImpl separate at first, this could avoid bring bugs into PartitionedConsumerImpl.

I think 99% of the logic should be unchanged with PartitionedConsumerImpl, the only difference would in constructor (a list of topics vs the number of partitions.. which can be easily be converted in a list of topics anyway), and using the partition index in the message id (which, again, we could use the "partition" topic anyway even in that case).

Yes, the changes for TopicMessageIdImpl, TopicMessageImpl, and UnAckedMessageTracker is aimed to make PartitionedConsumerImpl and TopicsConsumerImpl could be easy merged in the future. opened issue #1236 tracking this.

I would suggest to just rename PartitionedConsumerImpl, change the constructor to accept a list of topics (rather than the number of partitions) and switch to use TopicMessageIdImpl. I don't think that UnAckedMessageTracker usage would need to be changed.

That would avoid to add a lot of new code and having to merge it later.

zhaijack · 2018-02-14T13:54:31Z

opened issue #1237 tracking using builder to create topics consumer.

zhaijack · 2018-02-14T14:17:30Z

@merlimat @ivankelly, Thanks for your comments, Would you please help review it again?

merlimat

👍

sijie · 2018-02-21T21:28:50Z

ping @ivankelly can you take a look at the latest change, so that we can move forward with this feature for 2.0 release.

ivankelly

There's still too much instanceof and casting going on, but approved to unblock the patchset.

ivankelly · 2018-02-21T21:54:46Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java

 import org.apache.pulsar.common.util.collections.ConcurrentOpenHashSet;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

-import io.netty.util.Timeout;
-import io.netty.util.TimerTask;
-
 public class UnAckedMessageTracker implements Closeable {


Github mauled my previous comment. Added some formatting to clarify.

merlimat · 2018-02-21T22:16:32Z

@zhaijack Since the consumer builder changes went in already, please update the API to just be available through the builder.

zhaijack · 2018-02-21T22:18:39Z

@merlimat, Thanks for the reminder, will do the change for builder.

In PR #1103, we add Multi-Topics-Consumer in java client. This is a catch up work to add it in cpp client.

zhaijack changed the title ~~Issue 1100: Provide TopicsConsumer to consume from several topics under same namespace~~ Issue 1100: PIP-13-1: Provide TopicsConsumer to consume from several topics under same namespace Jan 25, 2018

zhaijack force-pushed the sub_regex_topics branch from f294e02 to 52bf92f Compare January 26, 2018 03:08

zhaijack changed the title ~~Issue 1100: PIP-13-1: Provide TopicsConsumer to consume from several topics under same namespace~~ Issue 1100: PIP-13-1/3: Provide TopicsConsumer to consume from several topics under same namespace Jan 26, 2018

zhaijack force-pushed the sub_regex_topics branch 2 times, most recently from 057e9c0 to f391fe2 Compare January 26, 2018 11:37

merlimat assigned zhaijack Jan 26, 2018

merlimat added the type/feature The PR added a new feature or issue requested a new feature label Jan 26, 2018

merlimat added this to the 2.0.0-incubating milestone Jan 26, 2018

zhaijack force-pushed the sub_regex_topics branch 2 times, most recently from 7a19964 to 43242c7 Compare January 31, 2018 15:58

zhaijack changed the title ~~Issue 1100: PIP-13-1/3: Provide TopicsConsumer to consume from several topics under same namespace~~ PIP-13-1/3: Provide TopicsConsumer to consume from several topics under same namespace Feb 1, 2018

This was referenced Feb 1, 2018

PIP-13-1: sub-regex: Provide TopicsConsumer to consume from several topics under same namespace. #1100

Closed

PIP-13-2/3: support regex based subscription #1165

Closed

zhaijack force-pushed the sub_regex_topics branch from 2ca75b5 to af5d6d0 Compare February 4, 2018 12:48

ivankelly requested changes Feb 6, 2018

View reviewed changes

merlimat reviewed Feb 9, 2018

View reviewed changes

zhaijack mentioned this pull request Feb 14, 2018

Make TopicsConsumerImpl extends PartitionedConsumerImpl #1236

Closed

jiazhai added 6 commits February 14, 2018 22:11

add TopicsConsumerImpl and some test

2ae7189

fix license header check, add get method, and fix unit-test fail

6fde7a9

add subscribe and unsubscribe for one topic, add tests

a63e6c9

change to use subscribeAsync(topic) to subscribe topics in init

37ed39b

change following comments

29da59c

rebase master, fix compile issue

a43cb0f

zhaijack force-pushed the sub_regex_topics branch from 26f85e0 to a43cb0f Compare February 14, 2018 14:15

merlimat approved these changes Feb 16, 2018

View reviewed changes

ivankelly approved these changes Feb 21, 2018

View reviewed changes

merlimat merged commit d073a7c into apache:master Feb 21, 2018

zhaijack mentioned this pull request Feb 28, 2018

Documentation for regex_pattern_subscribe #1176

Closed

zhaijack mentioned this pull request Jun 20, 2018

Cpp client: add multiTopicsConsumer #1996

Merged

sijie pushed a commit that referenced this pull request Jul 23, 2018

Cpp client: add multiTopicsConsumer (#1996)

2543ccf

In PR #1103, we add Multi-Topics-Consumer in java client. This is a catch up work to add it in cpp client.

sijie pushed a commit that referenced this pull request Aug 27, 2018

Cpp client: add multiTopicsConsumer (#1996)

500e393

In PR #1103, we add Multi-Topics-Consumer in java client. This is a catch up work to add it in cpp client.

BewareMyPower mentioned this pull request May 22, 2022

On partitioned topics with the Python client, receiver_queue_size=0 results in an InvalidConfiguration error, but is documented to work apache/pulsar-client-python#43

Open


		final String namespace = DestinationName.get(topics.stream().findFirst().get()).getNamespace();

		Optional<String> result = topics.stream()


		import org.apache.pulsar.client.api.MessageId;

		public class TopicMessageIdImpl implements MessageId {

PIP-13-1/3: Provide TopicsConsumer to consume from several topics under same namespace #1103

PIP-13-1/3: Provide TopicsConsumer to consume from several topics under same namespace #1103

Conversation

zhaijack commented Jan 24, 2018

Motivation

Modifications

Result

zhaijack commented Jan 26, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhaijack Feb 9, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhaijack Feb 9, 2018 • edited

Choose a reason for hiding this comment

ivankelly Feb 12, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhaijack Feb 9, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhaijack Feb 13, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhaijack Feb 13, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhaijack Feb 13, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhaijack Feb 14, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhaijack commented Feb 14, 2018

zhaijack commented Feb 14, 2018

merlimat left a comment

Choose a reason for hiding this comment

sijie commented Feb 21, 2018

ivankelly left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

merlimat commented Feb 21, 2018

zhaijack commented Feb 21, 2018

PIP-13-1/3: Provide `TopicsConsumer` to consume from several topics under same namespace #1103

PIP-13-1/3: Provide `TopicsConsumer` to consume from several topics under same namespace #1103

zhaijack Feb 9, 2018 •

edited

zhaijack Feb 9, 2018 •

edited

ivankelly Feb 12, 2018 •

edited

zhaijack Feb 9, 2018 •

edited

zhaijack Feb 13, 2018 •

edited

zhaijack Feb 13, 2018 •

edited

zhaijack Feb 13, 2018 •

edited

zhaijack Feb 14, 2018 •

edited