KAFKA-14761 Adding integration test for the prototype consumer #13303

philipnee · 2023-02-24T23:31:51Z

The goal of this PR is to add more tests to the PrototypeAsyncConsumer to test

Successful startup and shutdown.
Commit.

I also added integration tests:

Test commitAsync()
Test commitSync()
a. Note that I still need to implement committed() to test if commitSync() has been successfully committed.

Additional things:

Change KafkaConsumer<K, V> to Consumer<K, V> to use different implementations

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…philipnee/kafka into consumer-refactor-prototype-integration

philipnee · 2023-03-05T06:37:45Z

core/src/test/scala/integration/kafka/api/AbstractConsumerTest.scala

@@ -97,7 +97,7 @@ abstract class AbstractConsumerTest extends BaseRequestTest {
    }
  }

-  protected def createConsumerWithGroupId(groupId: String): KafkaConsumer[Array[Byte], Array[Byte]] = {


The only thing here was to change the type of Consumer<K, V>

philipnee · 2023-03-05T06:37:56Z

core/src/test/scala/integration/kafka/api/ConsumerBounceTest.scala

@@ -306,7 +306,7 @@ class ConsumerBounceTest extends AbstractConsumerTest with Logging {
    this.consumerConfig.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
    val partitions = createTopicPartitions(topic, numPartitions = partitionCount, replicationFactor = brokerCount)

-    addConsumersToGroupAndWaitForGroupAssignment(consumerCount, mutable.Buffer[KafkaConsumer[Array[Byte], Array[Byte]]](),


Same, change to Consumer<K, V>

philipnee · 2023-03-05T06:38:29Z

core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala

@@ -993,7 +993,7 @@ class PlaintextConsumerTest extends BaseConsumerTest {

    // subscribe all consumers to all topics and validate the assignment

-    val consumersInGroup = Buffer[KafkaConsumer[Array[Byte], Array[Byte]]]()


Same, make it Consumer<K, V>

philipnee · 2023-03-05T06:42:17Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

        });
+    }
+
+    private CompletableFuture<Void> commit(Map<TopicPartition, OffsetAndMetadata> offsets) {


I have a KIP proposal here:

WDYT about asyncCommit returning a CompletableFuture?

WDYT about poll returning CompletableFuture?

Aren't these methods part of the existing Consumer interface? Won't they be breaking changes?

Thanks, for 1 - I think I actually wanted to say commitSync(), sorry about that.

I think they aren't part of the existing interface, right? Currently, both commitSync and poll don't return a future, and I was wondering if it makes sense for us to do that. Also, it makes the interface feels more modern.

I thought we have those methods on the existing Consumer interface. Anyway, that doesn't impact this PR.

philipnee · 2023-03-05T16:30:07Z

Tests appear to be flaky:

Build / JDK 11 and Scala 2.13 / testMultiWorkerRestartOnlyConnector – org.apache.kafka.connect.integration.ConnectorRestartApiIntegrationTest
2m 24s
Build / JDK 11 and Scala 2.13 / testElectUncleanLeadersForManyPartitions(String).quorum=kraft – kafka.api.PlaintextAdminIntegrationTest
22s
Build / JDK 17 and Scala 2.13 / shouldThrowIllegalArgumentExceptionWhenCustomPartitionerReturnsMultiplePartitions() – org.apache.kafka.streams.integration.KTableKTableForeignKeyInnerJoinCustomPartitionerIntegrationTest
1m 43s
Build / JDK 17 and Scala 2.13 / shouldRestoreState() – org.apache.kafka.streams.integration.PurgeRepartitionTopicIntegrationTest
1m 4s
Fixed 126

rajinisivaram

@philipnee Thanks for the PR, looks good. Left some questions.

rajinisivaram · 2023-03-06T20:12:52Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

                                  final Deserializer<K> keyDeserializer,
                                  final Deserializer<V> valueDeserializer) {
-        this.time = time;
+        this.time = Time.SYSTEM;


Why did we remove time? Could be useful for unit testing?

Thanks for spotting this. I think the unit test consumer is meant to be the one below this. I think I should remove the time and make the constructor param list the same as the current one.

I think as we go along with this, we would need to add this param back to the constructor :) ANyways that's for future PRs then.

rajinisivaram · 2023-03-06T20:15:51Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

        });
