Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

KAFKA-18071: Avoid event to refresh regex if no pattern subscription #17917

Open
wants to merge 4 commits into
base: trunk
Choose a base branch
from
+36 −12
Open

KAFKA-18071: Avoid event to refresh regex if no pattern subscription #17917

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c01f39e
Avoid refresh event if no pattern
lianetm Nov 22, 2024
5f220b1
tests
lianetm Nov 22, 2024
26d2ef4
Merge branch 'trunk' into KAFKA-18071
lianetm Nov 22, 2024
f2fe627
Merge branch 'trunk' into KAFKA-18071
lianetm Nov 22, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 8 additions & 6 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1766,12 +1766,14 @@ private void sendPrefetches(Timer timer) {
@Override
public boolean updateAssignmentMetadataIfNeeded(Timer timer) {
offsetCommitCallbackInvoker.executeCallbacks();
try {
applicationEventHandler.addAndGet(new UpdatePatternSubscriptionEvent(calculateDeadlineMs(timer)));
} catch (TimeoutException e) {
return false;
} finally {
timer.update();
if (subscriptions.hasPatternSubscription()) {
try {
applicationEventHandler.addAndGet(new UpdatePatternSubscriptionEvent(calculateDeadlineMs(timer)));
} catch (TimeoutException e) {
return false;
} finally {
timer.update();
}
}
processBackgroundEvents();

Expand Down
7 changes: 4 additions & 3 deletions ...in/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -320,11 +320,12 @@ private void process(final TopicPatternSubscriptionChangeEvent event) {
* This will make the consumer send the updated subscription on the next poll.
*/
private void process(final UpdatePatternSubscriptionEvent event) {
if (!subscriptions.hasPatternSubscription()) {
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is this test really necessary? This event is only sent if the application thread knows there's a pattern subscription.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

truly it's not needed with the current usage of this event from the app thread. I just left it for extra protection in this thread really (ok to remove it if we prefer)

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just checking my understanding. It looks like the remains of an earlier thought :)

return;
}
if (this.metadataVersionSnapshot < metadata.updateVersion()) {
this.metadataVersionSnapshot = metadata.updateVersion();
if (subscriptions.hasPatternSubscription()) {
updatePatternSubscription(metadata.fetch());
}
updatePatternSubscription(metadata.fetch());
}
event.future().complete(null);
}
Expand Down
23 changes: 23 additions & 0 deletions ...nts/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@
import org.apache.kafka.clients.consumer.internals.events.TopicPatternSubscriptionChangeEvent;
import org.apache.kafka.clients.consumer.internals.events.TopicSubscriptionChangeEvent;
import org.apache.kafka.clients.consumer.internals.events.UnsubscribeEvent;
import org.apache.kafka.clients.consumer.internals.events.UpdatePatternSubscriptionEvent;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.TopicPartition;
Expand Down Expand Up @@ -1830,6 +1831,27 @@ public void testSeekToEnd() {
assertEquals(OffsetResetStrategy.LATEST, resetOffsetEvent.offsetResetStrategy());
}

@Test
public void testUpdatePatternSubscriptionEventGeneratedOnlyIfPatternUsed() {
consumer = newConsumer();
doReturn(Fetch.empty()).when(fetchCollector).collectFetch(any(FetchBuffer.class));
when(applicationEventHandler.addAndGet(any(CheckAndUpdatePositionsEvent.class))).thenReturn(true);
doReturn(LeaderAndEpoch.noLeaderOrEpoch()).when(metadata).currentLeader(any());
completeAssignmentChangeEventSuccessfully();
completeTopicPatternSubscriptionChangeEventSuccessfully();
completeUnsubscribeApplicationEventSuccessfully();

consumer.assign(singleton(new TopicPartition("topic1", 0)));
consumer.poll(Duration.ZERO);
verify(applicationEventHandler, never()).addAndGet(any(UpdatePatternSubscriptionEvent.class));

consumer.unsubscribe();

consumer.subscribe(Pattern.compile("t*"));
consumer.poll(Duration.ZERO);
verify(applicationEventHandler).addAndGet(any(UpdatePatternSubscriptionEvent.class));
}

@Test
public void testSubscribeToRe2JPatternValidation() {
consumer = newConsumer();
Expand Down Expand Up @@ -1927,6 +1949,7 @@ private void completeFetchedCommittedOffsetApplicationEventExceptionally(Excepti
private void completeUnsubscribeApplicationEventSuccessfully() {
doAnswer(invocation -> {
UnsubscribeEvent event = invocation.getArgument(0);
consumer.subscriptions().unsubscribe();
event.future().complete(null);
return null;
}).when(applicationEventHandler).add(ArgumentMatchers.isA(UnsubscribeEvent.class));
Expand Down
4 changes: 1 addition & 3 deletions ...ava/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -355,11 +355,10 @@ public void testUpdatePatternSubscriptionEventOnlyTakesEffectWhenMetadataHasNewV
UpdatePatternSubscriptionEvent event1 = new UpdatePatternSubscriptionEvent(12345);

setupProcessor(true);

when(subscriptionState.hasPatternSubscription()).thenReturn(true);
when(metadata.updateVersion()).thenReturn(0);

processor.process(event1);
verify(subscriptionState, never()).hasPatternSubscription();
assertDoesNotThrow(() -> event1.future().get());

Cluster cluster = mock(Cluster.class);
Expand All @@ -377,7 +376,6 @@ public void testUpdatePatternSubscriptionEventOnlyTakesEffectWhenMetadataHasNewV
UpdatePatternSubscriptionEvent event2 = new UpdatePatternSubscriptionEvent(12345);
processor.process(event2);
verify(metadata).requestUpdateForNewTopics();
verify(subscriptionState).hasPatternSubscription();
verify(subscriptionState).subscribeFromPattern(topics);
assertEquals(1, processor.metadataVersionSnapshot());
verify(membershipManager).onSubscriptionUpdated();
Expand Down