kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2818489870
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2729,6 +2734,193 @@ public void testCurrentLag(GroupProtocol groupProtocol)
throws InterruptedExcept
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void testCurrentLagClearsFlagOnFatalPartitionError(GroupProtocol
groupProtocol) throws InterruptedException {
+ try (final LogCaptureAppender appender =
LogCaptureAppender.createAndRegister()) {
+ appender.setClassLogger(OffsetFetcherUtils.class, Level.TRACE);
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ appender.setClassLogger(ClassicKafkaConsumer.class,
Level.INFO);
+ } else {
+ appender.setClassLogger(ApplicationEventProcessor.class,
Level.INFO);
+ appender.setClassLogger(OffsetsRequestManager.class,
Level.TRACE);
+ }
+
+ final ConsumerMetadata metadata = createMetadata(subscription);
+ final MockClient client = new MockClient(time, metadata);
+
+ initMetadata(client, Map.of(topic, 1));
+
+ consumer = newConsumer(groupProtocol, time, client, subscription,
metadata, assignor, false,
+ groupId, groupInstanceId, false);
+ consumer.assign(Set.of(tp0));
+
+ // poll once to update with the current metadata
+ consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No FIND_COORDINATOR request sent within allotted timeout"
+ );
+
client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId,
metadata.fetch().nodes().get(0)));
+
+ // Validate the state of the endOffsetRequested flag. It should be
unset before the call to currentLag(),
+ // then set immediately afterward.
+ assertFalse(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Requesting the log end offset for " + tp0 + " in order to
compute lag$"
+ );
Review Comment:
Log message verification removed.
--
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]
