MINOR: Rewrite FetchFromFollowerIntegrationTest from Scala to Java and move to server module by cychiu8 · Pull Request #22252 · apache/kafka

cychiu8 · 2026-05-11T04:35:33Z

Summary
This PR consolidates FetchRequest related tests into the server module by merging FetchFromFollowerIntegrationTest
into FetchRequestTest.

Key changes include:

Moving FetchRequestTest from core to the server module.
Rewriting FetchFromFollowerIntegrationTest in Java and subsequently merging its integration tests into
FetchRequestTest.
Adding necessary utility methods to server/src/test/java/org/apache/kafka/server/TestUtils.java to support the
integration tests.

and move FetchRequestTest to server module

m1a2st

Can we also delete FetchFromFollowerIntegrationTest.scala?

cychiu8 · 2026-05-11T23:37:19Z

Can we also delete FetchFromFollowerIntegrationTest.scala?

@m1a2st Thanks for reviewing. You're right! We should delete the scala one.

chia7712 · 2026-05-12T07:07:32Z

+     * @param timeoutMs The duration in ms to wait for the leader
+     * @return The leader broker id
+     */
+    public static int waitUntilLeaderIsKnown(


If this is used solely by FetchRequestTest, would you mind moving it into the test class instead?

Done. waitUntilLeaderIsKnown has been moved into FetchRequestTest as a protected static method.

chia7712 · 2026-05-12T07:09:40Z

+
+    @ClusterTest
+    @Timeout(60)
+    public void testFetchFromFollowerWithRoll(ClusterInstance cluster) throws Exception {


We should test both consumers, right?

Yes, you are correct! testFetchFromFollowerWithRoll now creates and uses both followerConsumer and leaderConsumer.

chia7712 · 2026-05-12T07:09:56Z

+    }
+
+    @ClusterTest
+    @Timeout(60)


this is redundant annotation. It is already included by ClusterTest

Done. All @timeout annotations have been removed.

chia7712 · 2026-05-12T07:10:11Z

+
+    @ClusterTest
+    @Timeout(15)
+    public void testFetchFromLeaderWhilePreferredReadReplicaIsUnavailable(ClusterInstance cluster) throws Exception {


chia7712 · 2026-05-12T07:10:21Z

+    Map<String, Uuid> topicIds;
+
+    @ClusterTest
+    @Timeout(15)


chia7712 · 2026-05-12T07:11:14Z

+
+    private void verifyRackAwareAssignments(
+            ExecutorService executor,
+            List<?> consumers,


Could you use Object or byte[] instead?

Done. The parameter is now List<Consumer<byte[], byte[]>>.

chia7712 · 2026-05-12T07:12:01Z

+        topics.add(topic);
+        topics.addAll(List.of(additionalTopics));
+
+        for (Object c : consumers) {


consumers.forEach(Consumer::close);

chia7712 · 2026-05-12T07:13:53Z

+        for (Object c : consumers) {
+            recordFutures.add(executor.submit(() -> {
+                try {
+                    @SuppressWarnings("unchecked")


@SuppressWarnings("unchecked") smells fishy to me ...

Done. Resolved as a consequence of switching from List<?> to List<Consumer<byte[], byte[]>>. No cast or suppression is needed anymore.

m1a2st

Thanks for this patch, left some comments

m1a2st · 2026-05-12T12:28:48Z

+    @Timeout(15)
+    public void testFollowerCompleteDelayedFetchesOnReplication(ClusterInstance cluster) throws Exception {
+        this.cluster = cluster;
+        try (Admin admin = cluster.admin()) {


Why is this Admin client created but never used?

My mistake. The unused try (Admin admin = ...) block has been removed.

m1a2st · 2026-05-12T12:34:55Z

+                verifyRackAwareAssignments(executor, consumers, producer, partitionList, topicWithAllPartitionsOnAllRacks, partitionList, topicWithSingleRackPartitions);
+
+            } finally {
+                executor.shutdown();


This should be executor.shutdownNow().

Done. Thank you for reminding me the difference between executor.shutdown() and executor.shutdownNow()!

m1a2st · 2026-05-12T12:38:09Z

+    public static final int FOLLOWER_BROKER_ID = 1;
+
+    private ClusterInstance cluster;
+    Map<String, Uuid> topicIds;


This field can be a local variable since it’s only used in one method.

m1a2st · 2026-05-12T12:40:46Z

+            List<Integer> partitionList,
+            String topic,
+            List<Integer> expectedPartitionOrder,
+            String... additionalTopics) throws Exception {


Could we use List<String> instead?

chia7712 · 2026-05-16T02:14:43Z

+                .toList();
+
+        List<Consumer<byte[], byte[]>> consumers = consumerConfigs.stream()
+                .map(config -> cluster.<byte[], byte[]>consumer(config))


List<Consumer<byte[], byte[]>> consumers = consumerConfigs.stream() .map(cluster::<byte[], byte[]>consumer) .toList();

thank you for the review! updated.

chia7712 · 2026-05-18T00:13:54Z

+    private ClusterInstance cluster;
+
+    @ClusterTest
+    public void testFollowerCompleteDelayedFetchesOnReplication(ClusterInstance cluster) throws Exception {


It should test both consumers, right?

Yes! Thank you for reminding me.

chia7712 · 2026-05-18T00:16:48Z

+
+        Supplier<Optional<Integer>> newLeaderExists = () -> {
+            if (expectedLeaderOpt.isPresent()) {
+                LOG.debug("Checking leader that has changed to {}", expectedLeaderOpt.get());


I'm wondering whether we should keep these debug messages in this migration.

We could remove them because the waitForCondition timeout message already identifies the partition and timeoutMs.

m1a2st

Thanks for the update, few more comments

m1a2st · 2026-05-18T13:02:31Z

+        });
+
+        String expectedTopic = version >= 13 ? null : tp.topic();
+        topicIdPartitionMap = fetchRequest.fetchData(Collections.emptyMap());


Suggested change

topicIdPartitionMap = fetchRequest.fetchData(Collections.emptyMap());

topicIdPartitionMap = fetchRequest.fetchData(Map.of());

Thank you for the review and catch the old usage part. Updated.

m1a2st · 2026-05-18T13:02:46Z

+        Optional<Integer> currentLeaderEpoch = Optional.of(121);
+        FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(topicId, fetchOffset, logStartOffset, maxBytes, currentLeaderEpoch);
+        FetchRequest fetchRequest = createFetchRequestByVersion(version, tp, partitionData);
+        Map<Uuid, String> topicNames = Collections.singletonMap(topicId, tp.topic());


Suggested change

Map<Uuid, String> topicNames = Collections.singletonMap(topicId, tp.topic());

Map<Uuid, String> topicNames = Map.of(topicId, tp.topic());

m1a2st · 2026-05-18T13:03:00Z

+        } else {
+            return FetchRequest.Builder
+                .forReplica(version, 0, 1, 1, 1, partitionDataMap)
+                .removed(Collections.singletonList(tp))


Suggested change

.removed(Collections.singletonList(tp))

.removed(List.of(tp))

m1a2st · 2026-05-18T13:03:14Z

+        if (version >= 13) {
+            return FetchRequest.Builder
+                .forReplica(version, 0, 1, 1, 1, partitionDataMap)
+                .replaced(Collections.singletonList(tp))


Suggested change

.replaced(Collections.singletonList(tp))

.replaced(List.of(tp))

m1a2st · 2026-05-18T13:03:25Z

+
+    private FetchRequest createFetchRequestByVersion(short version, TopicIdPartition tp,
+                                                     FetchRequest.PartitionData partitionData) {
+        Map<TopicPartition, FetchRequest.PartitionData> partitionDataMap = Collections.singletonMap(tp.topicPartition(), partitionData);


Suggested change

Map<TopicPartition, FetchRequest.PartitionData> partitionDataMap = Collections.singletonMap(tp.topicPartition(), partitionData);

Map<TopicPartition, FetchRequest.PartitionData> partitionDataMap = Map.of(tp.topicPartition(), partitionData);

m1a2st · 2026-05-18T13:04:10Z

+            boolean fetchRequestUsesTopicIds = version >= 13;
+
+            FetchRequest fetchRequest = FetchRequest.parse(FetchRequest.Builder
+                    .forReplica(version, 0, 1, 1, 1, Collections.emptyMap())


Suggested change

.forReplica(version, 0, 1, 1, 1, Collections.emptyMap())

.forReplica(version, 0, 1, 1, 1, Map.of())

m1a2st · 2026-05-18T13:04:19Z

+            Uuid topicId0 = Uuid.randomUuid();
+            Uuid topicId1 = Uuid.randomUuid();
+            // Only include topic IDs for the first topic partition.
+            Map<Uuid, String> topicNames = Collections.singletonMap(topicId0, topicPartition0.topic());


Suggested change

Map<Uuid, String> topicNames = Collections.singletonMap(topicId0, topicPartition0.topic());

Map<Uuid, String> topicNames = Map.of(topicId0, topicPartition0.topic());

m1a2st · 2026-05-18T13:04:40Z

+                .removed(Collections.emptyList())
+                .replaced(Collections.emptyList())


Suggested change

.removed(Collections.emptyList())

.replaced(Collections.emptyList())

.removed(List.of())

.replaced(List.of())

m1a2st · 2026-05-18T13:04:49Z

+        Uuid topicId1 = Uuid.randomUuid();
+
+        // Only include topic IDs for the first topic partition.
+        Map<Uuid, String> topicNames = Collections.singletonMap(topicId0, topicPartition0.topic());


Suggested change

Map<Uuid, String> topicNames = Collections.singletonMap(topicId0, topicPartition0.topic());

Map<Uuid, String> topicNames = Map.of(topicId0, topicPartition0.topic());

m1a2st · 2026-05-18T13:05:01Z

+        Map<TopicPartition, FetchRequest.PartitionData> partitionData = Collections.singletonMap(tp.topicPartition(),
+                new FetchRequest.PartitionData(topicId, 0, 0, 0, Optional.empty()));
+        List<TopicIdPartition> toReplace = Collections.singletonList(tp);


…n createPartitionMap - replaced the manual connect + serialize + send + receive block with IntegrationTestUtils.connectAndReceive

cychiu8 added 2 commits May 11, 2026 13:21

refactor: rewrite FetchFromFollowerIntegrationTest in Java

659127f

and move FetchRequestTest to server module

refactor: merge FetchRequestTest with FetchFromFollowerIntegrationTest

186b66e

github-actions Bot added triage PRs from the community core Kafka Broker tests Test fixes (including flaky tests) clients labels May 11, 2026

m1a2st added the ci-approved label May 11, 2026

m1a2st reviewed May 11, 2026

View reviewed changes

refactor: delete FetchFromFollowerIntegrationTest.scala

0b6b6f3

github-actions Bot removed the triage PRs from the community label May 12, 2026

chia7712 reviewed May 12, 2026

View reviewed changes

m1a2st reviewed May 12, 2026

View reviewed changes

cychiu8 added 3 commits May 13, 2026 22:28

address review feedbacks

1956e3c

address review feedbacks

c7990c2

address review feedbacks

f59a0f3

chia7712 reviewed May 17, 2026

View reviewed changes

address review feedbacks

4e551e5

chia7712 reviewed May 18, 2026

View reviewed changes

cychiu8 added 2 commits May 18, 2026 16:24

refactor: address review feedbacks

bdd121f

refactor: use cluster as parameter instead of using private field

afb47a9

m1a2st reviewed May 18, 2026

View reviewed changes

cychiu8 added 2 commits May 18, 2026 22:23

refactor: replace Collections.* with modern Map.of/List.of.

635daae

refactor: - getTopicIds(cluster) is now called once before the loop i…

c1708f4

…n createPartitionMap - replaced the manual connect + serialize + send + receive block with IntegrationTestUtils.connectAndReceive

	topicIdPartitionMap = fetchRequest.fetchData(Collections.emptyMap());
	topicIdPartitionMap = fetchRequest.fetchData(Map.of());

	Map<Uuid, String> topicNames = Collections.singletonMap(topicId, tp.topic());
	Map<Uuid, String> topicNames = Map.of(topicId, tp.topic());

	.removed(Collections.singletonList(tp))
	.removed(List.of(tp))

	.replaced(Collections.singletonList(tp))
	.replaced(List.of(tp))

	Map<TopicPartition, FetchRequest.PartitionData> partitionDataMap = Collections.singletonMap(tp.topicPartition(), partitionData);
	Map<TopicPartition, FetchRequest.PartitionData> partitionDataMap = Map.of(tp.topicPartition(), partitionData);

	.forReplica(version, 0, 1, 1, 1, Collections.emptyMap())
	.forReplica(version, 0, 1, 1, 1, Map.of())

		.removed(Collections.emptyList())
		.replaced(Collections.emptyList())

Conversation

cychiu8 commented May 11, 2026

Uh oh!

m1a2st left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cychiu8 commented May 11, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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!

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!

m1a2st 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!

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!

m1a2st 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!

m1a2st left a comment •

edited

Loading

cychiu8 commented May 11, 2026 •

edited

Loading