KAFKA-18181: Refactor ShareConsumerTest #18105

brandboat · 2024-12-09T14:02:15Z

related to https://issues.apache.org/jira/browse/KAFKA-18181

consumeMessages should return messagesConsumed instead of wrapping it in CompletableFuture
replace fail by assertDoesNotThrow or just throw exception
produceMessagesshould return messagesConsumed instead of wrapping it in CompletableFuture
use CompletableFuture.runAsync() instead of ExecutorService. We don't use the interrupt so it is fine to use default executor

Committer Checklist (excluded from commit message)

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

apoorvmittal10

Thanks for the PR, it's a good refactor. I have one doubt on the lost exception details with assertions. Can we somehow capture the exception as well?

apoorvmittal10 · 2024-12-09T22:39:03Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

-            } catch (Exception e) {
-                fail("Failed to send records: " + e);
+            for (int i = 0; i < 10; i++) {
+                assertDoesNotThrow(() -> producer.send(record).get(), "Failed to send records");


Will it be possible to capture the exception thrown as well so it's easy to find what went wrong. It's hard to find that with the current custom error message?

Thanks for the comment @apoorvmittal10, I did a quick test to check if assertDoesNotThrow captures the exception details.

Here’s the test I wrote:

@Test public void testT() { assertDoesNotThrow(() -> { throw new RuntimeException("foobar"); }, "Failed to send records"); }

When I run this, the output includes the custom error message, the thrown exception, and its stack trace:

Failed to send records ==> Unexpected exception thrown: java.lang.RuntimeException: foobar org.opentest4j.AssertionFailedError: Failed to send records ==> Unexpected exception thrown: java.lang.RuntimeException: foobar at org.junit.jupiter.api.AssertionFailureBuilder.build(AssertionFailureBuilder.java:152) at org.junit.jupiter.api.AssertDoesNotThrow.createAssertionFailedError(AssertDoesNotThrow.java:84) at org.junit.jupiter.api.AssertDoesNotThrow.assertDoesNotThrow(AssertDoesNotThrow.java:75) at org.junit.jupiter.api.AssertDoesNotThrow.assertDoesNotThrow(AssertDoesNotThrow.java:62) at org.junit.jupiter.api.Assertions.assertDoesNotThrow(Assertions.java:3249) at kafka.server.logger.RuntimeLoggerManagerTest.testT(RuntimeLoggerManagerTest.java:46) at java.base/java.lang.reflect.Method.invoke(Method.java:580) at java.base/java.util.ArrayList.forEach(ArrayList.java:1596) at java.base/java.util.ArrayList.forEach(ArrayList.java:1596) Caused by: java.lang.RuntimeException: foobar at kafka.server.logger.RuntimeLoggerManagerTest.lambda$testT$0(RuntimeLoggerManagerTest.java:47) at org.junit.jupiter.api.AssertDoesNotThrow.assertDoesNotThrow(AssertDoesNotThrow.java:71) ... 6 more

Is this the level of detail you were hoping to see, or is there something additional you'd like? Let me know!

Yeah this looks fine to me, as once we include the custom error message, we get both the stack trace of the exception and the custom message that we defined.

Thanks for looking into, this is good enough.

ShivsundarR · 2024-12-10T12:19:07Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

+        assertDoesNotThrow(() -> {
+            try (Admin admin = Admin.create(props)) {
+                admin.createTopics(Collections.singleton(new NewTopic(topicName, 1, (short) 1))).all().get();
+            }


Can we have the custom error message added here as well("Failed to create topic"). Like in other parts of the code, we could maybe add that as the second argument for assertDoesNotThrow()?

ShivsundarR · 2024-12-10T12:22:05Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

-                    fail("Failed to send record: " + e);
-                }
+                final int index = i;
+                assertDoesNotThrow(() -> recordFutures[index].get());


Here, can we add the custom error message("Failed to send record") to the assert as well.

ShivsundarR · 2024-12-10T12:24:52Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

+            try (KafkaShareConsumer<byte[], byte[]> shareConsumer = createShareConsumer(
+                    new ByteArrayDeserializer(), new ByteArrayDeserializer(), groupId)) {
+                shareConsumer.subscribe(Collections.singleton(tp.topic()));
+                return consumeMessages(shareConsumer, totalMessagesConsumed, totalMessages, consumerNumber, maxPolls, commit);


Here as well, can we add the custom error message with consumer number. There are a couple of more places where we do this, maybe we can change those places as well to include the custom error message.

Sorry for the late reply, I added error message to all assertDoesNotThrow methods in this test. Thanks!

ShivsundarR

Hi @brandboat , Thanks for the PR!. Just a few minor comments w.r.t the assertDoesNotThrow() changes.

chia7712

@brandboat thanks for this patch

chia7712 · 2024-12-11T22:48:15Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

-                    fail("Failed to send record: " + e);
-                }
+                final int index = i;
+                assertDoesNotThrow(() -> recordFutures[index].get(), "Failed to send record");


it seems we expect all messages should be succeed, and hence we can rewrite this function:

private int produceMessages(int messageCount) { try (KafkaProducer<byte[], byte[]> producer = createProducer(new ByteArraySerializer(), new ByteArraySerializer())) { ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, "key".getBytes(), "value".getBytes()); IntStream.range(0, messageCount).forEach(__ -> producer.send(record)); producer.flush(); } return messageCount; }

chia7712 · 2024-12-11T22:52:55Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

+        CompletableFuture.allOf(produceMessageFutures.toArray(CompletableFuture[]::new)).get(60, TimeUnit.SECONDS);
+        CompletableFuture.allOf(consumeMessagesFutures.toArray(CompletableFuture[]::new)).get(60, TimeUnit.SECONDS);
+
+        int totalSuccessResult = consumeMessagesFutures.stream().map(CompletableFuture::join).reduce(Integer::sum).orElse(0);


int totalSuccessResult = consumeMessagesFutures.stream().mapToInt(CompletableFuture::join).sum();

chia7712 · 2024-12-11T22:55:23Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

-            int actualMessagesSent = 0;
-            try {
-                producerExecutorService.awaitTermination(60, TimeUnit.SECONDS); // Wait for all producer threads to complete
+        int totalResult1 = consumeMessagesFutures1.stream().map(CompletableFuture::join).reduce(Integer::sum).orElse(0);


int totalResult1 = consumeMessagesFutures1.stream().mapToInt(CompletableFuture::join).sum(); int totalResult2 = consumeMessagesFutures2.stream().mapToInt(CompletableFuture::join).sum(); int totalResult3 = consumeMessagesFutures3.stream().mapToInt(CompletableFuture::join).sum();

chia7712 · 2024-12-11T22:56:42Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java

-        } catch (Exception e) {
-            fail("Exception occurred : " + e.getMessage());
-        }
+        assertEquals(totalMessagesSent, totalMessagesConsumedGroup1.get());


Do we actually need totalMessagesConsumedGroup1, or can we simply count the returned values from all consumeMessages instead?

Perhaps we can remove all AtomicInteger here, like you said, now we can rely on the returned values to verify how much records are consumed. WDYT ?

Update: I mean the AtomicInteger in consumeMessages.

yes, please remove it to simplify code :)

I just realized that we cannot remove the AtomicInteger in consumeMessages because it is essential for managing the total number of messages consumed across multiple calls to consumeMessages.

Without it, many tests would hang since the condition totalMessagesConsumed.get() < totalMessages would no longer be met, causing the while loop to run indefinitely [0].

[0] https://github.com/apache/kafka/blob/trunk/core/src/test/java/kafka/test/api/ShareConsumerTest.java#L1888

But you are right, the atomic integer assertions seems a bit redundant as now we can rely on the return value from consumeMessages. I'll remove those assertions, thanks.

apoorvmittal10

Thanks for the refactor, LGTM!

apoorvmittal10 · 2024-12-13T17:44:23Z

core/src/test/java/kafka/test/api/ShareConsumerTest.java


+        int maxBytes = 100000;
        for (int i = 0; i < consumerCount; i++) {


Should the variable be consumerCount => consumerGroupCount ?

Thanks for the comment ! I think consumerGroupCount might still be confusing since there are three consumer groups in this test. Instead of renaming, perhaps we could remove the consumerCount variable and inlining its value directly into the for-loop.

Reviewers: Apoorv Mittal <[email protected]>, Chia-Ping Tsai <[email protected]>

KAFKA-18181: Refactor ShareConsumerTest

7380dc9

github-actions bot added core Kafka Broker tests Test fixes (including flaky tests) labels Dec 9, 2024

AndrewJSchofield added the KIP-932 Queues for Kafka label Dec 9, 2024

apoorvmittal10 reviewed Dec 9, 2024

View reviewed changes

ShivsundarR reviewed Dec 10, 2024

View reviewed changes

ShivsundarR suggested changes Dec 10, 2024

View reviewed changes

brandboat added 2 commits December 11, 2024 22:04

Address comments

00a2808

Merge remote-tracking branch 'origin/trunk' into KAFKA-18181

ae24ff5

chia7712 reviewed Dec 11, 2024

View reviewed changes

brandboat added 3 commits December 12, 2024 20:08

Address comment

9f279f8

Address comments

61cf095

Merge remote-tracking branch 'origin/trunk' into KAFKA-18181

d47f11c

apoorvmittal10 approved these changes Dec 13, 2024

View reviewed changes

brandboat added 2 commits December 14, 2024 02:05

Address comment

65925d1

Merge remote-tracking branch 'origin/trunk' into KAFKA-18181

a9fce4d

chia7712 approved these changes Dec 14, 2024

View reviewed changes

chia7712 merged commit 9e60fcc into apache:trunk Dec 14, 2024
10 checks passed

frankvicky mentioned this pull request Dec 14, 2024

MINOR: Clean leaking resource in ShareConsumerTest #18073

Closed

3 tasks

brandboat deleted the KAFKA-18181 branch December 14, 2024 03:37

tedyu pushed a commit to tedyu/kafka that referenced this pull request Jan 6, 2025

KAFKA-18181 Refactor ShareConsumerTest (apache#18105)

1ddbdda

Reviewers: Apoorv Mittal <[email protected]>, Chia-Ping Tsai <[email protected]>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-18181: Refactor ShareConsumerTest #18105

KAFKA-18181: Refactor ShareConsumerTest #18105

brandboat commented Dec 9, 2024

apoorvmittal10 left a comment

apoorvmittal10 Dec 9, 2024

brandboat Dec 10, 2024

ShivsundarR Dec 10, 2024

apoorvmittal10 Dec 10, 2024

ShivsundarR Dec 10, 2024 •

edited

Loading

ShivsundarR Dec 10, 2024

ShivsundarR Dec 10, 2024 •

edited

Loading

brandboat Dec 11, 2024

ShivsundarR left a comment

chia7712 left a comment

chia7712 Dec 11, 2024

chia7712 Dec 11, 2024

chia7712 Dec 11, 2024

chia7712 Dec 11, 2024

brandboat Dec 12, 2024 •

edited

Loading

chia7712 Dec 12, 2024

brandboat Dec 13, 2024

apoorvmittal10 left a comment

apoorvmittal10 Dec 13, 2024

brandboat Dec 13, 2024


		int maxBytes = 100000;
		for (int i = 0; i < consumerCount; i++) {

KAFKA-18181: Refactor ShareConsumerTest #18105

KAFKA-18181: Refactor ShareConsumerTest #18105

Conversation

brandboat commented Dec 9, 2024

Committer Checklist (excluded from commit message)

apoorvmittal10 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

ShivsundarR Dec 10, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ShivsundarR Dec 10, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ShivsundarR left a comment

Choose a reason for hiding this comment

chia7712 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

brandboat Dec 12, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apoorvmittal10 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ShivsundarR Dec 10, 2024 •

edited

Loading

ShivsundarR Dec 10, 2024 •

edited

Loading

brandboat Dec 12, 2024 •

edited

Loading