You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "TaiJuWu (via GitHub)" <gi...@apache.org> on 2024/07/19 22:38:57 UTC
[PR] KAFKA-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
TaiJuWu opened a new pull request, #16638:
URL: https://github.com/apache/kafka/pull/16638
*More detailed description of your change,
See [KAFKA-17163](https://issues.apache.org/jira/browse/KAFKA-17163)
*Summary of testing strategy (including rationale)
### 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-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16638:
URL: https://github.com/apache/kafka/pull/16638#discussion_r1685701571
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2965,7 +2976,21 @@ public void testSubscriptionOnInvalidTopic(GroupProtocol groupProtocol) {
KafkaConsumer<String, String> consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
consumer.subscribe(singleton(invalidTopicName), getConsumerRebalanceListener(consumer));
- assertThrows(InvalidTopicException.class, () -> consumer.poll(Duration.ZERO));
+ if (isAyncConsumer(groupProtocol)) {
+ // New consumer poll(ZERO) needs to wait for the event added by a call to poll, to be processed
+ // by the background thread, so it can realize there is invalid topics and then
+ // throw the InvalidTopicException
+ TestUtils.waitForCondition(() -> {
Review Comment:
`isAyncConsumer` is simple enough, and the method needing helper is `TestUtils.waitForCondition` :)
--
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-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #16638:
URL: https://github.com/apache/kafka/pull/16638#discussion_r1685477697
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2965,7 +2976,21 @@ public void testSubscriptionOnInvalidTopic(GroupProtocol groupProtocol) {
KafkaConsumer<String, String> consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
consumer.subscribe(singleton(invalidTopicName), getConsumerRebalanceListener(consumer));
- assertThrows(InvalidTopicException.class, () -> consumer.poll(Duration.ZERO));
+ if (groupProtocol == GroupProtocol.CONSUMER) {
Review Comment:
Could you please add a helper method for this 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-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16638:
URL: https://github.com/apache/kafka/pull/16638#discussion_r1686693862
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -3399,4 +3424,11 @@ public void configure(Map<String, ?> configs) {
CLIENT_IDS.add(configs.get(ConsumerConfig.CLIENT_ID_CONFIG).toString());
}
}
+
+ private Boolean isAyncConsumer(GroupProtocol groupProtocol) {
+ if (groupProtocol == GroupProtocol.CONSUMER) {
+ return true;
+ }
+ return false;
Review Comment:
```suggestion
return groupProtocol == GroupProtocol.CONSUMER);
```
--
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-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #16638:
URL: https://github.com/apache/kafka/pull/16638#discussion_r1685589970
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2965,7 +2976,21 @@ public void testSubscriptionOnInvalidTopic(GroupProtocol groupProtocol) {
KafkaConsumer<String, String> consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
consumer.subscribe(singleton(invalidTopicName), getConsumerRebalanceListener(consumer));
- assertThrows(InvalidTopicException.class, () -> consumer.poll(Duration.ZERO));
+ if (groupProtocol == GroupProtocol.CONSUMER) {
Review Comment:
Updated, thanks for review.
--
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-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #16638:
URL: https://github.com/apache/kafka/pull/16638#discussion_r1687892353
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2965,7 +2976,21 @@ public void testSubscriptionOnInvalidTopic(GroupProtocol groupProtocol) {
KafkaConsumer<String, String> consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
consumer.subscribe(singleton(invalidTopicName), getConsumerRebalanceListener(consumer));
- assertThrows(InvalidTopicException.class, () -> consumer.poll(Duration.ZERO));
+ if (isAyncConsumer(groupProtocol)) {
+ // New consumer poll(ZERO) needs to wait for the event added by a call to poll, to be processed
+ // by the background thread, so it can realize there is invalid topics and then
+ // throw the InvalidTopicException
+ TestUtils.waitForCondition(() -> {
Review Comment:
Sorry for misunderstanding. Rewrite 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-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16638:
URL: https://github.com/apache/kafka/pull/16638#discussion_r1686689029
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2093,14 +2093,27 @@ public void testEndOffsetsAuthenticationFailure(GroupProtocol groupProtocol) {
assertThrows(AuthenticationException.class, () -> consumer.endOffsets(Collections.singleton(tp0)));
}
- // TODO: this test requires rebalance logic which is not yet implemented in the CONSUMER group protocol.
- // Once it is implemented, this should use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
- public void testPollAuthenticationFailure(GroupProtocol groupProtocol) {
+ @EnumSource(GroupProtocol.class)
+ public void testPollAuthenticationFailure(GroupProtocol groupProtocol) throws InterruptedException {
final KafkaConsumer<String, String> consumer = consumerWithPendingAuthenticationError(groupProtocol);
consumer.subscribe(singleton(topic));
- assertThrows(AuthenticationException.class, () -> consumer.poll(Duration.ZERO));
+
+ if (isAyncConsumer(groupProtocol)) {
+ // New consumer poll(ZERO) needs to wait for the event added by a call to poll, to be processed
+ // by the background thread, so it can realize there is authentication fail and then
+ // throw the AuthenticationException
+ TestUtils.waitForCondition(() -> {
+ try {
+ consumer.poll(Duration.ZERO);
+ return false;
+ } catch (AuthenticationException e) {
+ return true;
+ }
+ }, "Consumer was not able to update fetch positions on continuous calls with 0 timeout");
Review Comment:
the message should be a bit different I would say. Even though it's all called from the `updateFechPositions`, the situation here is that we need to allow for the authentication exception (that is discovered in the background, when polling the `networkClientDelegate.maybePropagateMetadataError`), to appear in the app thread as an ErrorEvent (so that the AuthenticationException can be thrown on poll `processBackgroundEvents`). So maybe something like _Consumer was not able to discover metadata error on continous..._?
--
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-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16638:
URL: https://github.com/apache/kafka/pull/16638#discussion_r1686689029
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2093,14 +2093,27 @@ public void testEndOffsetsAuthenticationFailure(GroupProtocol groupProtocol) {
assertThrows(AuthenticationException.class, () -> consumer.endOffsets(Collections.singleton(tp0)));
}
- // TODO: this test requires rebalance logic which is not yet implemented in the CONSUMER group protocol.
- // Once it is implemented, this should use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
- public void testPollAuthenticationFailure(GroupProtocol groupProtocol) {
+ @EnumSource(GroupProtocol.class)
+ public void testPollAuthenticationFailure(GroupProtocol groupProtocol) throws InterruptedException {
final KafkaConsumer<String, String> consumer = consumerWithPendingAuthenticationError(groupProtocol);
consumer.subscribe(singleton(topic));
- assertThrows(AuthenticationException.class, () -> consumer.poll(Duration.ZERO));
+
+ if (isAyncConsumer(groupProtocol)) {
+ // New consumer poll(ZERO) needs to wait for the event added by a call to poll, to be processed
+ // by the background thread, so it can realize there is authentication fail and then
+ // throw the AuthenticationException
+ TestUtils.waitForCondition(() -> {
+ try {
+ consumer.poll(Duration.ZERO);
+ return false;
+ } catch (AuthenticationException e) {
+ return true;
+ }
+ }, "Consumer was not able to update fetch positions on continuous calls with 0 timeout");
Review Comment:
the message should be a bit different in this case I would say. Even though it's all called from the `updateFechPositions`, the situation here is that we need to allow for the authentication exception (that is discovered in the background, when polling the `networkClientDelegate.maybePropagateMetadataError`), to appear in the app thread as an ErrorEvent (so that the AuthenticationException can be thrown on poll `processBackgroundEvents`). So maybe something like _Consumer was not able to discover metadata errors on continuous..._?
--
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-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16638:
URL: https://github.com/apache/kafka/pull/16638#discussion_r1686692620
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2965,7 +2976,21 @@ public void testSubscriptionOnInvalidTopic(GroupProtocol groupProtocol) {
KafkaConsumer<String, String> consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
consumer.subscribe(singleton(invalidTopicName), getConsumerRebalanceListener(consumer));
- assertThrows(InvalidTopicException.class, () -> consumer.poll(Duration.ZERO));
+ if (isAyncConsumer(groupProtocol)) {
+ // New consumer poll(ZERO) needs to wait for the event added by a call to poll, to be processed
+ // by the background thread, so it can realize there is invalid topics and then
+ // throw the InvalidTopicException
+ TestUtils.waitForCondition(() -> {
Review Comment:
Agreed, what about a kind of `assertPollEventuallyThrows`(expectedException, errorMsg) (or better named). We would use it for `NoOffsetForPartitionException`, `InvalidTopicException` and `AuthenticationException`
--
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-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #16638:
URL: https://github.com/apache/kafka/pull/16638#discussion_r1687898103
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2965,7 +2970,28 @@ public void testSubscriptionOnInvalidTopic(GroupProtocol groupProtocol) {
KafkaConsumer<String, String> consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
consumer.subscribe(singleton(invalidTopicName), getConsumerRebalanceListener(consumer));
- assertThrows(InvalidTopicException.class, () -> consumer.poll(Duration.ZERO));
+ if (groupProtocol == GroupProtocol.CONSUMER) {
+ // New consumer poll(ZERO) needs to wait for the event added by a call to poll, to be processed
+ // by the background thread, so it can realize there is invalid topics and then
+ // throw the InvalidTopicException
+ assertPollEventuallyThrows(consumer, InvalidTopicException.class,
+ "Consumer was not able to update fetch positions on continuous calls with 0 timeout");
+ } else {
+ assertThrows(InvalidTopicException.class, () -> consumer.poll(Duration.ZERO));
+ }
+ }
+
+ private <T extends Throwable> void assertPollEventuallyThrows(KafkaConsumer<?, ?> consumer,
Review Comment:
It can be static function, update later.
--
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-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16638:
URL: https://github.com/apache/kafka/pull/16638#discussion_r1686692620
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2965,7 +2976,21 @@ public void testSubscriptionOnInvalidTopic(GroupProtocol groupProtocol) {
KafkaConsumer<String, String> consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
consumer.subscribe(singleton(invalidTopicName), getConsumerRebalanceListener(consumer));
- assertThrows(InvalidTopicException.class, () -> consumer.poll(Duration.ZERO));
+ if (isAyncConsumer(groupProtocol)) {
+ // New consumer poll(ZERO) needs to wait for the event added by a call to poll, to be processed
+ // by the background thread, so it can realize there is invalid topics and then
+ // throw the InvalidTopicException
+ TestUtils.waitForCondition(() -> {
Review Comment:
Agreed, what about a kind of `assertPollEventuallyThrows(expectedException, errorMsg)` (or better named). We would use it for `NoOffsetForPartitionException`, `InvalidTopicException` and `AuthenticationException`
--
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-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #16638:
URL: https://github.com/apache/kafka/pull/16638#issuecomment-2245651512
@TaiJuWu please use the new helper to rewrite https://github.com/apache/kafka/blob/c71eb60a3b2e30cc376b81144cf9c2c411d8f28b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java#L937
--
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-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16638:
URL: https://github.com/apache/kafka/pull/16638#discussion_r1686665604
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -862,7 +862,7 @@ public void testFetchProgressWithMissingPartitionPosition(GroupProtocol groupPro
MockClient client = new MockClient(time, metadata);
initMetadata(client, Collections.singletonMap(topic, 2));
- if (groupProtocol == GroupProtocol.CONSUMER) {
+ if (isAyncConsumer(groupProtocol)) {
Review Comment:
typo Async
--
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-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on PR #16638:
URL: https://github.com/apache/kafka/pull/16638#issuecomment-2246655210
> @TaiJuWu please use the new helper to rewrite
>
> https://github.com/apache/kafka/blob/c71eb60a3b2e30cc376b81144cf9c2c411d8f28b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java#L937
Done. Thanks review.
--
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-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16638:
URL: https://github.com/apache/kafka/pull/16638#discussion_r1686689029
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -2093,14 +2093,27 @@ public void testEndOffsetsAuthenticationFailure(GroupProtocol groupProtocol) {
assertThrows(AuthenticationException.class, () -> consumer.endOffsets(Collections.singleton(tp0)));
}
- // TODO: this test requires rebalance logic which is not yet implemented in the CONSUMER group protocol.
- // Once it is implemented, this should use both group protocols.
@ParameterizedTest
- @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
- public void testPollAuthenticationFailure(GroupProtocol groupProtocol) {
+ @EnumSource(GroupProtocol.class)
+ public void testPollAuthenticationFailure(GroupProtocol groupProtocol) throws InterruptedException {
final KafkaConsumer<String, String> consumer = consumerWithPendingAuthenticationError(groupProtocol);
consumer.subscribe(singleton(topic));
- assertThrows(AuthenticationException.class, () -> consumer.poll(Duration.ZERO));
+
+ if (isAyncConsumer(groupProtocol)) {
+ // New consumer poll(ZERO) needs to wait for the event added by a call to poll, to be processed
+ // by the background thread, so it can realize there is authentication fail and then
+ // throw the AuthenticationException
+ TestUtils.waitForCondition(() -> {
+ try {
+ consumer.poll(Duration.ZERO);
+ return false;
+ } catch (AuthenticationException e) {
+ return true;
+ }
+ }, "Consumer was not able to update fetch positions on continuous calls with 0 timeout");
Review Comment:
the message should be a bit different in this case I would say. Even though it's all called from the `updateFechPositions`, the situation here is that we need to allow for the authentication exception (that is discovered in the background, when polling the `networkClientDelegate.maybePropagateMetadataError`), to appear in the app thread as an ErrorEvent (so that the AuthenticationException can be thrown on poll `processBackgroundEvents`). So maybe something like _Consumer was not able to discover metadata error on continous..._?
--
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-17163:revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #16638:
URL: https://github.com/apache/kafka/pull/16638
--
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