KAFKA-19042: [10/N] Move PlaintextConsumerAssignTest to clients-integration-tests module #19773
+327
−203
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
github-actions
bot
added
triage
Sorry, that is an accident when I merge trunk. |
brandboat
reviewed
May 26, 2025
| @@ -0,0 +1,4 @@ | |||
| package org.apache.kafka.clients.consumer; | |||
There was a problem hiding this comment.
Done. Thanks for checking.
brandboat
reviewed
May 26, 2025
Comment on lines
64
to
68
| String topic = "topic"; | ||
| int partition = 0; | ||
| int numPartitions = 3; | ||
| short numReplica = 3; | ||
| TopicPartition tp = new TopicPartition(topic, partition); |
There was a problem hiding this comment.
nit
Suggested change
| String topic = "topic"; | |
| int partition = 0; | |
| int numPartitions = 3; | |
| short numReplica = 3; | |
| TopicPartition tp = new TopicPartition(topic, partition); | |
| private final String topic = "topic"; | |
| private final int partition = 0; | |
| private final int numPartitions = 3; | |
| private final short numReplica = 3; | |
| private final TopicPartition tp = new TopicPartition(topic, partition); |
There was a problem hiding this comment.
TopicPartition tp = new TopicPartition(topic, partition); this line is missed
| } | ||
|
|
||
| @ClusterTest | ||
| void testClassicAssignAndCommitAsyncNotCommitted() throws Exception { |
There was a problem hiding this comment.
I know it might not be critical in test files, but could we add public to all test methods just to keep things consistent?
| } | ||
|
|
||
|
|
||
| void testAssignAndCommitAsyncNotCommitted(String groupProtocol) throws InterruptedException { |
| CountConsumerCommitCallback cb = new CountConsumerCommitCallback(); | ||
|
|
||
| try (Consumer<byte[], byte[]> consumer = clusterInstance.consumer(Map.of(GROUP_PROTOCOL_CONFIG, groupProtocol))) { | ||
| ClientsTestUtils.sendRecords(clusterInstance, tp, numRecords, startingTimestamp, -1); |
There was a problem hiding this comment.
Suggested change
| ClientsTestUtils.sendRecords(clusterInstance, tp, numRecords, startingTimestamp, -1); | |
| ClientsTestUtils.sendRecords(clusterInstance, tp, numRecords); |
|
|
||
| try (Consumer<byte[], byte[]> consumer = clusterInstance.consumer(Map.of(GROUP_PROTOCOL_CONFIG, groupProtocol))) { | ||
| long startingTimestamp = System.currentTimeMillis(); | ||
| ClientsTestUtils.sendRecords(clusterInstance, tp, numRecords, startingTimestamp, -1); |
There was a problem hiding this comment.
Suggested change
| ClientsTestUtils.sendRecords(clusterInstance, tp, numRecords, startingTimestamp, -1); | |
| ClientsTestUtils.sendRecords(clusterInstance, tp, numRecords, startingTimestamp); |
|
|
||
| @BeforeEach | ||
| public void setup() throws InterruptedException { | ||
| clusterInstance.createTopic(topic, numPartitions, numReplica); |
There was a problem hiding this comment.
The original test has replica count is 2, let's keep it.
Suggested change
| clusterInstance.createTopic(topic, numPartitions, numReplica); | |
| clusterInstance.createTopic(topic, 2, BROKER_COUNT); |
| types = {Type.KRAFT}, | ||
| brokers = PlaintextConsumerAssignTest.BROKER_COUNT, | ||
| serverProperties = { | ||
| @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"), |
There was a problem hiding this comment.
The original replica factor is 3
Suggested change
| @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"), | |
| @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "3"), |
| void testAsyncAssignAndCommitSyncNotCommitted() throws Exception { | ||
| testAssignAndCommitSyncNotCommitted(GroupProtocol.CONSUMER.name); | ||
| } | ||
| void testAssignAndCommitSyncNotCommitted(String groupProtocol) throws InterruptedException { |
There was a problem hiding this comment.
Suggested change
| void testAssignAndCommitSyncNotCommitted(String groupProtocol) throws InterruptedException { | |
| private void testAssignAndCommitSyncNotCommitted(String groupProtocol) { |
| } | ||
|
|
||
| @ClusterTest | ||
| void testClassicAssignAndCommitSyncNotCommitted() throws Exception { |
There was a problem hiding this comment.
Suggested change
| void testClassicAssignAndCommitSyncNotCommitted() throws Exception { | |
| public void testClassicAssignAndCommitSyncNotCommitted() { |
| } | ||
|
|
||
| @ClusterTest | ||
| void testAsyncAssignAndCommitSyncNotCommitted() throws Exception { |
There was a problem hiding this comment.
Suggested change
| void testAsyncAssignAndCommitSyncNotCommitted() throws Exception { | |
| public void testAsyncAssignAndCommitSyncNotCommitted() { |
|
Hi @brandboat , thanks for your detailed review and point out my mistakes, all comments are addressed, PTAL. |
brandboat
reviewed
May 27, 2025
|
|
||
| @BeforeEach | ||
| public void setup() throws InterruptedException { | ||
| clusterInstance.createTopic(topic, BROKER_COUNT, (short) 2); |
There was a problem hiding this comment.
I think the BROKER_COUNT should be replica count, right?
Suggested change
| clusterInstance.createTopic(topic, BROKER_COUNT, (short) 2); | |
| clusterInstance.createTopic(topic, 2, (short) BROKER_COUNT); |
…ka/clients/consumer/PlaintextConsumerAssignTest.java Co-authored-by: Kuan-Po Tseng <[email protected]>
m1a2st
reviewed
May 27, 2025
Comment on lines
88
to
95
| public static void pollUntilTrue(Consumer<byte[], byte[]> consumer, | ||
| Supplier<Boolean> testCondition, | ||
| long waitTimeMs, String msg) throws InterruptedException { | ||
| TestUtils.waitForCondition(() -> { | ||
| consumer.poll(Duration.ofMillis(100)); | ||
| return testCondition.get(); | ||
| }, waitTimeMs, msg); | ||
| } |
There was a problem hiding this comment.
Please follow the code style used in this file.
Suggested change
| public static void pollUntilTrue(Consumer<byte[], byte[]> consumer, | |
| Supplier<Boolean> testCondition, | |
| long waitTimeMs, String msg) throws InterruptedException { | |
| TestUtils.waitForCondition(() -> { | |
| consumer.poll(Duration.ofMillis(100)); | |
| return testCondition.get(); | |
| }, waitTimeMs, msg); | |
| } | |
| public static void pollUntilTrue( | |
| Consumer<byte[], byte[]> consumer, | |
| Supplier<Boolean> testCondition, | |
| long waitTimeMs, | |
| String msg | |
| ) throws InterruptedException { | |
| TestUtils.waitForCondition(() -> { | |
| consumer.poll(Duration.ofMillis(100)); | |
| return testCondition.get(); | |
| }, waitTimeMs, msg); | |
| } |
| testAssignAndCommitSyncNotCommitted(GroupProtocol.CONSUMER.name); | ||
| } | ||
|
|
||
| private void testAssignAndCommitSyncNotCommitted(String groupProtocol) { |
There was a problem hiding this comment.
Please use GroupProtocol instead of String
Suggested change
| private void testAssignAndCommitSyncNotCommitted(String groupProtocol) { | |
| private void testAssignAndCommitSyncNotCommitted(GroupProtocol groupProtocol) { |
| testAssignAndCommitSyncAllConsumed(GroupProtocol.CONSUMER.name); | ||
| } | ||
|
|
||
| private void testAssignAndCommitSyncAllConsumed(String groupProtocol) throws InterruptedException { |
There was a problem hiding this comment.
Suggested change
| private void testAssignAndCommitSyncAllConsumed(String groupProtocol) throws InterruptedException { | |
| private void testAssignAndCommitSyncAllConsumed(GroupProtocol groupProtocol) throws InterruptedException { |
| testAssignAndConsume(GroupProtocol.CONSUMER.name); | ||
| } | ||
|
|
||
| private void testAssignAndConsume(String groupProtocol) throws InterruptedException { |
There was a problem hiding this comment.
Suggested change
| private void testAssignAndConsume(String groupProtocol) throws InterruptedException { | |
| private void testAssignAndConsume(GroupProtocol groupProtocol) throws InterruptedException { |
| testAssignAndConsumeSkippingPosition(GroupProtocol.CONSUMER.name); | ||
| } | ||
|
|
||
| private void testAssignAndConsumeSkippingPosition(String groupProtocol) throws InterruptedException { |
There was a problem hiding this comment.
Suggested change
| private void testAssignAndConsumeSkippingPosition(String groupProtocol) throws InterruptedException { | |
| private void testAssignAndConsumeSkippingPosition(GroupProtocol groupProtocol) throws InterruptedException { |
| testAssignAndRetrievingCommittedOffsetsMultipleTimes(GroupProtocol.CONSUMER.name); | ||
| } | ||
|
|
||
| private void testAssignAndRetrievingCommittedOffsetsMultipleTimes(String groupProtocol) throws InterruptedException { |
There was a problem hiding this comment.
Suggested change
| private void testAssignAndRetrievingCommittedOffsetsMultipleTimes(String groupProtocol) throws InterruptedException { | |
| private void testAssignAndRetrievingCommittedOffsetsMultipleTimes(GroupProtocol groupProtocol) throws InterruptedException { |
| int numRecords = 100; | ||
| long startingTimestamp = System.currentTimeMillis(); | ||
|
|
||
| try (Consumer<byte[], byte[]> consumer = clusterInstance.consumer(Map.of(GROUP_PROTOCOL_CONFIG, groupProtocol, GROUP_ID_CONFIG, "group1"))) { |
There was a problem hiding this comment.
Does this test require manually setting the GROUP_ID_CONFIG?
There was a problem hiding this comment.
Remove it, thanks!``
Comment on lines
+294
to
+296
| // Check committed offsets twice with same consumer | ||
| assertEquals(numRecords, consumer.committed(Set.of(tp)).get(tp).offset()); | ||
| assertEquals(numRecords, consumer.committed(Set.of(tp)).get(tp).offset()); |
There was a problem hiding this comment.
Just curious, Why do we need to check the same consumer committed twice?
There was a problem hiding this comment.
From the test name testAssignAndRetrievingCommittedOffsetsMultipleTimes we can get this test would like to test committed offset multiple time so we need to keep the behavior .
| int offset = 10; | ||
| long startingTimestamp = System.currentTimeMillis(); | ||
|
|
||
| try (Consumer<byte[], byte[]> consumer = clusterInstance.consumer(Map.of(GROUP_PROTOCOL_CONFIG, groupProtocol, GROUP_ID_CONFIG, "group1"))) { |
There was a problem hiding this comment.
How about extract the same config as a variable.
val config = Map.of(GROUP_PROTOCOL_CONFIG, groupProtocol, GROUP_ID_CONFIG, "group1");
| int numRecords = 100; | ||
| long startingTimestamp = System.currentTimeMillis(); | ||
|
|
||
| try (Consumer<byte[], byte[]> consumer = clusterInstance.consumer(Map.of(GROUP_PROTOCOL_CONFIG, groupProtocol, GROUP_ID_CONFIG, "group1"))) { |
There was a problem hiding this comment.
How about extract the same config as a variable.
val config = Map.of(GROUP_PROTOCOL_CONFIG, groupProtocol, GROUP_ID_CONFIG, "group1");
|
Hi @m1a2st, sorry for delay. All comments are addressed, PTAL. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
The PR do following: