KAFKA-19301: Move and rewrite partition state classes to Java in org.apache.kafka.storage.internals.log #20083

joshua2519 · 2025-07-02T02:34:12Z

move following interface/class to
org.apache.kafka.storage.internals.log and rewrtie to java

PartitionListener‎ AlterPartitionListener AssignmentState
OngoingReassignmentState SimpleAssignmentState PartitionState
PendingPartitionChange PendingExpandIsr PendingShrinkIsr
CommittedPartitionState

…ernals-log

chia7712

@joshua2519 thanks for this patch.

storage/src/main/java/org/apache/kafka/storage/internals/log/PendingShrinkIsr.java

frankvicky

@joshua2519: Thanks for the patch.

storage/src/main/java/org/apache/kafka/storage/internals/log/PendingExpandIsr.java

storage/src/main/java/org/apache/kafka/storage/internals/log/PendingShrinkIsr.java

…ernals-log

chia7712

@joshua2519 Thanks for this patch. The migration introduces some extra overhead in deep copying. Please try to eliminate the copying as much as possible.

storage/src/main/java/org/apache/kafka/storage/internals/log/CommittedPartitionState.java

storage/src/main/java/org/apache/kafka/storage/internals/log/PendingExpandIsr.java

storage/src/main/java/org/apache/kafka/storage/internals/log/PendingPartitionChange.java

storage/src/main/java/org/apache/kafka/storage/internals/log/PartitionState.java

chia7712 · 2025-07-04T18:37:01Z

core/src/main/scala/kafka/cluster/Partition.scala

