KAFKA-14477: Move LogValidator and related to storage module

build.gradle

@@ -2413,6 +2413,7 @@ project(':jmh-benchmarks') {
     implementation project(':server-common')
     implementation project(':clients')
     implementation project(':metadata')
+    implementation project(':storage')
     implementation project(':streams')
     implementation project(':core')
     implementation project(':clients').sourceSets.test.output

checkstyle/suppressions.xml

@@ -318,9 +318,13 @@
     <suppress checks="AvoidStarImport"
               files="MetadataVersionTest.java"/>
 
-    <!-- Storage -->
-    <suppress checks="(CyclomaticComplexity|ParameterNumber)"
-              files="(RemoteLogManagerConfig).java"/>
+    <!-- storage -->
+    <suppress checks="CyclomaticComplexity"
+              files="(LogValidator|RemoteLogManagerConfig).java"/>
+    <suppress checks="NPathComplexity"
+              files="LogValidator.java"/>
+    <suppress checks="ParameterNumber"
+              files="RemoteLogManagerConfig.java"/>
 
     <!-- benchmarks -->
     <suppress checks="(ClassDataAbstractionCoupling|ClassFanOutComplexity)"

clients/src/main/java/org/apache/kafka/common/utils/PrimitiveRef.java

@@ -26,11 +26,34 @@ public static IntRef ofInt(int value) {
         return new IntRef(value);
     }
 
+    public static LongRef ofLong(long value) {
+        return new LongRef(value);
+    }
+
     public static class IntRef {
         public int value;
 
         IntRef(int value) {
             this.value = value;
         }
+
+        @Override
+        public String toString() {
+            return "IntRef(" + value + ")";
+        }
+
+    }
+
+    public static class LongRef {
+        public long value;
+
+        LongRef(long value) {
+            this.value = value;
+        }
+
+        @Override
+        public String toString() {
+            return "LongRef(" + value + ")";
+        }
     }
 }

clients/src/test/java/org/apache/kafka/common/utils/PrimitiveRefTest.java

@@ -0,0 +1,43 @@
+/*
+ * 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.common.utils;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class PrimitiveRefTest {
+
+    @Test
+    public void testIntRef() {
+        PrimitiveRef.IntRef ref = PrimitiveRef.ofInt(3);
+        assertEquals(3, ref.value++);
+        assertEquals(4, ref.value);
+        assertEquals(5, ++ref.value);
+        assertEquals(5, ref.value);
+    }
+
+    @Test
+    public void testLongRef() {
+        PrimitiveRef.LongRef ref = PrimitiveRef.ofLong(5L);
+        assertEquals(5L, ref.value++);
+        assertEquals(6L, ref.value);
+        assertEquals(7L, ++ref.value);
+        assertEquals(7L, ref.value);
+    }
+
+}

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

@@ -44,6 +44,7 @@ import org.apache.kafka.common.utils.Time
 import org.apache.kafka.common.{IsolationLevel, TopicPartition, Uuid}
 import org.apache.kafka.metadata.LeaderRecoveryState
 import org.apache.kafka.server.common.MetadataVersion
+import org.apache.kafka.server.log.internals.AppendOrigin
 
 import scala.collection.{Map, Seq}
 import scala.jdk.CollectionConverters._

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

@@ -24,11 +24,9 @@ import java.util.Optional
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.locks.ReentrantLock
 import java.util.concurrent.{ConcurrentHashMap, TimeUnit}
-
 import com.yammer.metrics.core.Gauge
 import kafka.common.OffsetAndMetadata
 import kafka.internals.generated.{GroupMetadataValue, OffsetCommitKey, OffsetCommitValue, GroupMetadataKey => GroupMetadataKeyData}
-import kafka.log.AppendOrigin
 import kafka.metrics.KafkaMetricsGroup
 import kafka.server.{FetchLogEnd, ReplicaManager, RequestLocal}
 import kafka.utils.CoreUtils.inLock
@@ -48,6 +46,7 @@ import org.apache.kafka.common.utils.{Time, Utils}
 import org.apache.kafka.common.{KafkaException, MessageFormatter, TopicPartition}
 import org.apache.kafka.server.common.MetadataVersion
 import org.apache.kafka.server.common.MetadataVersion.{IBP_0_10_1_IV0, IBP_2_1_IV0, IBP_2_1_IV1, IBP_2_3_IV0}
+import org.apache.kafka.server.log.internals.AppendOrigin
 
 import scala.collection._
 import scala.collection.mutable.ArrayBuffer
@@ -331,7 +330,7 @@ class GroupMetadataManager(brokerId: Int,
       timeout = config.offsetCommitTimeoutMs.toLong,
       requiredAcks = config.offsetCommitRequiredAcks,
       internalTopicsAllowed = true,
-      origin = AppendOrigin.Coordinator,
+      origin = AppendOrigin.COORDINATOR,
       entriesPerPartition = records,
       delayedProduceLock = Some(group.lock),
       responseCallback = callback,
@@ -890,7 +889,7 @@ class GroupMetadataManager(brokerId: Int,
                 // do not need to require acks since even if the tombstone is lost,
                 // it will be appended again in the next purge cycle
                 val records = MemoryRecords.withRecords(magicValue, 0L, compressionType, timestampType, tombstones.toArray: _*)
-                partition.appendRecordsToLeader(records, origin = AppendOrigin.Coordinator, requiredAcks = 0,
+                partition.appendRecordsToLeader(records, origin = AppendOrigin.COORDINATOR, requiredAcks = 0,
                   requestLocal = requestLocal)
 
                 offsetsRemoved += removedOffsets.size

core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala

@@ -16,12 +16,13 @@
  */
 package kafka.coordinator.transaction
 
