You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/07/24 08:40:41 UTC
(kafka) branch trunk updated: KAFKA-17163 revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure (#16638)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new f9f480ac7c1 KAFKA-17163 revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure (#16638)
f9f480ac7c1 is described below
commit f9f480ac7c1dc94e34e2d47d3259830400db6d7b
Author: TaiJuWu <tj...@gmail.com>
AuthorDate: Wed Jul 24 16:40:35 2024 +0800
KAFKA-17163 revisit testSubscriptionOnInvalidTopic and testPollAuthenticationFailure (#16638)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../kafka/clients/consumer/KafkaConsumerTest.java | 55 +++++++++++++++-------
1 file changed, 37 insertions(+), 18 deletions(-)
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
index dac592801d9..33e4a26389d 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
@@ -938,14 +938,8 @@ public class KafkaConsumerTest {
// New consumer poll(ZERO) needs to wait for the offset fetch event added by a call to poll, to be processed
// by the background thread, so it can realize there are no committed offsets and then
// throw the NoOffsetForPartitionException
- TestUtils.waitForCondition(() -> {
- try {
- consumer.poll(Duration.ZERO);
- return false;
- } catch (NoOffsetForPartitionException e) {
- return true;
- }
- }, "Consumer was not able to update fetch positions on continuous calls with 0 timeout");
+ assertPollEventuallyThrows(consumer, NoOffsetForPartitionException.class,
+ "Consumer was not able to update fetch positions on continuous calls with 0 timeout");
} else {
assertThrows(NoOffsetForPartitionException.class, () -> consumer.poll(Duration.ZERO));
}
@@ -2093,14 +2087,21 @@ public class KafkaConsumerTest {
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 (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 authentication fail and then
+ // throw the AuthenticationException
+ assertPollEventuallyThrows(consumer, AuthenticationException.class,
+ "he consumer was not able to discover metadata errors during continuous polling.");
+ } else {
+ assertThrows(AuthenticationException.class, () -> consumer.poll(Duration.ZERO));
+ }
}
// TODO: this test triggers a bug with the CONSUMER group protocol implementation.
@@ -2940,11 +2941,9 @@ public class KafkaConsumerTest {
}
}
- // 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 testSubscriptionOnInvalidTopic(GroupProtocol groupProtocol) {
+ @EnumSource(GroupProtocol.class)
+ public void testSubscriptionOnInvalidTopic(GroupProtocol groupProtocol) throws InterruptedException {
ConsumerMetadata metadata = createMetadata(subscription);
MockClient client = new MockClient(time, metadata);
@@ -2965,7 +2964,27 @@ public class KafkaConsumerTest {
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 static <T extends Throwable> void assertPollEventuallyThrows(KafkaConsumer<?, ?> consumer,
+ Class<T> expectedException, String errMsg) throws InterruptedException {
+ TestUtils.waitForCondition(() -> {
+ try {
+ consumer.poll(Duration.ZERO);
+ return false;
+ } catch (Throwable exception) {
+ return expectedException.isInstance(exception);
+ }
+ }, errMsg);
}
@ParameterizedTest