kirktrue commented on code in PR #16673:
URL: https://github.com/apache/kafka/pull/16673#discussion_r1722536349
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1443,12 +1442,9 @@ public void assign(Collection<TopicPartition>
partitions) {
// be no following rebalance.
//
// See the ApplicationEventProcessor.process() method that handles
this event for more detail.
- applicationEventHandler.add(new
AssignmentChangeEvent(subscriptions.allConsumed(), time.milliseconds()));
-
- log.info("Assigned to partition(s): {}",
partitions.stream().map(TopicPartition::toString).collect(Collectors.joining(",
")));
-
- if (subscriptions.assignFromUser(new HashSet<>(partitions)))
- applicationEventHandler.add(new
NewTopicsMetadataUpdateRequestEvent());
+ Timer timer = time.timer(defaultApiTimeoutMs);
+ AssignmentChangeEvent assignmentChangeEvent = new
AssignmentChangeEvent(timer.currentTimeMs(), calculateDeadlineMs(timer),
partitions);
+ applicationEventHandler.addAndGet(assignmentChangeEvent);
Review Comment:
With this change we're now blocking on this operation completing. Is that
change intentional?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerUtils.java:
##########
@@ -231,6 +231,8 @@ public static <T> T getResult(Future<T> future) {
try {
return future.get();
} catch (ExecutionException e) {
+ if (e.getCause() instanceof IllegalStateException)
Review Comment:
Do we need to have the same logic here as in the `getResult()` on line 218?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessorTest.java:
##########
@@ -145,6 +148,39 @@ public void testResetPositionsProcess() {
verify(applicationEventProcessor).process(any(ResetPositionsEvent.class));
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testAssignmentChangeEvent(boolean withGroupId) {
+ final long currentTimeMs = 12345;
+ AssignmentChangeEvent event = new AssignmentChangeEvent(currentTimeMs,
12345, Collections.emptyList());
+
+ setupProcessor(withGroupId);
+ doReturn(true).when(subscriptionState).assignFromUser(any());
+ processor.process(event);
+ if (withGroupId) {
+ verify(commitRequestManager).updateAutoCommitTimer(currentTimeMs);
+ verify(commitRequestManager).maybeAutoCommitAsync();
+ } else {
+ verify(commitRequestManager,
never()).updateAutoCommitTimer(currentTimeMs);
+ verify(commitRequestManager, never()).maybeAutoCommitAsync();
+ }
+ verify(metadata).requestUpdateForNewTopics();
+ assertDoesNotThrow(() -> event.future().get());
+
+ }
+
+ @Test
+ public void testAssignmentChangeEventWithException() {
+ AssignmentChangeEvent event = new AssignmentChangeEvent(12345, 12345,
Collections.emptyList());
+
+ setupProcessor(false);
+ doThrow(new
IllegalStateException()).when(subscriptionState).assignFromUser(any());
+ processor.process(event);
+
+ ExecutionException e = assertThrows(ExecutionException.class, () ->
event.future().get());
+ assertInstanceOf(IllegalStateException.class, e.getCause());
Review Comment:
Where is the `IllegalStateException` thrown in a production setting? I see
we special-cased `IllegalStateException` in `ConsumerUtils`, but I'm wondering
if an `IllegalStateException` is something a user can hit 🤔
--
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
