KAFKA-16901: Add unit tests for ConsumerRecords#records(String) by frankvicky · Pull Request #16227 · apache/kafka

frankvicky · 2024-06-06T12:57:51Z

Add unit tests for ConsumerRecords#records(String).

This commit adds new unit tests to the ConsumerRecordsTest class. One of the tests validates that an IllegalArgumentException occurs when a null topic is passed, and another test validates that the records(String) method is behaving as expected when valid topics are passed.

Committer Checklist (excluded from commit message)

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

chia7712

@frankvicky thanks for increasing the test coverage :)

chia7712 · 2024-06-06T16:58:59Z

+    @Test
+    public void testRecords() {
+        String targetTopic = "topic";
+        Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records = buildTestRecordsWithDummyRecords(targetTopic);


Could you build the ConsumerRecords with multiple topics? Also, we should verify the record for each topic.

Sure, will do it 😸

frankvicky · 2024-06-07T13:45:14Z

Hi @chia7712, I have modified the test case to cover the scenario of multiple topics and records, PLTA 😺

chia7712

@frankvicky thanks for updated PR

chia7712 · 2024-06-07T22:31:40Z

+        return records;
+    }
+
+    private Map<TopicPartition, List<ConsumerRecord<Integer, String>>> buildTopicTestRecords(int recordSize, String... topics) {


Maybe this method can return ConsumerRecords directly?

Yes, it is more clear that we return ConsumerRecords directly.

chia7712 · 2024-06-07T22:31:46Z

+    private Map<TopicPartition, List<ConsumerRecord<Integer, String>>> buildTopicTestRecords(int recordSize, String... topics) {
+        Map<TopicPartition, List<ConsumerRecord<Integer, String>>> partitionToRecords = new LinkedHashMap<>();
+        for (String topic : topics) {
+            ArrayList<ConsumerRecord<Integer, String>> records = new ArrayList<>(recordSize);


chia7712 · 2024-06-07T22:33:56Z

+        ConsumerRecord<Integer, String> record2 = new ConsumerRecord<>(topic, 1, 1, 0L, TimestampType.CREATE_TIME,
+            0, 0, 2, "value2", new RecordHeaders(), Optional.empty());
+
+        new ArrayList<>();


Oops, I accidentally leave this, I will remove it.

chia7712 · 2024-06-07T22:36:24Z

+        }
+    }
+
+    private Map<TopicPartition, List<ConsumerRecord<Integer, String>>> buildSingleTopicTestRecords(String topic) {


Can this implementation be rewrite by buildTopicTestRecords?

I keep this because the test case iterator has it' own test logic but I will try to rewrite it.

frankvicky · 2024-06-08T05:15:27Z

Hi @chia7712, I do few changes based on comments, PTAL 🐧

chia7712

@frankvicky thanks for updated PR!

chia7712 · 2024-06-08T12:05:03Z

+    public void testRecordsWithNullTopic() {
+        String nullTopic = null;
        Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records = new LinkedHashMap<>();
+        ConsumerRecords<Integer, String> consumerRecords = new ConsumerRecords<>(records);


Maybe we can use ConsumerRecords.empty();?

Cool, I don't know we have this utility function, I will replace with it.

chia7712 · 2024-06-08T12:07:08Z

+            Iterable<ConsumerRecord<Integer, String>> records = consumerRecords.records(topic);
+            Iterator<ConsumerRecord<Integer, String>> iterator = records.iterator();
+
+            for (int count = 0; iterator.hasNext(); count++) {


Could you verify the number of records from the iterable/iterator?

Sure, I will add the verification of it.

chia7712 · 2024-06-08T12:07:39Z

+    private ConsumerRecords<Integer, String> buildTopicTestRecords(int recordSize,
+                                                                   int partitionSize,
+                                                                   int emptyPartitionInterval,
+                                                                   String... topics) {


Could you please use Collection<String> ?

chia7712 · 2024-06-08T12:08:01Z

+                    ConsumerRecord<Integer, String> record = iterator.next();
+                    assertEquals(count, record.partition());
+                    assertEquals(topic, record.topic());
+                    assertEquals(offset, record.offset());


please verify the "key" and "value"

Oops, I miss both of them.

frankvicky · 2024-06-08T14:48:59Z

Hi @chia7712 , I do some changes based on comments, PTAL 😃

chia7712 · 2024-06-08T16:11:32Z

+        String topic = "topic";
+        int recordSize = 10;
+        int partitionSize = 15;
+        int emptyPartitionInterval = 3;


Could we define specific partition to be empty? that will get simplified I think

frankvicky · 2024-06-09T06:35:29Z

Hi @chia7712 , I have do some refactors, PTAL 🐧

chia7712

@frankvicky nice test cases and refactor.

chia7712 · 2024-06-09T17:21:01Z

 public class ConsumerRecordsTest {

    @Test
    public void iterator() throws Exception {


could you please rename it to testIterator?

Sure, don't even notice this case doesn't have test prefix 🤣

chia7712 · 2024-06-09T17:21:36Z

+
+
+    @Test
+    public void testRecords() {


Could you add test for records(TopicPartition)?

I will do it 😸

chia7712 · 2024-06-09T17:22:01Z

+                    currentPartition = record.partition();
+                }
+
+                assertEquals(topic, record.topic());


those checks are existent in both test cases. Maybe we can reuse them?

Cool, it will be more clear

frankvicky · 2024-06-10T02:17:52Z

Hi @chia7712 , I have do some changes and add a test case, PTAL 🐧

chia7712 · 2024-06-11T00:26:23Z

+            int partitionCount = 0;
+            int currentPartition = -1;
+
+            while (iterator.hasNext()) {


Maybe we can loop records? for (ConsumerRecord<Integer, String> record : records) {

Indeed, it will be more easy to read than using iterator.

frankvicky · 2024-06-11T06:02:43Z

Hi @chia7712, I have do a small change based on your feedback, PTAL 😄

chia7712

@frankvicky thanks for updated PR

chia7712 · 2024-06-11T19:28:05Z


    @Test
-    public void iterator() throws Exception {
+    public void testIterator() throws Exception {


Exception is unused.

Oops, I will remove it

chia7712 · 2024-06-11T19:28:18Z

+    }
+
+    private void validateEmptyPartition(ConsumerRecord<Integer, String> record, int emptyPartitionIndex) {
+        if (record.partition() == emptyPartitionIndex) {


It can be assertNotEquals(emptyPartitionIndex, record.partition(), "Partition " + record.partition() + " is not empty");, right?

Yes, in this way it could be more straightforward

chia7712

LGTM

@tag

commit f380cd1 Author: Edoardo Comar <ecomar@uk.ibm.com> Date: Thu Jun 13 15:01:08 2024 +0100 MINOR: Add integration tag to AdminFenceProducersIntegrationTest (apache#16326) Add @tag("integration") to AdminFenceProducersIntegrationTest Reviewers: Chris Egerton <chrise@aiven.io> commit 11c85a9 Author: Dongnuo Lyu <139248811+dongnuo123@users.noreply.github.com> Date: Thu Jun 13 05:11:01 2024 -0400 MINOR: Make online downgrade failure logs less noisy and update the timeouts scheduled in `convertToConsumerGroup` (apache#16290) This patch: - changes the order of the checks in `validateOnlineDowngrade`, so that only when the last member using the consumer protocol leave and the group still has classic member(s), `online downgrade is disabled` is logged if the policy doesn't allow downgrade. - changes the session timeout in `convertToConsumerGroup` from `consumerGroupSessionTimeoutMs` to `member.classicProtocolSessionTimeout().get()`. Reviewers: David Jacot <djacot@confluent.io> commit ea60666 Author: Ken Huang <100591800+m1a2st@users.noreply.github.com> Date: Thu Jun 13 17:11:37 2024 +0900 KAFKA-16921 [1/N] Migrate all junit 4 code to junit 5 for connect module (apache#16253) Reviewers: Chia-Ping Tsai <chia7712@gmail.com> commit 596b945 Author: gongxuanzhang <gongxuanzhang@foxmail.com> Date: Thu Jun 13 15:39:32 2024 +0800 KAFKA-16643 Add ModifierOrder checkstyle rule (apache#15890) Reviewers: Chia-Ping Tsai <chia7712@gmail.com> commit 103ff5c Author: Antoine Pourchet <antoine@responsive.dev> Date: Thu Jun 13 01:32:39 2024 -0600 KAFKA-15045: (KIP-924 pt. 24) internal TaskAssignor rename to LegacyTaskAssignor (apache#16318) Since the new public API for TaskAssignor shared a name, this rename will prevent users from confusing the internal definition with the public one. Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org> commit e59c887 Author: brenden20 <118419078+brenden20@users.noreply.github.com> Date: Thu Jun 13 02:30:05 2024 -0500 KAFKA-16557 Implemented OffsetFetchRequestState toStringBase and added a test for it (apache#16291) Reviewers: Lianet Magrans <lianetmr@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com> commit dd6fcc6 Author: TingIāu "Ting" Kì <kitingiao@gmail.com> Date: Thu Jun 13 14:35:33 2024 +0800 KAFKA-16901 Add unit tests for ConsumerRecords#records(String) (apache#16227) Reviewers: Chia-Ping Tsai <chia7712@gmail.com> commit fe98888 Author: Lianet Magrans <98415067+lianetm@users.noreply.github.com> Date: Thu Jun 13 08:31:16 2024 +0200 MINOR: Improving log for outstanding requests on close and cleanup (apache#16304) Reviewers: Andrew Schofield <aschofield@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com> commit 9ddd58b Author: Chris Egerton <chrise@aiven.io> Date: Thu Jun 13 05:43:33 2024 +0200 MINOR: Add readiness check for connector and separate Kafka cluster in ExactlyOnceSourceIntegrationTest::testSeparateOffsetsTopic (apache#16306) Reviewers: Greg Harris <gharris1727@gmail.com> commit 0a203a9 Author: TingIāu "Ting" Kì <kitingiao@gmail.com> Date: Thu Jun 13 09:47:51 2024 +0800 KAFKA-16938 non-dynamic props gets corrupted due to circular reference between DynamicBrokerConfig and DynamicConfig. (apache#16302) Reviewers: Chia-Ping Tsai <chia7712@gmail.com> commit 6d1f8f8 Author: Gantigmaa Selenge <39860586+tinaselenge@users.noreply.github.com> Date: Thu Jun 13 02:42:39 2024 +0100 MINOR: Clean up for KafkaAdminClientTest (apache#16285) Reviewers: Chia-Ping Tsai <chia7712@gmail.com> commit e76e1da Author: Chris Egerton <chrise@aiven.io> Date: Thu Jun 13 02:18:23 2024 +0200 KAFKA-16935: Automatically wait for cluster startup in embedded Connect integration tests (apache#16288) Reviewers: Greg Harris <gharris1727@gmail.com>

@tag

commit 4333af5 Author: A. Sophie Blee-Goldman <ableegoldman@gmail.com> Date: Thu Jun 13 11:27:50 2024 -0700 KAFKA-15045: (KIP-924 pt. 25) Rename old internal StickyTaskAssignor to LegacyStickyTaskAssignor (apache#16322) To avoid confusion in 3.8/until we fully remove all the old task assignors and internal config, we should rename the old internal assignor classes like the StickyTaskAssignor so that they won't be mixed up with the new version of the assignor (which is also named StickyTaskAssignor) Reviewers: Bruno Cadonna <cadonna@apache.org>, Josep Prat <josep.prat@aiven.io> commit f380cd1 Author: Edoardo Comar <ecomar@uk.ibm.com> Date: Thu Jun 13 15:01:08 2024 +0100 MINOR: Add integration tag to AdminFenceProducersIntegrationTest (apache#16326) Add @tag("integration") to AdminFenceProducersIntegrationTest Reviewers: Chris Egerton <chrise@aiven.io> commit 11c85a9 Author: Dongnuo Lyu <139248811+dongnuo123@users.noreply.github.com> Date: Thu Jun 13 05:11:01 2024 -0400 MINOR: Make online downgrade failure logs less noisy and update the timeouts scheduled in `convertToConsumerGroup` (apache#16290) This patch: - changes the order of the checks in `validateOnlineDowngrade`, so that only when the last member using the consumer protocol leave and the group still has classic member(s), `online downgrade is disabled` is logged if the policy doesn't allow downgrade. - changes the session timeout in `convertToConsumerGroup` from `consumerGroupSessionTimeoutMs` to `member.classicProtocolSessionTimeout().get()`. Reviewers: David Jacot <djacot@confluent.io> commit ea60666 Author: Ken Huang <100591800+m1a2st@users.noreply.github.com> Date: Thu Jun 13 17:11:37 2024 +0900 KAFKA-16921 [1/N] Migrate all junit 4 code to junit 5 for connect module (apache#16253) Reviewers: Chia-Ping Tsai <chia7712@gmail.com> commit 596b945 Author: gongxuanzhang <gongxuanzhang@foxmail.com> Date: Thu Jun 13 15:39:32 2024 +0800 KAFKA-16643 Add ModifierOrder checkstyle rule (apache#15890) Reviewers: Chia-Ping Tsai <chia7712@gmail.com> commit 103ff5c Author: Antoine Pourchet <antoine@responsive.dev> Date: Thu Jun 13 01:32:39 2024 -0600 KAFKA-15045: (KIP-924 pt. 24) internal TaskAssignor rename to LegacyTaskAssignor (apache#16318) Since the new public API for TaskAssignor shared a name, this rename will prevent users from confusing the internal definition with the public one. Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org> commit e59c887 Author: brenden20 <118419078+brenden20@users.noreply.github.com> Date: Thu Jun 13 02:30:05 2024 -0500 KAFKA-16557 Implemented OffsetFetchRequestState toStringBase and added a test for it (apache#16291) Reviewers: Lianet Magrans <lianetmr@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com> commit dd6fcc6 Author: TingIāu "Ting" Kì <kitingiao@gmail.com> Date: Thu Jun 13 14:35:33 2024 +0800 KAFKA-16901 Add unit tests for ConsumerRecords#records(String) (apache#16227) Reviewers: Chia-Ping Tsai <chia7712@gmail.com> commit fe98888 Author: Lianet Magrans <98415067+lianetm@users.noreply.github.com> Date: Thu Jun 13 08:31:16 2024 +0200 MINOR: Improving log for outstanding requests on close and cleanup (apache#16304) Reviewers: Andrew Schofield <aschofield@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com> commit 9ddd58b Author: Chris Egerton <chrise@aiven.io> Date: Thu Jun 13 05:43:33 2024 +0200 MINOR: Add readiness check for connector and separate Kafka cluster in ExactlyOnceSourceIntegrationTest::testSeparateOffsetsTopic (apache#16306) Reviewers: Greg Harris <gharris1727@gmail.com> commit 0a203a9 Author: TingIāu "Ting" Kì <kitingiao@gmail.com> Date: Thu Jun 13 09:47:51 2024 +0800 KAFKA-16938 non-dynamic props gets corrupted due to circular reference between DynamicBrokerConfig and DynamicConfig. (apache#16302) Reviewers: Chia-Ping Tsai <chia7712@gmail.com> commit 6d1f8f8 Author: Gantigmaa Selenge <39860586+tinaselenge@users.noreply.github.com> Date: Thu Jun 13 02:42:39 2024 +0100 MINOR: Clean up for KafkaAdminClientTest (apache#16285) Reviewers: Chia-Ping Tsai <chia7712@gmail.com> commit e76e1da Author: Chris Egerton <chrise@aiven.io> Date: Thu Jun 13 02:18:23 2024 +0200 KAFKA-16935: Automatically wait for cluster startup in embedded Connect integration tests (apache#16288) Reviewers: Greg Harris <gharris1727@gmail.com>

frankvicky added 2 commits June 6, 2024 20:51

KAFKA-16901: Add unit tests for ConsumerRecords#records(String)

3015cab

KAFKA-16901: Use static import for assertEquals method.

04b43bf

chia7712 reviewed Jun 6, 2024

View reviewed changes

frankvicky added 2 commits June 7, 2024 21:04

Merge branch 'trunk' into KAFKA-16901

1fecb3e

KAFKA-16901: Refactor the test case to verify mutiple topics and records

4b8ab1b

chia7712 reviewed Jun 7, 2024

View reviewed changes

frankvicky added 2 commits June 8, 2024 10:25

Merge branch 'trunk' into KAFKA-16901

fed6732

KAFKA-16901: Refactor unit tests.

a14d176

chia7712 reviewed Jun 8, 2024

View reviewed changes

frankvicky added 2 commits June 8, 2024 22:46

KAFKA-16901: Add verification of key, value and size of record.

ba5712d

Merge branch 'trunk' into KAFKA-16901

fd92c43

frankvicky added 2 commits June 8, 2024 22:53

KAFKA-16901: replace hardcoded value with variable.

60b9eee

KAFKA-16901: Extract complicated logic as helper function.

3803699

chia7712 reviewed Jun 8, 2024

View reviewed changes

frankvicky added 2 commits June 9, 2024 13:14

Merge branch 'trunk' into KAFKA-16901

ae3bf6d

KAFKA-16901: Refactor all test cases.

584c30b

chia7712 reviewed Jun 9, 2024

View reviewed changes

KAFKA-16901: Add test case for records(TopicPartition)

5cb64fe

chia7712 reviewed Jun 11, 2024

View reviewed changes

KAFKA-16901: Replace iterator with enhance-for

6d1f8a1

KAFKA-16901: Remove unnecessary variable

0e452a7

chia7712 reviewed Jun 11, 2024

View reviewed changes

KAFKA-16901: simpify assertion

f43af6d

chia7712 approved these changes Jun 12, 2024

View reviewed changes

chia7712 merged commit dd6fcc6 into apache:trunk Jun 13, 2024

Conversation

frankvicky commented Jun 6, 2024

Committer Checklist (excluded from commit message)

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

frankvicky commented Jun 7, 2024

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

frankvicky commented Jun 8, 2024

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

frankvicky commented Jun 8, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

frankvicky commented Jun 9, 2024

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

frankvicky commented Jun 10, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

frankvicky commented Jun 11, 2024

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment