[fix][client] Fix subscription topic name error.

[shibd]

#16374 #16556 #16375 #16599

Motivation

This is a bug, after [PIP-145], PatternMultiTopicsConsumer can receive CommandWatchTopicUpdate from broker to subscribe new topic.

But this topic name is with the partition index(case: public/default/test-topic-partition-0), This will produce unexpected behavior when executing the following subscribe method.

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

Lines 935 to 942 in 9c93ab4

	client.getPartitionedTopicMetadata(topicName)
	.thenAccept(metadata -> subscribeTopicPartitions(subscribeResult, fullTopicName, metadata.partitions,
	createTopicIfDoesNotExist))
	.exceptionally(ex1 -> {
	log.warn("[{}] Failed to get partitioned topic metadata: {}", fullTopicName, ex1.getMessage());
	subscribeResult.completeExceptionally(ex1);
	return null;
	});

The reasons for these failed unit tests are:

When the client receives the CommandWatchTopicUpdate command, it will enter the logic of processing non-partitions, then only process the consumer collection and not the partitionedTopics collection

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

Lines 1068 to 1094 in 9c93ab4

	} else {
	allTopicPartitionsNumber.incrementAndGet();
	CompletableFuture<Consumer<T>> subFuture = new CompletableFuture<>();
	consumers.compute(topicName, (key, existingValue) -> {
	if (existingValue != null) {
	String errorMessage = String.format("[%s] Failed to subscribe for topic [%s] in topics consumer. "
	+ "Topic is already being subscribed for in other thread.", topic, topicName);
	log.warn(errorMessage);
	subscribeResult.completeExceptionally(new PulsarClientException(errorMessage));
	return existingValue;
	} else {
	internalConfig.setStartPaused(paused);
	ConsumerImpl<T> newConsumer = createInternalConsumer(internalConfig, topicName,
	-1, subFuture, createIfDoesNotExist, schema);
	synchronized (pauseMutex) {
	if (paused) {
	newConsumer.pause();
	} else {
	newConsumer.resume();
	}
	}
	return newConsumer;
	}
	});

Then, when we execute the consumer1.run(consumer1.getRecheckPatternTimeout()) to trigger the update subscription on the unit test, By executing the following logic, will cause topic-4 to be removed and added at the same time. ultimately destabilizes unit tests (maybe deletes will be executed after additions)

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

Lines 115 to 121 in 9c93ab4

	final List<String> oldTopics = new ArrayList<>(getPartitionedTopics());
	for (String partition : getPartitions()) {
	TopicName topicName = TopicName.get(partition);
	if (!topicName.isPartitioned() || !oldTopics.contains(topicName.getPartitionedTopicName())) {
	oldTopics.add(partition);
	}
	}

Print log here:

Can See:

2022-07-21T19:43:35,883 - INFO  - [pulsar-client-io-397-1:PatternMultiTopicsConsumerImpl@154] - debug add: [persistent://my-property/my-ns/pattern-topic-4-AutoSubscribePatternConsumer]
2022-07-21T19:43:35,883 - INFO  - [pulsar-client-io-397-1:PatternMultiTopicsConsumerImpl@155] - debug remove: [persistent://my-property/my-ns/pattern-topic-4-AutoSubscribePatternConsumer-partition-0, persistent://my-property/my-ns/pattern-topic-4-AutoSubscribePatternConsumer-partition-1, persistent://my-property/my-ns/pattern-topic-4-AutoSubscribePatternConsumer-partition-2, persistent://my-property/my-ns/pattern-topic-4-AutoSubscribePatternConsumer-partition-3]

Modifications

• Change subscribe logic, use the getPartitionedTopicName().

Documentation

• doc-not-needed
  (Please explain why)

[shibd]

@codelipenghui @gaozhangmin @Technoboy- @nodece Please help review.

[codelipenghui]

Nice catch!

[mattisonchao]

Good catch!

[mattisonchao]

@andrasbeni Could you help to review it?

[BewareMyPower]

Could you add a unit test to protect this change?

[shibd]

Could you add a unit test to protect this change?

@BewareMyPower Thanks for your reminder, I looked again, and It is found that we cannot be modify the logic in the MultiTopicsConsumerImpl.subscribeAsync method. Because users may use consumer like this:

        client.newConsumer().topic(
                        "persistent://public/dafault/test-topic1-partition-0",
                        "persistent://public/dafault/test-topic2-partition-1"
         );

If force subscription to PartitionTopic inside the method, Then all partitions of test-topic1 and test-topic2 will be subscribed. This is a breaking change.

So, I revert the original changes.

The new changes are: Inside the PatternTopicsChangedListener.onTopicsAdded method let it use partitioned name to subscribe.

@codelipenghui @mattisonchao @Technoboy- @nodece @BewareMyPower Sorry, please help review again.

[codelipenghui]

The new changes are: Inside the PatternTopicsChangedListener.onTopicsAdded method let it use partitioned name to subscribe.

Make sense.

[shibd]

/pulsarbot run-failure-checks