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

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/coordinator/group/CoordinatorPartitionWriter.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/server/ReplicaManager.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.controller.StateChangeLogger
 import kafka.log.LogManager
 import kafka.server.HostedPartition.Online
@@ -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, LogReadResult, common}
 import org.apache.kafka.storage.internals.checkpoint.{LazyOffsetCheckpoints, OffsetCheckpointFile, OffsetCheckpoints}
-import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchDataInfo, LeaderHwChange, LogAppendInfo, LogConfig, LogDirFailureChannel, LogOffsetMetadata, LogReadInfo, OffsetResultHolder, RecordValidationException, RemoteLogReadResult, RemoteStorageFetchInfo, UnifiedLog, VerificationGuard}
+import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchDataInfo, LeaderHwChange, LogAppendInfo, LogConfig, LogDirFailureChannel, LogOffsetMetadata, LogReadInfo, OffsetResultHolder, RecordValidationException, RemoteLogReadResult, RemoteStorageFetchInfo, UnifiedLog, VerificationGuard,PartitionListener}
 import org.apache.kafka.storage.log.metrics.BrokerTopicStats
 
 import java.io.File

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

@@ -28,7 +28,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/AssignmentStateTest.scala

@@ -16,15 +16,14 @@
   */
 package kafka.cluster
 
+import org.apache.kafka.storage.internals.log.SimpleAssignmentState
 import org.apache.kafka.common.PartitionState
 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 +87,7 @@ class AssignmentStateTest extends AbstractPartitionTest {
   @MethodSource(Array("parameters"))
   def testPartitionAssignmentStatus(isr: util.List[Integer], replicas: util.List[Integer],
                                     adding: util.List[Integer], removing: util.List[Integer],
-                                    original: util.List[Int], isUnderReplicated: Boolean): Unit = {
+                                    original: util.List[Integer], isUnderReplicated: Boolean): Unit = {
     val leaderState = new PartitionState()
       .setLeader(brokerId)
       .setLeaderEpoch(6)
@@ -105,7 +104,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(leaderState, offsetCheckpoints, None)
     assertEquals(isReassigning, partition.isReassigning)

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

@@ -39,7 +39,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}