MINOR: Migrate CoordinatorLoaderImplTest from Scala to Java #20089

LoganZhuZzz · 2025-07-02T09:44:45Z

Summary of Changes

Rewrote both CoordinatorLoaderImpl and CoordinatorLoaderImplTest in Java, replacing their original Scala implementations.
Removed the direct dependency on ReplicaManager and replaced it with functional interfaces for partitionLogSupplier and partitionLogEndOffsetSupplier
Preserved original logic and test coverage during migration.

…rdinator-loader

m1a2st

Thanks @LoganZhuZzz for this patch, left some comments

m1a2st · 2025-07-02T14:20:16Z

build.gradle

@@ -1660,9 +1660,11 @@ project(':coordinator-common') {
    implementation libs.slf4jApi
    implementation libs.metrics
    implementation libs.hdrHistogram
+    implementation libs.scalaLibrary


I don't think we should add scalaLibrary to the :coordinator-common module.

Thank you for the feedback.

m1a2st · 2025-07-02T14:24:46Z

checkstyle/import-control-coordinator-common.xml

                <allow pkg="org.apache.kafka.server.util" />
                <allow pkg="org.apache.kafka.storage.internals.log" />
                <allow pkg="org.apache.kafka.test" />
                <allow pkg="org.apache.kafka.timeline" />
                <allow pkg="org.HdrHistogram" />
+                <allow pkg="scala" />


Thank you for the feedback. I’m keeping coordinator-common as a pure Java module, but since the tests depend on the Scala-based ReplicaManager, we still need scala, like OptionConverters in test scope.

Would you prefer that we wait until ReplicaManager is refactored into Java before removing all Scala dependencies from this module, or would it make more sense to move these tests elsewhere, or wrap the Scala APIs in Java-friendly facade?

My understanding is that ReplicaManager is still required for the tests. cc @TaiJuWu @chia7712

Yes, ReplicaManager is needed here so we need to wait @chia7712 feedback.

m1a2st · 2025-07-02T14:25:57Z

...-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImpl.java

+ */
+public class CoordinatorLoaderImpl<T> implements CoordinatorLoader<T> {
+
+    private static final Logger LOG = LoggerFactory.getLogger(CoordinatorLoader.class);


Should the logger class be CoordinatorLoaderImpl?

Thank you for the feedback.

TaiJuWu · 2025-07-03T02:09:57Z

checkstyle/import-control-coordinator-common.xml

-    <!-- no one depends on the server -->
-    <disallow pkg="kafka" />


Should we do this change?
From my understanding, this prevents developer to depend on core module again in migrating process.

frankvicky · 2025-07-05T12:22:20Z

...-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImpl.java

+        if (records instanceof MemoryRecords) {
+            return (MemoryRecords) records;


Suggested change

if (records instanceof MemoryRecords) {

return (MemoryRecords) records;

if (records instanceof MemoryRecords memoryRecords) {

return memoryRecords;

frankvicky · 2025-07-05T12:55:34Z

core/src/main/scala/kafka/server/BrokerServer.scala

@@ -610,7 +611,8 @@ class BrokerServer(
    )
    val loader = new CoordinatorLoaderImpl[CoordinatorRecord](
      time,
-      replicaManager,
+      tp => toJava(replicaManager.getLog(tp)),


Suggested change

tp => toJava(replicaManager.getLog(tp)),

tp => replicaManager.getLog(tp).toJava,

frankvicky · 2025-07-05T12:55:55Z

core/src/main/scala/kafka/server/BrokerServer.scala

@@ -640,7 +642,8 @@ class BrokerServer(
    val serde = new ShareCoordinatorRecordSerde
    val loader = new CoordinatorLoaderImpl[CoordinatorRecord](
      time,
-      replicaManager,
+      tp => toJava(replicaManager.getLog(tp)),


Suggested change

tp => toJava(replicaManager.getLog(tp)),

tp => replicaManager.getLog(tp).toJava,

Thanks for the feedback. I plan to submit all the fixes together in a single update.

chia7712

@LoganZhuZzz thanks for this patch!

chia7712 · 2025-07-07T16:01:57Z

...-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImpl.java

+
+                stats.readAtLeastOneRecord = fetchDataInfo.records.sizeInBytes() > 0;
+
+                MemoryRecords memoryRecords = toReadableMemoryRecords(tp, fetchDataInfo.records, buffer);


the buffer should be reusable, right? in the scala code, we keep reusing the "larger" buffer. By contrast, in this version, the buffer is recreated many times.

Thanks for pointing that out.

I’m currently considering two ways to restore this behavior:

Introduce a simple wrapper class for the buffer, allowing it to be updated directly within the method.

Modify toReadableMemoryRecords to return a tuple containing both the MemoryRecords and the potentially updated buffer.

Would you have a preference between the two approaches?

chia7712 · 2025-07-07T16:06:42Z

...-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImpl.java

+                        throw new RuntimeException(msg, ex);
+                    }
+
+                    if (coordinatorRecordOpt.isPresent()) {


please use ifPresent instead

chia7712 · 2025-07-07T16:17:38Z

...-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImpl.java

+        return new ReplayResult(currentOffset, previousHighWatermark);
+    }
+
+    private void completeLoadFuture(


the simple implementation takes 5 arguments, and it has only one usage. Perhaps it should be inlined instead.

Thanks for the feedback. I’ve fixed these issues. PTAL

chia7712 · 2025-07-08T15:38:30Z

...-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorLoaderImpl.java

+
+                stats.readAtLeastOneRecord = fetchDataInfo.records.sizeInBytes() > 0;
+
+                MemoryRecords memoryRecords = toReadableMemoryRecords(tp, fetchDataInfo.records, bufferHolder);


Do we really need BufferHolder? Could you please try to take the buffer from MemoryRecords?

LoganZhuZzz added 2 commits July 2, 2025 17:32

MINOR: Migrate CoordinatorLoaderImplTest from Scala to Java

6ee09c7

Merge remote-tracking branch 'upstream/trunk' into minor-refactor-coo…

92097d8

…rdinator-loader

github-actions bot added triage PRs from the community core Kafka Broker build Gradle build or GitHub Actions group-coordinator labels Jul 2, 2025

gongxuanzhang added the ci-approved label Jul 2, 2025

m1a2st reviewed Jul 2, 2025

View reviewed changes

TaiJuWu reviewed Jul 3, 2025

View reviewed changes

github-actions bot removed the triage PRs from the community label Jul 3, 2025

frankvicky reviewed Jul 5, 2025

View reviewed changes

misc cleanup

fa501ad

chia7712 reviewed Jul 7, 2025

View reviewed changes

reuse buffer

79bb6a4

chia7712 reviewed Jul 8, 2025

View reviewed changes

		<!-- no one depends on the server -->
		<disallow pkg="kafka" />

		if (records instanceof MemoryRecords) {
		return (MemoryRecords) records;

	tp => toJava(replicaManager.getLog(tp)),
	tp => replicaManager.getLog(tp).toJava,


		stats.readAtLeastOneRecord = fetchDataInfo.records.sizeInBytes() > 0;

		MemoryRecords memoryRecords = toReadableMemoryRecords(tp, fetchDataInfo.records, buffer);

MINOR: Migrate CoordinatorLoaderImplTest from Scala to Java #20089

Are you sure you want to change the base?

MINOR: Migrate CoordinatorLoaderImplTest from Scala to Java #20089

Conversation

LoganZhuZzz commented Jul 2, 2025

Summary of Changes

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!

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!

Uh oh!