This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 2283829d791 MINOR: Rewrite ReplicaFetchTest from scala to java (#23059)
2283829d791 is described below
commit 2283829d7914a06c8a147e40eea949f7234c7d2f
Author: Ken Huang <[email protected]>
AuthorDate: Thu Aug 6 00:17:39 2026 +0800
MINOR: Rewrite ReplicaFetchTest from scala to java (#23059)
Rewrite ReplicaFetchTest and move to server module
Reviewers: Chia-Ping Tsai <[email protected]>
---
.../scala/unit/kafka/server/ReplicaFetchTest.scala | 73 -------------------
.../org/apache/kafka/server/ReplicaFetchTest.java | 85 ++++++++++++++++++++++
2 files changed, 85 insertions(+), 73 deletions(-)
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaFetchTest.scala
b/core/src/test/scala/unit/kafka/server/ReplicaFetchTest.scala
deleted file mode 100644
index 66b41c0aaf1..00000000000
--- a/core/src/test/scala/unit/kafka/server/ReplicaFetchTest.scala
+++ /dev/null
@@ -1,73 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
-*/
-
-package kafka.server
-
-import org.junit.jupiter.api.{AfterEach, Test}
-import kafka.utils.TestUtils
-import TestUtils._
-import kafka.api.IntegrationTestHarness
-import org.apache.kafka.clients.producer.ProducerRecord
-import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.serialization.StringSerializer
-
-class ReplicaFetchTest extends IntegrationTestHarness {
- val topic1 = "foo"
- val topic2 = "bar"
-
- @AfterEach
- override def tearDown(): Unit = {
- TestUtils.shutdownServers(brokers)
- super.tearDown()
- }
-
- override def brokerCount: Int = 2
-
- @Test
- def testReplicaFetcherThread(): Unit = {
- val partition = 0
- val testMessageList1 = List("test1", "test2", "test3", "test4")
- val testMessageList2 = List("test5", "test6", "test7", "test8")
-
- // create a topic and partition and await leadership
- for (topic <- List(topic1,topic2)) {
- createTopic(topic, replicationFactor = 2)
- }
-
- // send test messages to leader
- val producer =
TestUtils.createProducer(TestUtils.plaintextBootstrapServers(brokers),
- keySerializer = new
StringSerializer,
- valueSerializer = new
StringSerializer)
- val records = testMessageList1.map(m => new ProducerRecord(topic1, m, m))
++
- testMessageList2.map(m => new ProducerRecord(topic2, m, m))
- records.map(producer.send).foreach(_.get)
- producer.close()
-
- def logsMatch(): Boolean = {
- var result = true
- for (topic <- List(topic1, topic2)) {
- val tp = new TopicPartition(topic, partition)
- val expectedOffset =
brokers.head.logManager.getLog(tp).get.logEndOffset
- result = result && expectedOffset > 0 && brokers.forall { item =>
- expectedOffset == item.logManager.getLog(tp).get.logEndOffset
- }
- }
- result
- }
- waitUntilTrue(logsMatch _, "Broker logs should be identical")
- }
-}
diff --git a/server/src/test/java/org/apache/kafka/server/ReplicaFetchTest.java
b/server/src/test/java/org/apache/kafka/server/ReplicaFetchTest.java
new file mode 100644
index 00000000000..6060ddd02d7
--- /dev/null
+++ b/server/src/test/java/org/apache/kafka/server/ReplicaFetchTest.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.server;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.test.TestUtils;
+
+import java.util.List;
+import java.util.Map;
+
+@ClusterTestDefaults(brokers = 2, types = {Type.KRAFT})
+public class ReplicaFetchTest {
+
+ private final ClusterInstance cluster;
+
+ ReplicaFetchTest(ClusterInstance cluster) {
+ this.cluster = cluster;
+ }
+
+ @ClusterTest
+ public void testReplicaFetcherThread() throws Exception {
+ var topic1 = "foo";
+ var topic2 = "bar";
+ var partition = 0;
+ var testMessageList1 = List.of("test1", "test2", "test3", "test4");
+ var testMessageList2 = List.of("test5", "test6", "test7", "test8");
+
+ // create topics with replication factor 2 and await leadership
+ cluster.createTopic(topic1, 1, (short) 2);
+ cluster.createTopic(topic2, 1, (short) 2);
+
+ // send test messages to leader
+ try (var producer = cluster.producer(Map.of(
+ ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName(),
+ ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName()))) {
+ for (var m : testMessageList1) {
+ producer.send(new ProducerRecord<>(topic1, m, m)).get();
+ }
+ for (var m : testMessageList2) {
+ producer.send(new ProducerRecord<>(topic2, m, m)).get();
+ }
+ }
+
+ TestUtils.waitForCondition(
+ () -> {
+ for (var topic : List.of(topic1, topic2)) {
+ var tp = new TopicPartition(topic, partition);
+ long expectedOffset = -1;
+ for (var broker : cluster.brokers().values()) {
+ var logEndOffset =
broker.logManager().getLog(tp).map(log -> log.logEndOffset()).orElse(0L);
+ if (expectedOffset == -1) {
+ expectedOffset = logEndOffset;
+ }
+ if (expectedOffset <= 0 || logEndOffset !=
expectedOffset) {
+ return false;
+ }
+ }
+ }
+ return true;
+ },
+ "Broker logs should be identical"
+ );
+ }
+}
