hachikuji commented on code in PR #17418:
URL: https://github.com/apache/kafka/pull/17418#discussion_r1795953043
##########
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()) {
Review Comment:
Just a side comment, but this state should be impossible? Can we fail the
event with an exception? No one sees warnings.
For a future patch, I think we need to separate the processor for the share
consumer.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -519,12 +518,10 @@ public void subscribe(final Collection<String> topics) {
}
log.info("Subscribed to topic(s): {}", String.join(", ",
topics));
- if (subscriptions.subscribeToShareGroup(new HashSet<>(topics)))
- metadata.requestUpdateForNewTopics();
// Trigger subscribe event to effectively join the group if
not already part of it,
// or just send the new subscription to the broker.
- applicationEventHandler.add(new
ShareSubscriptionChangeEvent());
+ applicationEventHandler.addAndGet(new
ShareSubscriptionChangeEvent(topics));
Review Comment:
nit: Would it makes sense to move the info message above to follow this?
Either that or change to "Subscribing" (such as "Unsubscribing" below), but I
think it is more useful to print the message after the operation has completed.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImplTest.java:
##########
@@ -197,23 +212,34 @@ public void testWakeupBeforeCallingPoll() {
@Test
public void testWakeupAfterEmptyFetch() {
- consumer = newConsumer();
+ SubscriptionState subscriptions = new SubscriptionState(new
LogContext(), OffsetResetStrategy.NONE);
+ consumer = newConsumer(
+ mock(ShareFetchBuffer.class),
+ subscriptions,
+ "group-id",
+ "client-id");
final String topicName = "foo";
- final int partition = 3;
doAnswer(invocation -> {
consumer.wakeup();
return ShareFetch.empty();
}).doAnswer(invocation ->
ShareFetch.empty()).when(fetchCollector).collect(any(ShareFetchBuffer.class));
- consumer.subscribe(singletonList(topicName));
+ final List<String> subscriptionTopic =
Collections.singletonList(topicName);
+
completeShareSubscriptionChangeApplicationEventSuccessfully(subscriptions,
subscriptionTopic);
+ consumer.subscribe(subscriptionTopic);
assertThrows(WakeupException.class, () ->
consumer.poll(Duration.ofMinutes(1)));
assertDoesNotThrow(() -> consumer.poll(Duration.ZERO));
}
@Test
public void testWakeupAfterNonEmptyFetch() {
- consumer = newConsumer();
+ SubscriptionState subscriptions = new SubscriptionState(new
LogContext(), OffsetResetStrategy.NONE);
+ consumer = newConsumer(
Review Comment:
nit: a new helper might make sense since it looks like the same args in all
these tests.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
