kirktrue commented on code in PR #21457:
URL: https://github.com/apache/kafka/pull/21457#discussion_r2802316351
##########
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:
Just for clarity, I will add the `verify()` bit for sure.
##########
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$"
+ );
+
+ if (groupProtocol == GroupProtocol.CLASSIC) {
+ // Classic consumer does not send the LIST_OFFSETS right away
(requires an explicit poll),
+ // different from the new async consumer, that will send the
LIST_OFFSETS request in the background
+ // thread on the next background thread poll.
+ consumer.poll(Duration.ofMillis(0));
+ }
+
+ TestUtils.waitForCondition(
+ () -> requestGenerated(client, ApiKeys.LIST_OFFSETS),
+ "No LIST_OFFSETS request sent within allotted timeout"
+ );
+
+ // Validate the state of the endOffsetRequested flag. It should
still be set before the call to
+ // currentLag(), because the previous LIST_OFFSETS call has not
received a response.
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
+ assertTrue(subscription.partitionEndOffsetRequested(tp0));
+ assertLogEmitted(
+ appender,
+ "^Not requesting the log end offset for " + tp0 + " to compute
lag as an outstanding request already exists$"
+ );
Review Comment:
Agreed. Will do.
--
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]
