FrankYang0529 commented on code in PR #17418:
URL: https://github.com/apache/kafka/pull/17418#discussion_r1804894283
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -358,12 +358,19 @@ private void process(final ShareAcknowledgeAsyncEvent
event) {
* consumer join the share group if it is not part of it yet, or send the
updated subscription if
* it is already a member.
*/
- private void process(final ShareSubscriptionChangeEvent ignored) {
+ private void process(final ShareSubscriptionChangeEvent event) {
if (!requestManagers.shareHeartbeatRequestManager.isPresent()) {
- log.warn("Group membership manager not present when processing a
subscribe event");
+ KafkaException error = new KafkaException("Group membership
manager not present when processing a subscribe event");
+ event.future().completeExceptionally(error);
return;
}
+
+ if (subscriptions.subscribeToShareGroup(event.topics()))
+ metadata.requestUpdateForNewTopics();
Review Comment:
Hi @chia7712, I tried to write a test case in
`PlaintextConsumerSubscriptionTest`, but I can't reproduce it. Could you share
your test case? Thank you.
```scala
@ParameterizedTest(name =
TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
@MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
def testSubscribeOtherTopicNotInPreviousSubscriptions(quorum: String,
groupProtocol: String): Unit = {
val initialAssignment = Set(new TopicPartition(topic, 0), new
TopicPartition(topic, 1))
val consumer = createConsumer()
consumer.subscribe(List(topic).asJava)
awaitAssignment(consumer, initialAssignment)
assertDoesNotThrow(new Executable {
override def execute(): Unit = consumer.poll(Duration.ofMillis(1000))
})
val otherTopic = "other"
createTopic(otherTopic, 2, brokerCount)
val otherAssignment = Set(new TopicPartition(otherTopic, 0), new
TopicPartition(otherTopic, 1))
consumer.subscribe(List(otherTopic).asJava)
awaitAssignment(consumer, otherAssignment)
assertDoesNotThrow(new Executable {
override def execute(): Unit = consumer.poll(Duration.ofMillis(1000))
})
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
