AndrewJSchofield commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2811570115
##########
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:
I tend to agree with @lianetm about log checks in the tests. Checking
whether the relevant methods were or were not called is better. It was only
necessary to make the internal class `OffsetFetcherUtils` public in order to
capture its logs. So, I'd vote for not checking the logs in the 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