+    }
+
+    private CompletableFuture<Void> commit(Map<TopicPartition, OffsetAndMetadata> offsets) {


Aren't these methods part of the existing Consumer interface? Won't they be breaking changes?

rajinisivaram · 2023-03-06T20:17:24Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

+        CompletableFuture<Void> future = commit(offsets);
+        future.whenComplete((r, t) -> {
+            if (t != null) {
+                callback.onComplete(offsets, new RuntimeException(t));


Any reason why this is a RuntimeException rather than a KafkaException?

You are right, it should've been a KafkaException.

rajinisivaram · 2023-03-06T20:57:22Z

...ts/src/test/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumerTest.java

-                new NoOpConsumerRebalanceListener());
-        assertEquals(1, consumer.subscription().size());
+    public void testBackgroundThreadRunning() {
+        consumer = newConsumer(time, new StringDeserializer(), new StringDeserializer());


Don't we need to add some verification since the test name suggests we are testing if the background thread was started?

I'm thinking what's the best way to test that. As we don't have direct access to the background thread from the consumer. I guess we could do that passively by trying to catch the exception, or maybe, actively, I should add a public State state() to the eventHandler? to allow user to prob the background thread stat... I feel we need both, but WDYT? @rajinisivaram @guozhangwang

maybe we don't need this afterall actually. I feel for the purpose of unit test, we should just try to test the API and method calls. The integration test in this PR is already kind of testing this (otherwise request won't come through)

rajinisivaram · 2023-03-06T20:59:37Z

...ts/src/test/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumerTest.java

+    }
+
+    private void injectConsumerConfigs() {
+        consumerProps.put(BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");


I guess this test currently attempts to connect to this? Is the intention to convert this into a unit test with mocks or only use it for tests where successful connections are not required?

I think that's the intent: I don't think it is necessary to make the actual connection to the broker on the level of unit test.

For the longer term, it will be good to use a MockClient rather than one that attempts to connect and fills the log with exceptions. For now, this seems reasonable.

philipnee · 2023-03-06T21:23:10Z

Thanks for reviewing this, @rajinisivaram - fixes are on the way.

guozhangwang

re: Can we commit? We added two tests: I only saw one inside BaseAsyncConsumerTest? re: Unit tests in PrototypeConsumerTest: you mean PrototypeAsyncConsumerTest` right?

BTW could we also test fetch now with the pool method?

guozhangwang · 2023-03-07T00:06:01Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java

@@ -213,7 +215,7 @@ public UnsentRequest(
            Objects.requireNonNull(requestBuilder);
            this.requestBuilder = requestBuilder;
            this.node = node;
-            this.callback = new FutureCompletionHandler();


nit: just call the member field handler as well?

guozhangwang · 2023-03-07T00:08:20Z

...ts/src/test/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumerTest.java

-                new NoOpConsumerRebalanceListener());
-        assertEquals(1, consumer.subscription().size());
+    public void testBackgroundThreadRunning() {
+        consumer = newConsumer(time, new StringDeserializer(), new StringDeserializer());


guozhangwang · 2023-03-07T00:10:40Z

core/src/test/scala/integration/kafka/api/BaseAsyncConsumerTest.scala

+    waitUntilTrue(() => {
+      cb.successCount == 1
+    }, "wait until commit is completed successfully", 5000)
+    consumer.commitSync();


Why call commitSync here again? If the goal is to make sure the second commit also goes through shall we verify that on the broker's side?

+1 : although i don't know how to verify that but let me look into it. thanks!

I think the idea was to test both commitSync and commitAsync.

Maybe I'll break it up into two tests there. It doesn't really make sense to call commit twice in a row.

Other integration tests test the commit with committed method call. I'll leave this as TODO.

philipnee · 2023-03-07T01:41:08Z

re. @guozhangwang - sorry, I think originally I added 2 tests, then I reduced to 1. Thanks for catching this.

For testing poll() - The fetcher isn't available but I could try to test poll?

Thanks!

rajinisivaram

@philipnee Thanks for the updates, LGTM. Looks like @guozhangwang 's comments have been addressed as well, but he may want to review again.

rajinisivaram · 2023-03-07T19:22:59Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

        });
+    }
+
+    private CompletableFuture<Void> commit(Map<TopicPartition, OffsetAndMetadata> offsets) {


I thought we have those methods on the existing Consumer interface. Anyway, that doesn't impact this PR.

rajinisivaram · 2023-03-07T19:25:38Z

...ts/src/test/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumerTest.java

+    }
+
+    private void injectConsumerConfigs() {
+        consumerProps.put(BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");


For the longer term, it will be good to use a MockClient rather than one that attempts to connect and fills the log with exceptions. For now, this seems reasonable.

philipnee · 2023-03-07T19:43:44Z

Hey @rajinisivaram - Thanks! To your comments
I thought we have those methods on the existing Consumer interface. Anyway, that doesn't impact this PR. : The current consumer uses kafka specific future, but in the new re-write, we are kind of migrating to java CompletableFuture

philipnee · 2023-03-07T22:39:39Z

@guozhangwang - Regarding testing the commit results, can we defer this to the subsequent PR, i.e. verify it once committed() is implemented.

guozhangwang · 2023-03-08T00:12:56Z

@guozhangwang - Regarding testing the commit results, can we defer this to the subsequent PR, i.e. verify it once committed() is implemented.

SGTM.

philipnee added 9 commits February 24, 2023 15:25

KAFKA-14761 integration testing

85f8c58

fix

709ba9d

Merge branch 'consumer-refactor-prototype-integration' of github.com:…

575aad9

…philipnee/kafka into consumer-refactor-prototype-integration

Merge branch 'apache:trunk' into consumer-refactor-prototype-integration

5707df3

change it test type

79d0b2b

merge tests

e465f3d

inject bootstrap url to each test

0406c8a

fixes

726154c

Integration test

eab2bc1

philipnee commented Mar 5, 2023

View reviewed changes

philipnee marked this pull request as ready for review March 5, 2023 06:38

philipnee commented Mar 5, 2023

View reviewed changes

appropriate naming

4aff737

rajinisivaram reviewed Mar 6, 2023

View reviewed changes

guozhangwang reviewed Mar 7, 2023

View reviewed changes

philipnee added 4 commits March 6, 2023 18:17

Address the comments

7844988

commit unit test

b8d9fe4

clean up

c8118e1

address comment

8145105

rajinisivaram approved these changes Mar 7, 2023

View reviewed changes

guozhangwang approved these changes Mar 8, 2023

View reviewed changes

guozhangwang merged commit 6fbe4d8 into apache:trunk Mar 8, 2023

philipnee added KIP-848 ctr labels May 23, 2023

		@@ -993,7 +993,7 @@ class PlaintextConsumerTest extends BaseConsumerTest {

		// subscribe all consumers to all topics and validate the assignment

		val consumersInGroup = Buffer[KafkaConsumer[Array[Byte], Array[Byte]]]()

KAFKA-14761 Adding integration test for the prototype consumer #13303

KAFKA-14761 Adding integration test for the prototype consumer #13303

Conversation

philipnee commented Feb 24, 2023 • edited

Committer Checklist (excluded from commit message)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

philipnee Mar 5, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

philipnee commented Mar 5, 2023

rajinisivaram left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

philipnee commented Mar 6, 2023

guozhangwang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

philipnee commented Mar 7, 2023

rajinisivaram left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

philipnee commented Mar 7, 2023

philipnee commented Mar 7, 2023

guozhangwang commented Mar 8, 2023

philipnee commented Feb 24, 2023 •

edited

philipnee Mar 5, 2023 •

edited