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

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`

Reviewers: TengYao Chi <[email protected]>, TaiJuWu
 <[email protected]>, Chia-Ping Tsai <[email protected]>

Author: joshua2519

core/src/main/java/kafka/server/share/SharePartitionManager.java

@@ -16,7 +16,6 @@
  */
 package kafka.server.share;
 
-import kafka.cluster.PartitionListener;
 import kafka.server.ReplicaManager;
 
 import org.apache.kafka.common.TopicIdPartition;
@@ -59,6 +58,7 @@
 import org.apache.kafka.server.util.timer.SystemTimerReaper;
 import org.apache.kafka.server.util.timer.Timer;
 import org.apache.kafka.server.util.timer.TimerTask;
+import org.apache.kafka.storage.internals.log.PartitionListener;
 import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
 
 import org.slf4j.Logger;

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

@@ -16,7 +16,6 @@
  */
 package kafka.coordinator.group
 
-import kafka.cluster.PartitionListener
 import kafka.server.ReplicaManager
 import org.apache.kafka.common.{TopicIdPartition, TopicPartition}
 import org.apache.kafka.common.protocol.Errors
@@ -25,7 +24,7 @@ import org.apache.kafka.coordinator.common.runtime.PartitionWriter
 import org.apache.kafka.server.ActionQueue
 import org.apache.kafka.server.common.RequestLocal
 import org.apache.kafka.server.transaction.AddPartitionsToTxnManager
-import org.apache.kafka.storage.internals.log.{AppendOrigin, LogConfig, VerificationGuard}
+import org.apache.kafka.storage.internals.log.{AppendOrigin, LogConfig, PartitionListener, VerificationGuard}
 
 import java.util.concurrent.CompletableFuture
 import scala.collection.Map

core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala

@@ -17,7 +17,7 @@
 package kafka.server
 
 import com.yammer.metrics.core.Meter
-import kafka.cluster.{Partition, PartitionListener}
+import kafka.cluster.{Partition}
 import kafka.log.LogManager
 import kafka.server.HostedPartition.Online
 import kafka.server.QuotaFactory.QuotaManagers
@@ -65,7 +65,7 @@ import org.apache.kafka.server.util.timer.{SystemTimer, TimerTask}
 import org.apache.kafka.server.util.{Scheduler, ShutdownableThread}
 import org.apache.kafka.server.{ActionQueue, DelayedActionQueue, common}
 import org.apache.kafka.storage.internals.checkpoint.{LazyOffsetCheckpoints, OffsetCheckpointFile, OffsetCheckpoints}
-import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchDataInfo, FetchPartitionStatus, LeaderHwChange, LogAppendInfo, LogConfig, LogDirFailureChannel, LogOffsetMetadata, LogReadInfo, LogReadResult, OffsetResultHolder, RecordValidationException, RemoteLogReadResult, RemoteStorageFetchInfo, UnifiedLog, VerificationGuard}
+import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchDataInfo, FetchPartitionStatus, LeaderHwChange, LogAppendInfo, LogConfig, LogDirFailureChannel, LogOffsetMetadata, LogReadInfo, LogReadResult, OffsetResultHolder, RecordValidationException, RemoteLogReadResult, RemoteStorageFetchInfo, UnifiedLog, VerificationGuard,PartitionListener}
 import org.apache.kafka.storage.log.metrics.BrokerTopicStats
 
 import java.io.File

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

@@ -27,7 +27,7 @@ import org.apache.kafka.server.common.MetadataVersion
 import org.apache.kafka.server.config.ReplicationConfigs
 import org.apache.kafka.server.util.MockTime
 import org.apache.kafka.storage.internals.checkpoint.OffsetCheckpoints
-import org.apache.kafka.storage.internals.log.{CleanerConfig, LogConfig}
+import org.apache.kafka.storage.internals.log.{CleanerConfig, LogConfig,AlterPartitionListener}
 import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
 import org.junit.jupiter.api.{AfterEach, BeforeEach}
 import org.mockito.ArgumentMatchers

core/src/test/scala/unit/kafka/cluster/AbstractPartitionTest.scala

@@ -16,14 +16,14 @@
   */
 package kafka.cluster
 
+import org.apache.kafka.storage.internals.log.SimpleAssignmentState
 import org.apache.kafka.common.DirectoryId
 import org.apache.kafka.metadata.{LeaderRecoveryState, PartitionRegistration}
 import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
 import org.junit.jupiter.params.ParameterizedTest
 import org.junit.jupiter.params.provider.{Arguments, MethodSource}
 
 import java.util
-import scala.jdk.CollectionConverters._
 
 object AssignmentStateTest {
   import AbstractPartitionTest._
@@ -88,7 +88,7 @@ class AssignmentStateTest extends AbstractPartitionTest {
   @MethodSource(Array("parameters"))
   def testPartitionAssignmentStatus(isr: Array[Int], replicas: Array[Int],
                                     adding: Array[Int], removing: Array[Int],
-                                    original: util.List[Int], isUnderReplicated: Boolean): Unit = {
+                                    original: util.List[Integer], isUnderReplicated: Boolean): Unit = {
     val partitionRegistration = new PartitionRegistration.Builder()
       .setLeader(brokerId)
       .setLeaderRecoveryState(LeaderRecoveryState.RECOVERED)
@@ -103,7 +103,7 @@ class AssignmentStateTest extends AbstractPartitionTest {
 
     // set the original replicas as the URP calculation will need them
     if (!original.isEmpty)
-      partition.assignmentState = SimpleAssignmentState(original.asScala)
+      partition.assignmentState = new SimpleAssignmentState(original)
     // do the test
     partition.makeLeader(partitionRegistration, isNew = false, offsetCheckpoints, None)
     val isReassigning = !adding.isEmpty || !removing.isEmpty

core/src/test/scala/unit/kafka/cluster/AssignmentStateTest.scala

@@ -37,7 +37,7 @@ import org.apache.kafka.server.storage.log.{FetchIsolation, FetchParams}
 import org.apache.kafka.server.util.MockTime
 import org.apache.kafka.storage.internals.checkpoint.OffsetCheckpoints
 import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
-import org.apache.kafka.storage.internals.log.{AppendOrigin, CleanerConfig, LocalLog, LogAppendInfo, LogConfig, LogDirFailureChannel, LogLoader, LogOffsetsListener, LogSegments, ProducerStateManager, ProducerStateManagerConfig, UnifiedLog, VerificationGuard}
+import org.apache.kafka.storage.internals.log.{AlterPartitionListener, AppendOrigin, CleanerConfig, CommittedPartitionState, LocalLog, LogAppendInfo, LogConfig, LogDirFailureChannel, LogLoader, LogOffsetsListener, LogSegments, PendingShrinkIsr, ProducerStateManager, ProducerStateManagerConfig, UnifiedLog, VerificationGuard}
 import org.apache.kafka.storage.log.metrics.BrokerTopicStats
 import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
 import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}