@@ -1949,9 +1799,9 @@ class Partition(val topicPartition: TopicPartition,
      //   2) leaderAndIsr.partitionEpoch == partitionEpoch: No update was performed since proposed and actual state are the same.
      // In both cases, we want to move from Pending to Committed state to ensure new updates are processed.

-      partitionState = CommittedPartitionState(leaderAndIsr.isr.asScala.map(_.toInt).toSet, leaderAndIsr.leaderRecoveryState)
+      partitionState = new CommittedPartitionState(util.Set.copyOf(leaderAndIsr.isr), leaderAndIsr.leaderRecoveryState)


Could you change leaderAndIsr.isr to return immutable set?

Could you change leaderAndIsr.isr to return immutable set?

Thanks for that.
I changed it to immutable set and patched related test.

core/src/main/scala/kafka/cluster/Partition.scala

chia7712 · 2025-07-04T18:41:33Z

core/src/main/scala/kafka/cluster/Partition.scala

@@ -1298,7 +1148,7 @@ class Partition(val topicPartition: TopicPartition,
  def getOutOfSyncReplicas(maxLagMs: Long): Set[Int] = {
    val current = partitionState
    if (!current.isInflight) {
-      val candidateReplicaIds = current.isr - localBrokerId
+      val candidateReplicaIds = current.isr.asScala.map(_.toInt).toSet - localBrokerId


ditto

@chia7712
To reduce collection copies, we could use an iterator like this:
current.isr.asScala.iterator.map(_.toInt).filter(_ != localBrokerId)).to(Set)
However, this approach is less concise.

How about

current.isr.stream().filter(isr => isr != localBrokerId).collect(Collectors.toUnmodifiableSet).asScala

chia7712 · 2025-07-04T18:41:53Z

core/src/main/scala/kafka/cluster/Partition.scala

@@ -1098,7 +948,7 @@ class Partition(val topicPartition: TopicPartition,
            case (brokerId, logEndOffset) => s"broker $brokerId: $logEndOffset"
          }

-          val curInSyncReplicaObjects = (curMaximalIsr - localBrokerId).flatMap(getReplica)
+          val curInSyncReplicaObjects = (curMaximalIsr.asScala.map(_.toInt) - localBrokerId).flatMap(getReplica)


Same as #20083 (comment)

curMaximalIsr.asScala.iterator.filter(_.intValue() != localBrokerId).map(_.toInt).flatMap(getReplica).to(Set)

…ernals-log

chia7712

@joshua2519 thanks for updates. some comments are left. PTAL

storage/src/main/java/org/apache/kafka/storage/internals/log/CommittedPartitionState.java

storage/src/main/java/org/apache/kafka/storage/internals/log/PendingPartitionChange.java

chia7712 · 2025-07-15T14:41:33Z

storage/src/main/java/org/apache/kafka/storage/internals/log/PartitionState.java

+     * This set may include un-committed ISR members following an expansion. This "effective" ISR is used for advancing
+     * the high watermark as well as determining which replicas are required for acks=all produce requests.
+     *
+     * Only applicable as of IBP 2.7-IV2, for older versions this will return the committed ISR


this comment is out-of-date, shouldn't it?

Yes, the minimum versions is 3.3-IV3 now.
I removed this part.

…tion-scala-to-storage-internals-log

chia7712

@joshua2519 thanks for your patch. it is almost there. two small comments left.

storage/src/main/java/org/apache/kafka/storage/internals/log/AssignmentState.java

storage/src/main/java/org/apache/kafka/storage/internals/log/SimpleAssignmentState.java

…tion-scala-to-storage-internals-log

chia7712

@joshua2519 thanks for your patch. overall LGTM

storage/src/main/java/org/apache/kafka/storage/internals/log/OngoingReassignmentState.java

storage/src/main/java/org/apache/kafka/storage/internals/log/CommittedPartitionState.java

…tion-scala-to-storage-internals-log

joshua2519 added 6 commits July 1, 2025 22:54

move PartitionListener.java

61d3b8e

move AlterPartitionListener

3df2115

move AssignmentState and PartitionState

740fb12

check

90c8318

change to java collection

93c20a2

Merge branch 'trunk' into move-partial-Partition-scala-to-storage-int…

238ed5f

…ernals-log

github-actions bot added triage PRs from the community core Kafka Broker performance storage Pull requests that target the storage module KIP-932 Queues for Kafka labels Jul 2, 2025

frankvicky added the ci-approved label Jul 2, 2025

joshua2519 changed the title ~~MINOR: Move and rewrite partition state classes to Java in org.apache.kafka.storage.internals.log~~ KAFKA-19301: Move and rewrite partition state classes to Java in org.apache.kafka.storage.internals.log Jul 2, 2025

chia7712 reviewed Jul 3, 2025

View reviewed changes

storage/src/main/java/org/apache/kafka/storage/internals/log/PendingShrinkIsr.java Show resolved Hide resolved

frankvicky reviewed Jul 4, 2025

View reviewed changes

storage/src/main/java/org/apache/kafka/storage/internals/log/PendingExpandIsr.java Outdated Show resolved Hide resolved

storage/src/main/java/org/apache/kafka/storage/internals/log/PendingShrinkIsr.java Outdated Show resolved Hide resolved

github-actions bot added tiered-storage Related to the Tiered Storage feature clients group-coordinator and removed triage PRs from the community labels Jul 4, 2025

joshua2519 force-pushed the move-partial-Partition-scala-to-storage-internals-log branch from 197889a to 238ed5f Compare July 4, 2025 05:06

joshua2519 added 2 commits July 4, 2025 13:11

Merge branch 'trunk' into move-partial-Partition-scala-to-storage-int…

929b42f

…ernals-log

remove unnesscessry override

4c3efde

frankvicky approved these changes Jul 4, 2025

View reviewed changes

chia7712 reviewed Jul 4, 2025

View reviewed changes

joshua2519 added 6 commits July 8, 2025 15:44

Merge branch 'trunk' into move-partial-Partition-scala-to-storage-int…

abb764a

…ernals-log

remove override

94f5b00

fix boolean

b97411e

fix boolean

daefe12

fix collection copy

5ed44aa

change LeaderAndIsr.isr to return immutable set

9309b7e

Merge branch 'trunk' into move-partial-Partition-scala-to-storage-int…

075753f

…ernals-log

github-actions bot added the kraft label Jul 8, 2025

joshua2519 added 2 commits July 9, 2025 11:22

fix collection copy

19bdd00

to Set

91013d9

chia7712 reviewed Jul 15, 2025

View reviewed changes

joshua2519 added 4 commits July 16, 2025 15:50

Merge remote-tracking branch 'upstream/trunk' into move-partial-Parti…

75193f0

…tion-scala-to-storage-internals-log

Merge remote-tracking branch 'upstream/trunk' into move-partial-Parti…

c0f3ea3

…tion-scala-to-storage-internals-log

initialize isr as immutable

a5e3e62

Decouple AlterPartitionListener from PendingPartitionChange.

0ef88b9

TaiJuWu approved these changes Jul 17, 2025

View reviewed changes

joshua2519 added 2 commits July 17, 2025 23:05

Merge remote-tracking branch 'upstream/trunk' into move-partial-Parti…

854f343

…tion-scala-to-storage-internals-log

fix comments

3c9a1d8

chia7712 reviewed Jul 30, 2025

View reviewed changes

storage/src/main/java/org/apache/kafka/storage/internals/log/AssignmentState.java Outdated Show resolved Hide resolved

storage/src/main/java/org/apache/kafka/storage/internals/log/SimpleAssignmentState.java Outdated Show resolved Hide resolved

joshua2519 added 6 commits July 31, 2025 08:59

Merge remote-tracking branch 'upstream/trunk' into move-partial-Parti…

2a761fc

…tion-scala-to-storage-internals-log

remove default implementation in interface

b43bb0b

remove unmodifiableset

b50256b

make list immutable

b5979a0

implement methods in subclass

1fc3399

fix import

c5d1718

chia7712 reviewed Aug 5, 2025

View reviewed changes

storage/src/main/java/org/apache/kafka/storage/internals/log/OngoingReassignmentState.java Outdated Show resolved Hide resolved

storage/src/main/java/org/apache/kafka/storage/internals/log/CommittedPartitionState.java Outdated Show resolved Hide resolved

joshua2519 added 2 commits August 6, 2025 09:16

Merge remote-tracking branch 'upstream/trunk' into move-partial-Parti…

51a55d2

…tion-scala-to-storage-internals-log

compact constructor

b264418

KAFKA-19301: Move and rewrite partition state classes to Java in org.apache.kafka.storage.internals.log #20083

Are you sure you want to change the base?

KAFKA-19301: Move and rewrite partition state classes to Java in org.apache.kafka.storage.internals.log #20083

Uh oh!

Conversation

joshua2519 commented Jul 2, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

frankvicky left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

chia7712 Jul 4, 2025

Choose a reason for hiding this comment

Uh oh!

joshua2519 Jul 9, 2025 • 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!

Uh oh!

chia7712 Jul 4, 2025

Choose a reason for hiding this comment

Uh oh!

joshua2519 Jul 9, 2025

Choose a reason for hiding this comment

Uh oh!

frankvicky Jul 9, 2025

Choose a reason for hiding this comment

Uh oh!

chia7712 Jul 4, 2025

Choose a reason for hiding this comment

Uh oh!

joshua2519 Jul 9, 2025 • 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!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

chia7712 Jul 15, 2025

Choose a reason for hiding this comment

Uh oh!

joshua2519 Jul 17, 2025

Choose a reason for hiding this comment

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

joshua2519 commented Jul 2, 2025 •

edited by github-actions bot

Loading

joshua2519 Jul 9, 2025 •

edited

Loading

joshua2519 Jul 9, 2025 •

edited

Loading