You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "FrankYang0529 (via GitHub)" <gi...@apache.org> on 2024/08/23 10:05:53 UTC
[PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
FrankYang0529 opened a new pull request, #16982:
URL: https://github.com/apache/kafka/pull/16982
Fix following tests for CONSUMER group protocol:
* testCurrentLag
* testFetchStableOffsetThrowInPoll
* testListOffsetShouldUpdateSubscriptions
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1729165197
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2471,6 +2477,8 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
// poll once to update with the current metadata
consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(() -> client.requests().stream().anyMatch(request -> request.requestBuilder().apiKey().equals(ApiKeys.FIND_COORDINATOR)),
Review Comment:
Updated it. Thanks for the suggestion.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1752608667
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2521,35 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
+ @EnumSource(GroupProtocol.class)
public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
consumer.seek(tp0, 50L);
- client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
+ client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
Review Comment:
for the record, I was concerned here mostly about fetch requests (since we have a partition assigned and not paused), but just noticed we're not calling consumer.poll (so we shouldn't expect FETCH requests to be generated)
Still, worth mentioning that without this in-flight fix https://github.com/apache/kafka/pull/17035 for Fetch, we could wrongly generate fetch requests even without polling I guess. So wonder if enabling this test like this would be flaky for the new consumer without that fix.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1728850049
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2471,6 +2477,8 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
// poll once to update with the current metadata
consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(() -> client.requests().stream().anyMatch(request -> request.requestBuilder().apiKey().equals(ApiKeys.FIND_COORDINATOR)),
Review Comment:
interesting, makes sense how you keep a single poll, and we just have to wait a bit for the background to actually generate the request. What about encapsulating this bit to make it clearer in all places it's used, to end up with something like:
`TestUtils.waitForCondition(() -> requestGenerated(ApiKeys.FIND_COORDINATOR))`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1752557956
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2521,35 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
+ @EnumSource(GroupProtocol.class)
public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
consumer.seek(tp0, 50L);
- client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
+ client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
Review Comment:
makes sense, but I still have a concern with this (wonder if it would make this test flaky). Let's say the moment we get to prepare this response, there is another request that got generated (a fetch request, for instance). Wouldn't that make that the endOffsets request will never find the response? (Not sure if I'm missing details of how prepareResponse works)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1733248177
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2481,23 +2489,31 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> {
+ boolean hasListOffsetRequest = requestGenerated(client, ApiKeys.LIST_OFFSETS);
+ boolean fetchRequest = requestGenerated(client, ApiKeys.FETCH);
+ return hasListOffsetRequest && fetchRequest;
+ }, "No list-offset & fetch request sent");
// no error for no end offset (so unknown lag)
assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
// poll once again, which should return the list-offset response
// and hence next call would return correct lag result
- client.respond(listOffsetsResponse(singletonMap(tp0, 90L)));
+ Optional<ClientRequest> listOffsetRequest = client.requests().stream().filter(request -> request.requestBuilder().apiKey().equals(ApiKeys.LIST_OFFSETS)).findFirst();
Review Comment:
this line seems tricky to read and we need it twice (for now). Would it be clearer to have a helper along the lines of `findRequest(client, ApiKeys.LIST_OFFSETS)`?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1739840322
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2521,35 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
+ @EnumSource(GroupProtocol.class)
public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
consumer.seek(tp0, 50L);
- client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
+ client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
Review Comment:
Can `client.prepareResponse` be moved to `newConsumer`?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1762133880
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2479,23 +2487,29 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
Review Comment:
Hi @lianetm, thanks for filing the Jira. I would like to give "eventually updated" a try in this PR. If it still can't make the test stable, we can put it in next Jira. What do you think? Thank you.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1776899718
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2486,68 +2492,96 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
// poll once to update with the current metadata
consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No metadata requests sent");
client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
// no error for no current position
assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
- assertEquals(0, client.inFlightRequestCount());
-
+ 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.
+ assertEquals(0, client.inFlightRequestCount());
+ }
// poll once again, which should send the list-offset request
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> {
+ boolean hasListOffsetRequest = requestGenerated(client, ApiKeys.LIST_OFFSETS);
+ boolean hasFetchRequest = requestGenerated(client, ApiKeys.FETCH);
+ return hasListOffsetRequest && hasFetchRequest;
+ }, "No list-offset & fetch request sent");
// no error for no end offset (so unknown lag)
assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
// poll once again, which should return the list-offset response
// and hence next call would return correct lag result
- client.respond(listOffsetsResponse(singletonMap(tp0, 90L)));
+ ClientRequest listOffsetRequest = findRequest(client, ApiKeys.LIST_OFFSETS);
+ client.respondToRequest(listOffsetRequest, listOffsetsResponse(singletonMap(tp0, 90L)));
consumer.poll(Duration.ofMillis(0));
- assertEquals(OptionalLong.of(40L), consumer.currentLag(tp0));
+ // For AsyncKafkaConsumer, subscription sate is updated in background, so the result will eventually be updated.
+ TestUtils.waitForCondition(() -> {
+ OptionalLong result = consumer.currentLag(tp0);
+ return result.isPresent() && result.getAsLong() == 40L;
+ }, "Subscription state is not updated");
// requests: fetch
- assertEquals(1, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FETCH), "No fetch request sent");
// one successful fetch should update the log end offset and the position
+ ClientRequest fetchRequest = findRequest(client, ApiKeys.FETCH);
final FetchInfo fetchInfo = new FetchInfo(1L, 99L, 50L, 5);
- client.respond(fetchResponse(singletonMap(tp0, fetchInfo)));
+ client.respondToRequest(fetchRequest, fetchResponse(singletonMap(tp0, fetchInfo)));
final ConsumerRecords<String, String> records = (ConsumerRecords<String, String>) consumer.poll(Duration.ofMillis(1));
assertEquals(5, records.count());
assertEquals(55L, consumer.position(tp0));
// correct lag result
- assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
+ // For AsyncKafkaConsumer, subscription sate is updated in background, so the result will eventually be updated.
+ TestUtils.waitForCondition(() -> {
Review Comment:
Yes, you're right, if `consumer.position` can get `45, then the subscription state has already been updated. Remove `TestUtils.waitForCondition` here. Thanks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #16982:
URL: https://github.com/apache/kafka/pull/16982#issuecomment-2313116121
Hey @FrankYang0529 , some other minor comments left, almost there. Thanks!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1739141821
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -1053,12 +1053,20 @@ public void testFetchStableOffsetThrowInCommitted(GroupProtocol groupProtocol) {
assertThrows(UnsupportedVersionException.class, () -> setupThrowableConsumer(groupProtocol).committed(Collections.singleton(tp0)));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
- public void testFetchStableOffsetThrowInPoll(GroupProtocol groupProtocol) {
- assertThrows(UnsupportedVersionException.class, () -> setupThrowableConsumer(groupProtocol).poll(Duration.ZERO));
+ @EnumSource(GroupProtocol.class)
+ public void testFetchStableOffsetThrowInPoll(GroupProtocol groupProtocol) throws InterruptedException {
+ setupThrowableConsumer(groupProtocol);
+ TestUtils.waitForCondition(() -> {
Review Comment:
this change still has me thinking. This test is about a single call to `poll(ZERO)`, that is expected to throw an exception, but interesting fact is that the exception is generated when building the request [here](https://github.com/apache/kafka/blob/70dd577286de31ef20dc4f198e95f9b9e4479b47/clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java#L146) (it does not require the actual send or response). So I wonder if the async consumer should somehow ensure that when poll returns (even with low time), it has allowed for at least one run of the background thread runOnce?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1739840322
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2521,35 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
+ @EnumSource(GroupProtocol.class)
public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
consumer.seek(tp0, 50L);
- client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
+ client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
Review Comment:
Could `client.prepareResponse` be moved to `newConsumer` before?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1752608667
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2521,35 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
+ @EnumSource(GroupProtocol.class)
public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
consumer.seek(tp0, 50L);
- client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
+ client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
Review Comment:
for the record, I was concerned here mostly about fetch requests (since we have a partition assigned and not paused), but just noticed we're not calling consumer.poll (so we shouldn't expect FETCH requests to be generated)
Still, worth mentioning that without this in-flight fix https://github.com/apache/kafka/pull/17035 for Fetch, we could wrongly generate fetch requests even without polling. So enabling this test like this would be flaky for the new consumer without that fix I guess.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1752541951
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2479,23 +2487,29 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
Review Comment:
similar to the reasoning that led to this fix, there is a check above for:
> assertEquals(0, client.inFlightRequestCount())
I would say that may be flaky for the new consumer, right?
We cannot ensure that no requests will be generated just because the last api we called shouldn't generate one. Given that we called poll before (and there is a background thread running) I expect there could be requests generated for fetching and fetching offsets maybe? The point of that assert is to ensure there was no ListOffsets request generated, so I would say that we should look for that exact request type, and assert there is none. Makes sense?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1731802528
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2523,16 +2537,32 @@ public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol)
// poll once to update with the current metadata
consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No metadata requests sent");
client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
consumer.seek(tp0, 50L);
+
+ if (groupProtocol == GroupProtocol.CONSUMER) {
Review Comment:
I see how you make the test pass, and that is fine and works, but I believe we could greatly simplify this test if we get the FindCoord, OffsetFetch and FetchRequests out of the picture, which seem to me are really not relevant to this test?
This test is all about partition offsets (end offsets stored in the leader), compared to a position set manually with seek, so not related at all with committed offsets are affected by fetch.
So, this is the simplification I'm thinking about:
1. we create the consumer without groupId -> this will ensure we don't send any OffsetFetch request. We don't really need them given that we're only playing with the partition offsets
2. we pause the partition right after assign -> this ensures that we don't issue fetch requests. We don't really need them for this test, and having them makes the test different for both consumers
With those 2 small changes, I would expect we can keep the same test for both consumer, without any specifics for the new consumer, and it would still be true to its purpose, testing what it has always tested. What do you think?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1734259073
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2481,23 +2489,31 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> {
+ boolean hasListOffsetRequest = requestGenerated(client, ApiKeys.LIST_OFFSETS);
+ boolean fetchRequest = requestGenerated(client, ApiKeys.FETCH);
+ return hasListOffsetRequest && fetchRequest;
+ }, "No list-offset & fetch request sent");
// no error for no end offset (so unknown lag)
assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
// poll once again, which should return the list-offset response
// and hence next call would return correct lag result
- client.respond(listOffsetsResponse(singletonMap(tp0, 90L)));
+ Optional<ClientRequest> listOffsetRequest = client.requests().stream().filter(request -> request.requestBuilder().apiKey().equals(ApiKeys.LIST_OFFSETS)).findFirst();
+ assertTrue(listOffsetRequest.isPresent());
+ client.respondToRequest(listOffsetRequest.get(), listOffsetsResponse(singletonMap(tp0, 90L)));
consumer.poll(Duration.ofMillis(0));
assertEquals(OptionalLong.of(40L), consumer.currentLag(tp0));
// requests: fetch
- assertEquals(1, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FETCH), "No fetch request sent");
// one successful fetch should update the log end offset and the position
+ Optional<ClientRequest> fetchRequest = client.requests().stream().filter(request -> request.requestBuilder().apiKey().equals(ApiKeys.FETCH)).findFirst();
final FetchInfo fetchInfo = new FetchInfo(1L, 99L, 50L, 5);
- client.respond(fetchResponse(singletonMap(tp0, fetchInfo)));
+ assertTrue(fetchRequest.isPresent());
Review Comment:
Hi @lianetm, thanks for the review and suggestion. I addressed other comments. For this one, I would like to keep it. If we remove it, there may have warning message like `'Optional. get()' without 'isPresent()' check`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1733254682
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2481,23 +2489,31 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> {
+ boolean hasListOffsetRequest = requestGenerated(client, ApiKeys.LIST_OFFSETS);
+ boolean fetchRequest = requestGenerated(client, ApiKeys.FETCH);
+ return hasListOffsetRequest && fetchRequest;
+ }, "No list-offset & fetch request sent");
// no error for no end offset (so unknown lag)
assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
// poll once again, which should return the list-offset response
// and hence next call would return correct lag result
- client.respond(listOffsetsResponse(singletonMap(tp0, 90L)));
+ Optional<ClientRequest> listOffsetRequest = client.requests().stream().filter(request -> request.requestBuilder().apiKey().equals(ApiKeys.LIST_OFFSETS)).findFirst();
+ assertTrue(listOffsetRequest.isPresent());
+ client.respondToRequest(listOffsetRequest.get(), listOffsetsResponse(singletonMap(tp0, 90L)));
consumer.poll(Duration.ofMillis(0));
assertEquals(OptionalLong.of(40L), consumer.currentLag(tp0));
// requests: fetch
- assertEquals(1, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FETCH), "No fetch request sent");
// one successful fetch should update the log end offset and the position
+ Optional<ClientRequest> fetchRequest = client.requests().stream().filter(request -> request.requestBuilder().apiKey().equals(ApiKeys.FETCH)).findFirst();
final FetchInfo fetchInfo = new FetchInfo(1L, 99L, 50L, 5);
- client.respond(fetchResponse(singletonMap(tp0, fetchInfo)));
+ assertTrue(fetchRequest.isPresent());
Review Comment:
we probably don't need this assertion given that we already checked that the request was generated on the waitForCondition on ln 2510?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1752541951
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2479,23 +2487,29 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
Review Comment:
similar to the reasoning that led to this fix, there is a check above for:
> assertEquals(0, client.inFlightRequestCount())
I would say that may be flaky for the new consumer, right?
We cannot ensure that no requests will be generated just because the last api we called shouldn't generate one. Given that we called poll before (and there is a background thread running) I expect there could be requests generated for fetching and fetching offsets maybe. The point of that assert is to ensure there was no ListOffsets request generated, so I would say that we should look for that exact request type, and assert there is none. Makes sense?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2479,23 +2487,29 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
Review Comment:
similar to the reasoning that led to this fix, there is a check above for:
> assertEquals(0, client.inFlightRequestCount())
I would say that may be flaky for the new consumer, right?
We cannot ensure that no requests will be generated just because the last api we called shouldn't generate one. Given that we called poll before (and there is a background thread running) I expect there could be requests generated for fetching and fetching offsets maybe. The point of that assert is to ensure there was no ListOffsets request generated, so I would say that we should look for that exact request type, and assert there is none. Makes sense?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1756698094
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2479,23 +2487,29 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
Review Comment:
Hi @lianetm, I added the assertion back and comments. I run 50 rounds on my laptop and there is no error. Let's latest CI result. Thank you.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1761932369
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2479,23 +2487,29 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
Review Comment:
Hey @FrankYang0529, this testCurrentLag still seems flaky even after the latest changes, I filed https://issues.apache.org/jira/browse/KAFKA-17560 with what I see and where I imagine the flakiness may be, but it needs more thinking probably. Here's a suggestion to make progress:
- we could leave `testCurrrentLag` disabled for the new consumer on this PR (just as it was before)
- we unblock the other 2 tests, that seems to pass consistently after your changes
- we address `testCurrentLag` in a separate PR, with that jira I created (including the changes you had here, just that I think it needs more)
What do you think?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1766797447
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2479,23 +2487,29 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
Review Comment:
Hi @lianetm, I rebased the code again today and latest CI result looks good. Could you help me review it when you have time? Thank you.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1746944955
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -1053,12 +1053,20 @@ public void testFetchStableOffsetThrowInCommitted(GroupProtocol groupProtocol) {
assertThrows(UnsupportedVersionException.class, () -> setupThrowableConsumer(groupProtocol).committed(Collections.singleton(tp0)));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
- public void testFetchStableOffsetThrowInPoll(GroupProtocol groupProtocol) {
- assertThrows(UnsupportedVersionException.class, () -> setupThrowableConsumer(groupProtocol).poll(Duration.ZERO));
+ @EnumSource(GroupProtocol.class)
+ public void testFetchStableOffsetThrowInPoll(GroupProtocol groupProtocol) throws InterruptedException {
+ setupThrowableConsumer(groupProtocol);
+ TestUtils.waitForCondition(() -> {
Review Comment:
Hi @lianetm, the `ApplicationEventProcessor#process(PollEvent)` only resets timer for some request managers. Even if we use `ApplicationEventHandler#addAndGet`, we still can't guarantee `runOnce` is executed. We may need `ConsumerNetworkThread` to send a `BackgroundEvent` to make sure the `runOnce` is happened. However, this approach may make the process more complex. If we want to make this, we probably can create another Jira to handle this. WDYT?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1736345815
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2523,33 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
- public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
+ @EnumSource(GroupProtocol.class)
+ public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) throws InterruptedException {
Review Comment:
Thanks for the reminder. Removed it.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1736346494
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2523,33 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
- public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
+ @EnumSource(GroupProtocol.class)
+ public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) throws InterruptedException {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
consumer.seek(tp0, 50L);
- client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
+ client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
assertEquals(singletonMap(tp0, 90L), consumer.endOffsets(Collections.singleton(tp0)));
// correct lag result should be returned as well
assertEquals(OptionalLong.of(40L), consumer.currentLag(tp0));
}
+ private Optional<ClientRequest> findRequest(MockClient client, ApiKeys apiKey) {
+ return client.requests().stream().filter(request -> request.requestBuilder().apiKey().equals(apiKey)).findFirst();
+ }
+
+ private boolean requestGenerated(MockClient client, ApiKeys apiKey) {
+ return client.requests().stream().anyMatch(request -> request.requestBuilder().apiKey().equals(apiKey));
Review Comment:
Sorry, I applied @kirktrue's suggestion, so we can't use `findRequest(client, apiKey).isPresent()` here.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1733154825
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -1053,12 +1053,20 @@ public void testFetchStableOffsetThrowInCommitted(GroupProtocol groupProtocol) {
assertThrows(UnsupportedVersionException.class, () -> setupThrowableConsumer(groupProtocol).committed(Collections.singleton(tp0)));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
- public void testFetchStableOffsetThrowInPoll(GroupProtocol groupProtocol) {
- assertThrows(UnsupportedVersionException.class, () -> setupThrowableConsumer(groupProtocol).poll(Duration.ZERO));
+ @EnumSource(GroupProtocol.class)
+ public void testFetchStableOffsetThrowInPoll(GroupProtocol groupProtocol) throws InterruptedException {
+ setupThrowableConsumer(groupProtocol);
+ TestUtils.waitForCondition(() -> {
+ try {
+ // For CONSUMER protocol, the offset request is sent in the background thread,
+ // so we need to wait until the offset request is sent.
+ consumer.poll(Duration.ZERO);
+ return false;
+ } catch (UnsupportedVersionException e) {
+ return true;
+ }
+ }, "Failed to fetch stable offset");
Review Comment:
I think this msg is not quite right. If this condition fails, it means that the "consumer failed to throw UnsupportedVersionException on poll" (we actually expect that it fails to fetch the offsets)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1731802528
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2523,16 +2537,32 @@ public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol)
// poll once to update with the current metadata
consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No metadata requests sent");
client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
consumer.seek(tp0, 50L);
+
+ if (groupProtocol == GroupProtocol.CONSUMER) {
Review Comment:
I see how you make the test pass, and that is fine and works, but I believe we could greatly simplify this test if we get the FindCoord, OffsetFetch and FetchRequests out of the picture, which seem to me are really not relevant to this test?
This test is all about partition offsets (end offsets stored in the leader), compared to a position set manually with seek, so not related at all with committed offsets or affected by fetch.
So, this is the simplification I'm thinking about:
1. we create the consumer without groupId -> this will ensure we don't send any OffsetFetch request. We don't really need them given that we're only playing with the partition offsets
2. we pause the partition right after assign -> this ensures that we don't issue fetch requests. We don't really need them for this test, and having them makes the test different for both consumers
With those 2 small changes, I would expect we can keep the same test for both consumer, without any specifics for the new consumer, and it would still be true to its purpose, testing what it has always tested. What do you think?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1764245222
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2479,23 +2487,29 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
Review Comment:
Hi @lianetm, do you think I need to rebase code, so we can have more CI running result? Or, it's good to keep current state? Thank you.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1776899718
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2486,68 +2492,96 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
// poll once to update with the current metadata
consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No metadata requests sent");
client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
// no error for no current position
assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
- assertEquals(0, client.inFlightRequestCount());
-
+ 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.
+ assertEquals(0, client.inFlightRequestCount());
+ }
// poll once again, which should send the list-offset request
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> {
+ boolean hasListOffsetRequest = requestGenerated(client, ApiKeys.LIST_OFFSETS);
+ boolean hasFetchRequest = requestGenerated(client, ApiKeys.FETCH);
+ return hasListOffsetRequest && hasFetchRequest;
+ }, "No list-offset & fetch request sent");
// no error for no end offset (so unknown lag)
assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
// poll once again, which should return the list-offset response
// and hence next call would return correct lag result
- client.respond(listOffsetsResponse(singletonMap(tp0, 90L)));
+ ClientRequest listOffsetRequest = findRequest(client, ApiKeys.LIST_OFFSETS);
+ client.respondToRequest(listOffsetRequest, listOffsetsResponse(singletonMap(tp0, 90L)));
consumer.poll(Duration.ofMillis(0));
- assertEquals(OptionalLong.of(40L), consumer.currentLag(tp0));
+ // For AsyncKafkaConsumer, subscription sate is updated in background, so the result will eventually be updated.
+ TestUtils.waitForCondition(() -> {
+ OptionalLong result = consumer.currentLag(tp0);
+ return result.isPresent() && result.getAsLong() == 40L;
+ }, "Subscription state is not updated");
// requests: fetch
- assertEquals(1, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FETCH), "No fetch request sent");
// one successful fetch should update the log end offset and the position
+ ClientRequest fetchRequest = findRequest(client, ApiKeys.FETCH);
final FetchInfo fetchInfo = new FetchInfo(1L, 99L, 50L, 5);
- client.respond(fetchResponse(singletonMap(tp0, fetchInfo)));
+ client.respondToRequest(fetchRequest, fetchResponse(singletonMap(tp0, fetchInfo)));
final ConsumerRecords<String, String> records = (ConsumerRecords<String, String>) consumer.poll(Duration.ofMillis(1));
assertEquals(5, records.count());
assertEquals(55L, consumer.position(tp0));
// correct lag result
- assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
+ // For AsyncKafkaConsumer, subscription sate is updated in background, so the result will eventually be updated.
+ TestUtils.waitForCondition(() -> {
Review Comment:
Yes, you're right, if `consumer.position` can get `45`, then the subscription state has already been updated. Remove `TestUtils.waitForCondition` here. Thanks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1776796286
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2486,68 +2492,96 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
// poll once to update with the current metadata
consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No metadata requests sent");
client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
// no error for no current position
assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
- assertEquals(0, client.inFlightRequestCount());
-
+ 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.
+ assertEquals(0, client.inFlightRequestCount());
+ }
// poll once again, which should send the list-offset request
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> {
+ boolean hasListOffsetRequest = requestGenerated(client, ApiKeys.LIST_OFFSETS);
+ boolean hasFetchRequest = requestGenerated(client, ApiKeys.FETCH);
+ return hasListOffsetRequest && hasFetchRequest;
+ }, "No list-offset & fetch request sent");
// no error for no end offset (so unknown lag)
assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
// poll once again, which should return the list-offset response
// and hence next call would return correct lag result
- client.respond(listOffsetsResponse(singletonMap(tp0, 90L)));
+ ClientRequest listOffsetRequest = findRequest(client, ApiKeys.LIST_OFFSETS);
+ client.respondToRequest(listOffsetRequest, listOffsetsResponse(singletonMap(tp0, 90L)));
consumer.poll(Duration.ofMillis(0));
- assertEquals(OptionalLong.of(40L), consumer.currentLag(tp0));
+ // For AsyncKafkaConsumer, subscription sate is updated in background, so the result will eventually be updated.
+ TestUtils.waitForCondition(() -> {
+ OptionalLong result = consumer.currentLag(tp0);
+ return result.isPresent() && result.getAsLong() == 40L;
+ }, "Subscription state is not updated");
// requests: fetch
- assertEquals(1, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FETCH), "No fetch request sent");
// one successful fetch should update the log end offset and the position
+ ClientRequest fetchRequest = findRequest(client, ApiKeys.FETCH);
final FetchInfo fetchInfo = new FetchInfo(1L, 99L, 50L, 5);
- client.respond(fetchResponse(singletonMap(tp0, fetchInfo)));
+ client.respondToRequest(fetchRequest, fetchResponse(singletonMap(tp0, fetchInfo)));
final ConsumerRecords<String, String> records = (ConsumerRecords<String, String>) consumer.poll(Duration.ofMillis(1));
assertEquals(5, records.count());
assertEquals(55L, consumer.position(tp0));
// correct lag result
- assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
+ // For AsyncKafkaConsumer, subscription sate is updated in background, so the result will eventually be updated.
+ TestUtils.waitForCondition(() -> {
Review Comment:
Is this change really needed? In this case we just did a successful fetch, so position is updated to 55 (ln 2541). We should be able to retrieve the lag of 45 (end offsets is already known to be 100). (Is not exactly the same case as above, where we needed to allow for the ListOffsets response to be processed in the background). Makes sense?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1747137112
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2521,35 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
+ @EnumSource(GroupProtocol.class)
public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
consumer.seek(tp0, 50L);
- client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
+ client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
Review Comment:
Hi @lianetm and @TaiJuWu, thanks for the suggestion, but I think we can keep this line just before `consumer.endOffsets`.
In `AsyncKafkaConsumer#endOffsets`, it sends `ListOffsetsEvent` and `ApplicationEventProcessor#process(ListOffsetsEvent)` calls `OffsetsRequestManager#fetchOffsets`.
In `OffsetsRequestManager#fetchOffsets`, it builds the request and `AsyncKafkaConsumer#endOffsets` uses `ApplicationEventHandler#addAndGet` to wait for the result, so I think it's safe to put it just before `consumer.endOffsets`. WDYT? Thank you.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1747085039
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2521,35 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
+ @EnumSource(GroupProtocol.class)
public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
consumer.seek(tp0, 50L);
- client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
+ client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
Review Comment:
@FrankYang0529 could you address this? I guess the suggestion comes from the recent fix to some tests that were flaky for a similar situation (https://github.com/apache/kafka/pull/17056)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1776792978
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2486,68 +2492,96 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
// poll once to update with the current metadata
consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No metadata requests sent");
client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
// no error for no current position
assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
- assertEquals(0, client.inFlightRequestCount());
-
+ 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.
+ assertEquals(0, client.inFlightRequestCount());
+ }
// poll once again, which should send the list-offset request
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> {
+ boolean hasListOffsetRequest = requestGenerated(client, ApiKeys.LIST_OFFSETS);
+ boolean hasFetchRequest = requestGenerated(client, ApiKeys.FETCH);
+ return hasListOffsetRequest && hasFetchRequest;
+ }, "No list-offset & fetch request sent");
// no error for no end offset (so unknown lag)
assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
// poll once again, which should return the list-offset response
// and hence next call would return correct lag result
- client.respond(listOffsetsResponse(singletonMap(tp0, 90L)));
+ ClientRequest listOffsetRequest = findRequest(client, ApiKeys.LIST_OFFSETS);
+ client.respondToRequest(listOffsetRequest, listOffsetsResponse(singletonMap(tp0, 90L)));
consumer.poll(Duration.ofMillis(0));
- assertEquals(OptionalLong.of(40L), consumer.currentLag(tp0));
+ // For AsyncKafkaConsumer, subscription sate is updated in background, so the result will eventually be updated.
+ TestUtils.waitForCondition(() -> {
+ OptionalLong result = consumer.currentLag(tp0);
+ return result.isPresent() && result.getAsLong() == 40L;
+ }, "Subscription state is not updated");
// requests: fetch
- assertEquals(1, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FETCH), "No fetch request sent");
// one successful fetch should update the log end offset and the position
+ ClientRequest fetchRequest = findRequest(client, ApiKeys.FETCH);
final FetchInfo fetchInfo = new FetchInfo(1L, 99L, 50L, 5);
- client.respond(fetchResponse(singletonMap(tp0, fetchInfo)));
+ client.respondToRequest(fetchRequest, fetchResponse(singletonMap(tp0, fetchInfo)));
final ConsumerRecords<String, String> records = (ConsumerRecords<String, String>) consumer.poll(Duration.ofMillis(1));
assertEquals(5, records.count());
assertEquals(55L, consumer.position(tp0));
// correct lag result
- assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
+ // For AsyncKafkaConsumer, subscription sate is updated in background, so the result will eventually be updated.
Review Comment:
typo state
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2486,68 +2492,96 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
// poll once to update with the current metadata
consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No metadata requests sent");
client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
// no error for no current position
assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
- assertEquals(0, client.inFlightRequestCount());
-
+ 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.
+ assertEquals(0, client.inFlightRequestCount());
+ }
// poll once again, which should send the list-offset request
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> {
+ boolean hasListOffsetRequest = requestGenerated(client, ApiKeys.LIST_OFFSETS);
+ boolean hasFetchRequest = requestGenerated(client, ApiKeys.FETCH);
+ return hasListOffsetRequest && hasFetchRequest;
+ }, "No list-offset & fetch request sent");
// no error for no end offset (so unknown lag)
assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
// poll once again, which should return the list-offset response
// and hence next call would return correct lag result
- client.respond(listOffsetsResponse(singletonMap(tp0, 90L)));
+ ClientRequest listOffsetRequest = findRequest(client, ApiKeys.LIST_OFFSETS);
+ client.respondToRequest(listOffsetRequest, listOffsetsResponse(singletonMap(tp0, 90L)));
consumer.poll(Duration.ofMillis(0));
- assertEquals(OptionalLong.of(40L), consumer.currentLag(tp0));
+ // For AsyncKafkaConsumer, subscription sate is updated in background, so the result will eventually be updated.
+ TestUtils.waitForCondition(() -> {
+ OptionalLong result = consumer.currentLag(tp0);
+ return result.isPresent() && result.getAsLong() == 40L;
+ }, "Subscription state is not updated");
// requests: fetch
- assertEquals(1, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FETCH), "No fetch request sent");
// one successful fetch should update the log end offset and the position
+ ClientRequest fetchRequest = findRequest(client, ApiKeys.FETCH);
final FetchInfo fetchInfo = new FetchInfo(1L, 99L, 50L, 5);
- client.respond(fetchResponse(singletonMap(tp0, fetchInfo)));
+ client.respondToRequest(fetchRequest, fetchResponse(singletonMap(tp0, fetchInfo)));
final ConsumerRecords<String, String> records = (ConsumerRecords<String, String>) consumer.poll(Duration.ofMillis(1));
assertEquals(5, records.count());
assertEquals(55L, consumer.position(tp0));
// correct lag result
- assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
+ // For AsyncKafkaConsumer, subscription sate is updated in background, so the result will eventually be updated.
+ TestUtils.waitForCondition(() -> {
Review Comment:
Is this change really needed? In this case we just did a successful fetch, so position is updated to 55. We should be able to retrieve the lag of 45 (end offsets is already known to be 100). (Is not exactly the same case as above, where we needed to allow for the ListOffsets response to be processed in the background). Makes sense?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1747085039
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2521,35 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
+ @EnumSource(GroupProtocol.class)
public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
consumer.seek(tp0, 50L);
- client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
+ client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
Review Comment:
@FrankYang0529 could you address this? I guess the suggestion comes from the recent fix to some tests that were flaky for a similar situation (https://github.com/apache/kafka/pull/170560)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1749547985
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2521,35 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
+ @EnumSource(GroupProtocol.class)
public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
consumer.seek(tp0, 50L);
- client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
+ client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
Review Comment:
Thanks for your explanations.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1754351703
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2479,23 +2487,29 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
Review Comment:
Hi @lianetm, thanks for catching this. I tried to add `Thread.sleep(1000)` before `assertEquals(0, client.inFlightRequestCount())`. It looks like both `ListOffsetsRequest` and `OffsetFetchRequest` will be sent after getting `FindCoordinatorResponse`. I remove the assertion now. Do you think that we should keep this assertion for CLASSIC group protocol only? Thank you.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #16982:
URL: https://github.com/apache/kafka/pull/16982#issuecomment-2340311968
Hey @FrankYang0529 , there are some related build failures on `testCurrentLag`, could you please take a look? Thanks!
https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-16982/7/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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1735186029
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2523,33 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
- public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
+ @EnumSource(GroupProtocol.class)
+ public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) throws InterruptedException {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
consumer.seek(tp0, 50L);
- client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
+ client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
assertEquals(singletonMap(tp0, 90L), consumer.endOffsets(Collections.singleton(tp0)));
// correct lag result should be returned as well
assertEquals(OptionalLong.of(40L), consumer.currentLag(tp0));
}
+ private Optional<ClientRequest> findRequest(MockClient client, ApiKeys apiKey) {
+ return client.requests().stream().filter(request -> request.requestBuilder().apiKey().equals(apiKey)).findFirst();
+ }
+
+ private boolean requestGenerated(MockClient client, ApiKeys apiKey) {
+ return client.requests().stream().anyMatch(request -> request.requestBuilder().apiKey().equals(apiKey));
Review Comment:
could we maybe reuse and simplify here to `return findRequest(client, apiKey).isPresent()`?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2523,33 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
- public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
+ @EnumSource(GroupProtocol.class)
+ public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) throws InterruptedException {
Review Comment:
I guess we don't need this throw Interrupted here anymore?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1731802528
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2523,16 +2537,32 @@ public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol)
// poll once to update with the current metadata
consumer.poll(Duration.ofMillis(0));
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
+ "No metadata requests sent");
client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
consumer.seek(tp0, 50L);
+
+ if (groupProtocol == GroupProtocol.CONSUMER) {
Review Comment:
I see how you make the test pass, and that is fine and works, but I believe we could greatly simplify this test if we get the FindCoord, OffsetFetch and FetchRequests out of the picture, which seem to me are really not relevant to this test?
This test is all about partition offsets (end offsets stored in the leader), compared to a position set manually with seek, so not related at all with committed offsets are affected by fetch.
So, this is the simplification I'm picturing:
1. we create the consumer without groupId -> this will ensure we don't send any OffsetFetch request. We don't really need them given that we're only playing with the partition offsets
2. we pause the partition right after assign -> this ensures that we don't issue fetch requests. We don't really need them for this test, and having them makes the test different for both consumers
With those 2 small changes, I would expect we can keep the same test for both consumer, without any specifics for the new consumer, and it would still be true to its purpose, testing what it has always tested. What do you think?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1733248177
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2481,23 +2489,31 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> {
+ boolean hasListOffsetRequest = requestGenerated(client, ApiKeys.LIST_OFFSETS);
+ boolean fetchRequest = requestGenerated(client, ApiKeys.FETCH);
+ return hasListOffsetRequest && fetchRequest;
+ }, "No list-offset & fetch request sent");
// no error for no end offset (so unknown lag)
assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
// poll once again, which should return the list-offset response
// and hence next call would return correct lag result
- client.respond(listOffsetsResponse(singletonMap(tp0, 90L)));
+ Optional<ClientRequest> listOffsetRequest = client.requests().stream().filter(request -> request.requestBuilder().apiKey().equals(ApiKeys.LIST_OFFSETS)).findFirst();
Review Comment:
this line seems tricky to read and we need it twice (for now). Would it be clearer to have a helper along the lines of `findRequest(client.requests(), ApiKeys.LIST_OFFSETS)`?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm merged PR #16982:
URL: https://github.com/apache/kafka/pull/16982
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1752557956
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2521,35 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
+ @EnumSource(GroupProtocol.class)
public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
consumer.seek(tp0, 50L);
- client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
+ client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
Review Comment:
makes sense, but I still have a concern with this (wonder if it would make this test flaky). Let's say the moment we get to prepare this response, there is another request that got generated, wouldn't that make that the endOffsets request will never find the response? (Not sure if I'm missing details of how prepareResponse works)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1736346494
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2523,33 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
- public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
+ @EnumSource(GroupProtocol.class)
+ public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) throws InterruptedException {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
consumer.seek(tp0, 50L);
- client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
+ client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
assertEquals(singletonMap(tp0, 90L), consumer.endOffsets(Collections.singleton(tp0)));
// correct lag result should be returned as well
assertEquals(OptionalLong.of(40L), consumer.currentLag(tp0));
}
+ private Optional<ClientRequest> findRequest(MockClient client, ApiKeys apiKey) {
+ return client.requests().stream().filter(request -> request.requestBuilder().apiKey().equals(apiKey)).findFirst();
+ }
+
+ private boolean requestGenerated(MockClient client, ApiKeys apiKey) {
+ return client.requests().stream().anyMatch(request -> request.requestBuilder().apiKey().equals(apiKey));
Review Comment:
I applied @kirktrue's suggestion. Thank you. 👍
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1735406032
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2523,33 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
- public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
+ @EnumSource(GroupProtocol.class)
+ public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) throws InterruptedException {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
Review Comment:
Out of curiosity, why do we no longer want to ensure a find coordinator request/response occurred?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2523,33 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
- public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
+ @EnumSource(GroupProtocol.class)
+ public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) throws InterruptedException {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
consumer.seek(tp0, 50L);
- client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
+ client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
assertEquals(singletonMap(tp0, 90L), consumer.endOffsets(Collections.singleton(tp0)));
// correct lag result should be returned as well
assertEquals(OptionalLong.of(40L), consumer.currentLag(tp0));
}
+ private Optional<ClientRequest> findRequest(MockClient client, ApiKeys apiKey) {
+ return client.requests().stream().filter(request -> request.requestBuilder().apiKey().equals(apiKey)).findFirst();
+ }
Review Comment:
It appears that the calls to this method quickly fail if there wasn't a match for the request type. How about something like this:
```suggestion
private ClientRequest findRequest(MockClient client, ApiKeys apiKey) {
Optional<ClientRequest> request = client.requests().stream().filter(request -> request.requestBuilder().apiKey().equals(apiKey)).findFirst();
assertTrue(request.isPresent(), "No " + apiKey + " request was submitted to the client... or whatever");
return request.get();
}
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1734259073
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2481,23 +2489,31 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> {
+ boolean hasListOffsetRequest = requestGenerated(client, ApiKeys.LIST_OFFSETS);
+ boolean fetchRequest = requestGenerated(client, ApiKeys.FETCH);
+ return hasListOffsetRequest && fetchRequest;
+ }, "No list-offset & fetch request sent");
// no error for no end offset (so unknown lag)
assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
// poll once again, which should return the list-offset response
// and hence next call would return correct lag result
- client.respond(listOffsetsResponse(singletonMap(tp0, 90L)));
+ Optional<ClientRequest> listOffsetRequest = client.requests().stream().filter(request -> request.requestBuilder().apiKey().equals(ApiKeys.LIST_OFFSETS)).findFirst();
+ assertTrue(listOffsetRequest.isPresent());
+ client.respondToRequest(listOffsetRequest.get(), listOffsetsResponse(singletonMap(tp0, 90L)));
consumer.poll(Duration.ofMillis(0));
assertEquals(OptionalLong.of(40L), consumer.currentLag(tp0));
// requests: fetch
- assertEquals(1, client.inFlightRequestCount());
+ TestUtils.waitForCondition(() -> requestGenerated(client, ApiKeys.FETCH), "No fetch request sent");
// one successful fetch should update the log end offset and the position
+ Optional<ClientRequest> fetchRequest = client.requests().stream().filter(request -> request.requestBuilder().apiKey().equals(ApiKeys.FETCH)).findFirst();
final FetchInfo fetchInfo = new FetchInfo(1L, 99L, 50L, 5);
- client.respond(fetchResponse(singletonMap(tp0, fetchInfo)));
+ assertTrue(fetchRequest.isPresent());
Review Comment:
Hi @lianetm, thanks for the review and suggestion. I addressed other comments. For this one, I would like to keep it. If we remove it, there may have warning message like `'Optional.get()' without 'isPresent()' check`.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1762153062
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2479,23 +2487,29 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
Review Comment:
With latest update, I run `I=0; while ./gradlew cleanTest :clients:test --tests KafkaConsumerTest --rerun --fail-fast; do (( I=$I+1 )); echo "Completed run: $I"; sleep 1; done` 100 times without error.
<img width="1311" alt="Screenshot 2024-09-17 at 9 26 24 AM" src="https://github.com/user-attachments/assets/01254142-3e0a-4d7f-b604-8639d24dbb91">
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #16982:
URL: https://github.com/apache/kafka/pull/16982#issuecomment-2311095871
Hey @FrankYang0529 , I took another look, left a comment for consideration. Thanks!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1736371829
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2523,33 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
- public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
+ @EnumSource(GroupProtocol.class)
+ public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) throws InterruptedException {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
Review Comment:
Like @lianetm's suggestion https://github.com/apache/kafka/pull/16982#discussion_r1731802528. In `testListOffsetShouldUpdateSubscriptions`, we want to check `endOffsets` function. It doesn't need group coordinator.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1739840322
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2507,32 +2521,35 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
assertEquals(OptionalLong.of(45L), consumer.currentLag(tp0));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
+ @EnumSource(GroupProtocol.class)
public void testListOffsetShouldUpdateSubscriptions(GroupProtocol groupProtocol) {
final ConsumerMetadata metadata = createMetadata(subscription);
final MockClient client = new MockClient(time, metadata);
initMetadata(client, singletonMap(topic, 1));
- consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
-
+ consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, false,
+ null, groupInstanceId, false);
consumer.assign(singleton(tp0));
-
- // poll once to update with the current metadata
- consumer.poll(Duration.ofMillis(0));
- client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));
-
consumer.seek(tp0, 50L);
- client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
+ client.prepareResponse(listOffsetsResponse(singletonMap(tp0, 90L)));
Review Comment:
Can `client.prepareResponse` be moved to `newConsumer` before?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #16982:
URL: https://github.com/apache/kafka/pull/16982#issuecomment-2351340156
Hi @lianetm, I think the test result is stable now. Could you take a look when you have time? Thank you.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1754814823
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2479,23 +2487,29 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
Review Comment:
Well definitely this bit is somewhere where the 2 consumers internasl are different, even though I see them conceptually aligned:
- what's the same in both consumers: a call to consumer lag will "signal" that it needs to retrieve the log endOffsets
- what's internally done different in both consumers: classic will only generate the request on the next poll (on the single thread it had and didn't want to block waiting for the offsets) vs async consumer, where the background thread poll will pick up the intention already expressed in the app thread and generate the request to get end offsets.
So I would say we keep the assertion (for the classic as you suggested), and it will be helpful to show this difference in the test actually. I would add an explanation for it too: 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. Makes sense?
With all these tests we're enabling, worth running them repeatedly locally to try to spot any other flakiness similar to the ones we've been catching.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #16982:
URL: https://github.com/apache/kafka/pull/16982#issuecomment-2340467295
Hi @lianetm, thanks for the reminder. Rebased it to see latest result.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1747073433
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -1053,12 +1053,20 @@ public void testFetchStableOffsetThrowInCommitted(GroupProtocol groupProtocol) {
assertThrows(UnsupportedVersionException.class, () -> setupThrowableConsumer(groupProtocol).committed(Collections.singleton(tp0)));
}
- // TODO: this test triggers a bug with the CONSUMER group protocol implementation.
- // The bug will be investigated and fixed so this test can use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
- public void testFetchStableOffsetThrowInPoll(GroupProtocol groupProtocol) {
- assertThrows(UnsupportedVersionException.class, () -> setupThrowableConsumer(groupProtocol).poll(Duration.ZERO));
+ @EnumSource(GroupProtocol.class)
+ public void testFetchStableOffsetThrowInPoll(GroupProtocol groupProtocol) throws InterruptedException {
+ setupThrowableConsumer(groupProtocol);
+ TestUtils.waitForCondition(() -> {
Review Comment:
Agree that it's better to discuss separately. Definitely we could consider to align the behaviour of the 2 consumers a bit more regarding poll with low timeouts and the guarantees of requests sent, but there are tradeoffs to consider as @chia7712 pointed out on his [comment](https://issues.apache.org/jira/browse/KAFKA-16792?focusedCommentId=17878298&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#comment-17878298) on the Jira (totally agree on the trade-offs).
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16792: Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #16982:
URL: https://github.com/apache/kafka/pull/16982#discussion_r1756698094
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2479,23 +2487,29 @@ public void testCurrentLag(GroupProtocol groupProtocol) {
consumer.seek(tp0, 50L);
consumer.poll(Duration.ofMillis(0));
// requests: list-offset, fetch
- assertEquals(2, client.inFlightRequestCount());
Review Comment:
Hi @lianetm, I added the assertion back and comments. I run 50 rounds on my laptop and there is no error. Let's check latest CI result. Thank you.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org