+import kafka.log.LogConfig
+
 import java.nio.ByteBuffer
 import java.util.Properties
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.locks.ReentrantReadWriteLock
-import kafka.log.{AppendOrigin, LogConfig}
 import kafka.server.{Defaults, FetchLogEnd, ReplicaManager, RequestLocal}
 import kafka.utils.CoreUtils.{inReadLock, inWriteLock}
 import kafka.utils.{Logging, Pool, Scheduler}
@@ -36,6 +37,7 @@ import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
 import org.apache.kafka.common.requests.TransactionResult
 import org.apache.kafka.common.utils.{Time, Utils}
 import org.apache.kafka.common.{KafkaException, TopicPartition}
+import org.apache.kafka.server.log.internals.AppendOrigin
 import org.apache.kafka.server.record.BrokerCompressionType
 
 import scala.jdk.CollectionConverters._
@@ -282,7 +284,7 @@ class TransactionStateManager(brokerId: Int,
         config.requestTimeoutMs,
         TransactionLog.EnforcedRequiredAcks,
         internalTopicsAllowed = true,
-        origin = AppendOrigin.Coordinator,
+        origin = AppendOrigin.COORDINATOR,
         entriesPerPartition = Map(transactionPartition -> tombstoneRecords),
         removeFromCacheCallback,
         requestLocal = RequestLocal.NoCaching)
@@ -761,7 +763,7 @@ class TransactionStateManager(brokerId: Int,
                 newMetadata.txnTimeoutMs.toLong,
                 TransactionLog.EnforcedRequiredAcks,
                 internalTopicsAllowed = true,
-                origin = AppendOrigin.Coordinator,
+                origin = AppendOrigin.COORDINATOR,
                 recordsPerPartition,
                 updateCacheCallback,
                 requestLocal = requestLocal)

core/src/main/scala/kafka/log/LogSegment.scala

@@ -32,7 +32,7 @@ import org.apache.kafka.common.errors.CorruptRecordException
 import org.apache.kafka.common.record.FileRecords.{LogOffsetPosition, TimestampAndOffset}
 import org.apache.kafka.common.record._
 import org.apache.kafka.common.utils.{BufferSupplier, Time}
-import org.apache.kafka.server.log.internals.{AbortedTxn, CompletedTxn, LazyIndex, OffsetIndex, OffsetPosition, TimeIndex, TimestampOffset, TransactionIndex, TxnIndexSearchResult}
+import org.apache.kafka.server.log.internals.{AbortedTxn, AppendOrigin, CompletedTxn, LazyIndex, OffsetIndex, OffsetPosition, TimeIndex, TimestampOffset, TransactionIndex, TxnIndexSearchResult}
 
 import java.util.Optional
 import scala.jdk.CollectionConverters._
@@ -248,7 +248,7 @@ class LogSegment private[log] (val log: FileRecords,
   private def updateProducerState(producerStateManager: ProducerStateManager, batch: RecordBatch): Unit = {
     if (batch.hasProducerId) {
       val producerId = batch.producerId
-      val appendInfo = producerStateManager.prepareUpdate(producerId, origin = AppendOrigin.Replication)
+      val appendInfo = producerStateManager.prepareUpdate(producerId, origin = AppendOrigin.REPLICATION)
       val maybeCompletedTxn = appendInfo.append(batch, firstOffsetMetadataOpt = None)
       producerStateManager.update(appendInfo)
       maybeCompletedTxn.foreach { completedTxn =>