KAFKA-14484: Move UnifiedLog to storage module (#19030)

Rewrite UnifiedLog in Java

Reviewers: Jun Rao <[email protected]>, Chia-Ping Tsai <[email protected]>

Author: mimaison

Diff for: checkstyle/suppressions.xml

@@ -355,11 +355,13 @@
 
     <!-- storage -->
     <suppress checks="CyclomaticComplexity"
-              files="(LogLoader|LogValidator|RemoteLogManagerConfig|RemoteLogManager).java"/>
+              files="(LogLoader|LogValidator|RemoteLogManagerConfig|RemoteLogManager|UnifiedLog).java"/>
     <suppress checks="NPathComplexity"
-              files="(LocalLog|LogLoader|LogValidator|RemoteLogManager|RemoteIndexCache).java"/>
+              files="(LocalLog|LogLoader|LogValidator|RemoteLogManager|RemoteIndexCache|UnifiedLog).java"/>
     <suppress checks="ParameterNumber"
-              files="(LogAppendInfo|LogLoader|RemoteLogManagerConfig).java"/>
+              files="(LogAppendInfo|LogLoader|RemoteLogManagerConfig|UnifiedLog).java"/>
+    <suppress checks="(ClassDataAbstractionCoupling|ClassFanOutComplexity)"
+              files="(UnifiedLog).java"/>
 
     <!-- benchmarks -->
     <suppress checks="(ClassDataAbstractionCoupling|ClassFanOutComplexity)"

Diff for: clients/src/main/java/org/apache/kafka/common/utils/LogContext.java

@@ -43,7 +43,14 @@ public LogContext() {
     }
 
     public Logger logger(Class<?> clazz) {
-        Logger logger = LoggerFactory.getLogger(clazz);
+        return logger(LoggerFactory.getLogger(clazz));
+    }
+
+    public Logger logger(String clazz) {
+        return logger(LoggerFactory.getLogger(clazz));
+    }
+
+    private Logger logger(Logger logger) {
         if (logger instanceof LocationAwareLogger) {
             return new LocationAwareKafkaLogger(logPrefix, (LocationAwareLogger) logger);
         } else {

Diff for: core/src/main/java/kafka/log/remote/RemoteLogManager.java

@@ -17,7 +17,6 @@
 package kafka.log.remote;
 
 import kafka.cluster.Partition;
-import kafka.log.UnifiedLog;
 import kafka.server.DelayedRemoteListOffsets;
 
 import org.apache.kafka.common.Endpoint;
@@ -88,6 +87,7 @@
 import org.apache.kafka.storage.internals.log.RemoteStorageThreadPool;
 import org.apache.kafka.storage.internals.log.TransactionIndex;
 import org.apache.kafka.storage.internals.log.TxnIndexSearchResult;
+import org.apache.kafka.storage.internals.log.UnifiedLog;
 import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
 
 import com.yammer.metrics.core.Timer;
@@ -138,12 +138,9 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
-import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import scala.jdk.javaapi.CollectionConverters;
-
 import static org.apache.kafka.server.config.ServerLogConfigs.LOG_DIR_CONFIG;
 import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_COMMON_CLIENT_PREFIX;
 import static org.apache.kafka.server.log.remote.quota.RLMQuotaManagerConfig.INACTIVE_SENSOR_EXPIRATION_TIME_SECONDS;
@@ -668,7 +665,7 @@ public AsyncOffsetReadFutureHolder<OffsetResultHolder.FileRecordsOrError> asyncO
             long timestamp,
             long startingOffset,
             LeaderEpochFileCache leaderEpochCache,
-            Supplier<Optional<FileRecords.TimestampAndOffset>> searchLocalLog) {
+            TimestampAndOffsetSupplier searchLocalLog) {
         CompletableFuture<OffsetResultHolder.FileRecordsOrError> taskFuture = new CompletableFuture<>();
         Future<Void> jobFuture = remoteStorageReaderThreadPool.submit(
                 new RemoteLogOffsetReader(this, topicPartition, timestamp, startingOffset, leaderEpochCache, searchLocalLog, result -> {
@@ -733,7 +730,7 @@ public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(TopicParti
                     && isRemoteSegmentWithinLeaderEpochs(rlsMetadata, unifiedLog.logEndOffset(), epochWithOffsets)
                     && rlsMetadata.state().equals(RemoteLogSegmentState.COPY_SEGMENT_FINISHED)) {
                     // cache to avoid race conditions
-                    List<LogSegment> segmentsCopy = new ArrayList<>(unifiedLog.logSegments());
+                    List<LogSegment> segmentsCopy = unifiedLog.logSegments();
                     if (segmentsCopy.isEmpty() || rlsMetadata.startOffset() < segmentsCopy.get(0).baseOffset()) {
                         // search in remote-log
                         return lookupTimestamp(rlsMetadata, timestamp, startingOffset);
@@ -904,7 +901,7 @@ private void maybeUpdateCopiedOffset(UnifiedLog log) throws RemoteStorageExcepti
          */
         List<EnrichedLogSegment> candidateLogSegments(UnifiedLog log, Long fromOffset, Long lastStableOffset) {
             List<EnrichedLogSegment> candidateLogSegments = new ArrayList<>();
-            List<LogSegment> segments = CollectionConverters.asJava(log.logSegments(fromOffset, Long.MAX_VALUE).toSeq());
+            List<LogSegment> segments = log.logSegments(fromOffset, Long.MAX_VALUE);
             if (!segments.isEmpty()) {
                 for (int idx = 1; idx < segments.size(); idx++) {
                     LogSegment previousSeg = segments.get(idx - 1);
@@ -1776,7 +1773,7 @@ private FetchDataInfo addAbortedTransactions(long startOffset,
 
         Consumer<List<AbortedTxn>> accumulator =
                 abortedTxns -> abortedTransactions.addAll(abortedTxns.stream()
-                        .map(AbortedTxn::asAbortedTransaction).collect(Collectors.toList()));
+                        .map(AbortedTxn::asAbortedTransaction).toList());
 
         long startTimeNs = time.nanoseconds();
         collectAbortedTransactions(startOffset, upperBoundOffset, segmentMetadata, accumulator, log);

Diff for: core/src/main/java/kafka/log/remote/RemoteLogOffsetReader.java

@@ -19,6 +19,7 @@
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.record.FileRecords;
 import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache;
+import org.apache.kafka.storage.internals.log.AsyncOffsetReader;
 import org.apache.kafka.storage.internals.log.OffsetResultHolder;
 
 import org.slf4j.Logger;
@@ -27,7 +28,6 @@
 import java.util.Optional;
 import java.util.concurrent.Callable;
 import java.util.function.Consumer;
-import java.util.function.Supplier;
 
 public class RemoteLogOffsetReader implements Callable<Void> {
     private static final Logger LOGGER = LoggerFactory.getLogger(RemoteLogOffsetReader.class);
@@ -36,15 +36,15 @@ public class RemoteLogOffsetReader implements Callable<Void> {
     private final long timestamp;
     private final long startingOffset;
     private final LeaderEpochFileCache leaderEpochCache;
-    private final Supplier<Optional<FileRecords.TimestampAndOffset>> searchInLocalLog;
+    private final AsyncOffsetReader.TimestampAndOffsetSupplier searchInLocalLog;
     private final Consumer<OffsetResultHolder.FileRecordsOrError> callback;
 
     public RemoteLogOffsetReader(RemoteLogManager rlm,
                                  TopicPartition tp,
                                  long timestamp,
                                  long startingOffset,
                                  LeaderEpochFileCache leaderEpochCache,
-                                 Supplier<Optional<FileRecords.TimestampAndOffset>> searchInLocalLog,
+                                 AsyncOffsetReader.TimestampAndOffsetSupplier searchInLocalLog,
                                  Consumer<OffsetResultHolder.FileRecordsOrError> callback) {
         this.rlm = rlm;
         this.tp = tp;
@@ -60,8 +60,11 @@ public Void call() throws Exception {
         OffsetResultHolder.FileRecordsOrError result;
         try {
             // If it is not found in remote storage, then search in the local storage starting with local log start offset.
-            Optional<FileRecords.TimestampAndOffset> timestampAndOffsetOpt = 
-                    rlm.findOffsetByTimestamp(tp, timestamp, startingOffset, leaderEpochCache).or(searchInLocalLog);
+            Optional<FileRecords.TimestampAndOffset> timestampAndOffsetOpt =
+                    rlm.findOffsetByTimestamp(tp, timestamp, startingOffset, leaderEpochCache);
+            if (timestampAndOffsetOpt.isEmpty()) {
+                timestampAndOffsetOpt = searchInLocalLog.get();
+            }
             result = new OffsetResultHolder.FileRecordsOrError(Optional.empty(), timestampAndOffsetOpt);
         } catch (Exception e) {
             // NOTE: All the exceptions from the secondary storage are caught instead of only the KafkaException.

Diff for: core/src/main/java/kafka/server/TierStateMachine.java

@@ -18,7 +18,6 @@
 package kafka.server;
 
 import kafka.cluster.Partition;
-import kafka.log.UnifiedLog;
 import kafka.log.remote.RemoteLogManager;
 
 import org.apache.kafka.common.KafkaException;
@@ -36,6 +35,7 @@
 import org.apache.kafka.storage.internals.checkpoint.LeaderEpochCheckpointFile;
 import org.apache.kafka.storage.internals.log.EpochEntry;
 import org.apache.kafka.storage.internals.log.LogFileUtils;
+import org.apache.kafka.storage.internals.log.UnifiedLog;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -53,6 +53,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import scala.Option;
 import scala.jdk.javaapi.CollectionConverters;
@@ -186,7 +187,7 @@ private Long buildRemoteLogAuxState(TopicPartition topicPartition,
                                         Long leaderLogStartOffset,
                                         UnifiedLog unifiedLog) throws IOException, RemoteStorageException {
 
-        if (!unifiedLog.remoteStorageSystemEnable() || !unifiedLog.config().remoteStorageEnable()) {
+        if (!unifiedLog.remoteLogEnabled()) {
             // If the tiered storage is not enabled throw an exception back so that it will retry until the tiered storage
             // is set as expected.
             throw new RemoteStorageException("Couldn't build the state from remote store for partition " + topicPartition + ", as remote log storage is not yet enabled");
@@ -240,7 +241,7 @@ private Long buildRemoteLogAuxState(TopicPartition topicPartition,
 
         // Truncate the existing local log before restoring the leader epoch cache and producer snapshots.
         Partition partition = replicaMgr.getPartitionOrException(topicPartition);
-        partition.truncateFullyAndStartAt(nextOffset, useFutureLog, Option.apply(leaderLogStartOffset));
+        partition.truncateFullyAndStartAt(nextOffset, useFutureLog, Optional.of(leaderLogStartOffset));
         // Increment start offsets
         unifiedLog.maybeIncrementLogStartOffset(leaderLogStartOffset, LeaderOffsetIncremented);
         unifiedLog.maybeIncrementLocalLogStartOffset(nextOffset, LeaderOffsetIncremented);

Diff for: core/src/main/scala/kafka/cluster/Partition.scala

@@ -16,6 +16,7 @@
  */
 package kafka.cluster
 
+import java.lang.{Long => JLong}
 import java.util.concurrent.locks.ReentrantReadWriteLock
 import java.util.Optional
 import java.util.concurrent.{CompletableFuture, CopyOnWriteArrayList}
@@ -39,7 +40,7 @@ import org.apache.kafka.common.requests.OffsetsForLeaderEpochResponse.{UNDEFINED
 import org.apache.kafka.common.utils.Time
 import org.apache.kafka.metadata.{LeaderAndIsr, LeaderRecoveryState}
 import org.apache.kafka.server.common.RequestLocal
-import org.apache.kafka.storage.internals.log.{AppendOrigin, AsyncOffsetReader, FetchDataInfo, LeaderHwChange, LogAppendInfo, LogOffsetMetadata, LogOffsetSnapshot, LogOffsetsListener, LogReadInfo, LogStartOffsetIncrementReason, OffsetResultHolder, VerificationGuard}
+import org.apache.kafka.storage.internals.log.{AppendOrigin, AsyncOffsetReader, FetchDataInfo, LeaderHwChange, LogAppendInfo, LogOffsetMetadata, LogOffsetSnapshot, LogOffsetsListener, LogReadInfo, LogStartOffsetIncrementReason, OffsetResultHolder, UnifiedLog, VerificationGuard}
 import org.apache.kafka.server.metrics.KafkaMetricsGroup
 import org.apache.kafka.server.purgatory.{DelayedOperationPurgatory, TopicPartitionOperationKey}
 import org.apache.kafka.server.share.fetch.DelayedShareFetchPartitionKey
@@ -49,7 +50,7 @@ import org.slf4j.event.Level
 
 import scala.collection.Seq
 import scala.jdk.CollectionConverters._
-import scala.jdk.OptionConverters.RichOption
+import scala.jdk.OptionConverters.{RichOption, RichOptional}
 import scala.jdk.javaapi.OptionConverters
 
 /**
@@ -494,7 +495,7 @@ class Partition(val topicPartition: TopicPartition,
     logManager.initializingLog(topicPartition)
     var maybeLog: Option[UnifiedLog] = None
     try {
-      val log = logManager.getOrCreateLog(topicPartition, isNew, isFutureReplica, topicId, targetLogDirectoryId)
+      val log = logManager.getOrCreateLog(topicPartition, isNew, isFutureReplica, topicId.toJava, targetLogDirectoryId)
       if (!isFutureReplica) log.setLogOffsetsListener(logOffsetsListener)
       maybeLog = Some(log)
       updateHighWatermark(log)
@@ -593,7 +594,7 @@ class Partition(val topicPartition: TopicPartition,
    */
   def topicId: Option[Uuid] = {
     if (_topicId.isEmpty || _topicId.contains(Uuid.ZERO_UUID)) {
-      _topicId = this.log.orElse(logManager.getLog(topicPartition)).flatMap(_.topicId)
+      _topicId = this.log.orElse(logManager.getLog(topicPartition)).flatMap(_.topicId.toScala)
     }
     _topicId
   }
@@ -1170,7 +1171,7 @@ class Partition(val topicPartition: TopicPartition,
       }
     }
 
-    leaderLog.maybeIncrementHighWatermark(newHighWatermark) match {
+    leaderLog.maybeIncrementHighWatermark(newHighWatermark).toScala match {
       case Some(oldHighWatermark) =>
         debug(s"High watermark updated from $oldHighWatermark to $newHighWatermark")
         true
@@ -1369,8 +1370,7 @@ class Partition(val topicPartition: TopicPartition,
               s"live replica(s) broker.id are : $inSyncReplicaIds")
           }
 
-          val info = leaderLog.appendAsLeader(records, leaderEpoch = this.leaderEpoch, origin,
-            requestLocal, verificationGuard)
+          val info = leaderLog.appendAsLeader(records, this.leaderEpoch, origin, requestLocal, verificationGuard)
 
           // we may need to increment high watermark since ISR could be down to 1
           (info, maybeIncrementLeaderHW(leaderLog))
@@ -1622,7 +1622,7 @@ class Partition(val topicPartition: TopicPartition,
       case Some(producers) =>
         producerState
           .setErrorCode(Errors.NONE.code)
-          .setActiveProducers(producers.asJava)
+          .setActiveProducers(producers)
       case None =>
         producerState
           .setErrorCode(Errors.NOT_LEADER_OR_FOLLOWER.code)
@@ -1696,7 +1696,7 @@ class Partition(val topicPartition: TopicPartition,
     */
   def truncateFullyAndStartAt(newOffset: Long,
                               isFuture: Boolean,
-                              logStartOffsetOpt: Option[Long] = None): Unit = {
+                              logStartOffsetOpt: Optional[JLong] = Optional.empty): Unit = {
     // The read lock is needed to prevent the follower replica from being truncated while ReplicaAlterDirThread
     // is executing maybeReplaceCurrentWithFutureReplica() to replace follower replica with the future replica.
     inReadLock(leaderIsrUpdateLock) {
@@ -1724,7 +1724,7 @@ class Partition(val topicPartition: TopicPartition,
       val localLogOrError = getLocalLog(currentLeaderEpoch, fetchOnlyFromLeader)
       localLogOrError match {
         case Left(localLog) =>
-          localLog.endOffsetForEpoch(leaderEpoch) match {
+          localLog.endOffsetForEpoch(leaderEpoch).toScala match {
             case Some(epochAndOffset) => new EpochEndOffset()
               .setPartition(partitionId)
               .setErrorCode(Errors.NONE.code)

Diff for: core/src/main/scala/kafka/coordinator/group/CoordinatorLoaderImpl.scala

@@ -103,12 +103,7 @@ class CoordinatorLoaderImpl[T](
           var numRecords = 0L
           var numBytes = 0L
           while (currentOffset < logEndOffset && readAtLeastOneRecord && isRunning.get) {
-            val fetchDataInfo = log.read(
-              startOffset = currentOffset,
-              maxLength = loadBufferSize,
-              isolation = FetchIsolation.LOG_END,
-              minOneMessage = true
-            )
+            val fetchDataInfo = log.read(currentOffset, loadBufferSize, FetchIsolation.LOG_END, true)
 
             readAtLeastOneRecord = fetchDataInfo.records.sizeInBytes > 0
 

Diff for: core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala

@@ -601,10 +601,7 @@ class GroupMetadataManager(brokerId: Int,
         var readAtLeastOneRecord = true
 
         while (currOffset < logEndOffset && readAtLeastOneRecord && !shuttingDown.get()) {
-          val fetchDataInfo = log.read(currOffset,
-            maxLength = config.loadBufferSize,
-            isolation = FetchIsolation.LOG_END,
-            minOneMessage = true)
+          val fetchDataInfo = log.read(currOffset, config.loadBufferSize, FetchIsolation.LOG_END, true)
 
           readAtLeastOneRecord = fetchDataInfo.records.sizeInBytes > 0
 

Diff for: core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala

@@ -438,10 +438,7 @@ class TransactionStateManager(brokerId: Int,
           while (currOffset < logEndOffset && readAtLeastOneRecord && !shuttingDown.get() && inReadLock(stateLock) {
             loadingPartitions.exists { idAndEpoch: TransactionPartitionAndLeaderEpoch =>
               idAndEpoch.txnPartitionId == topicPartition.partition && idAndEpoch.coordinatorEpoch == coordinatorEpoch}}) {
-            val fetchDataInfo = log.read(currOffset,
-              maxLength = config.transactionLogLoadBufferSize,
-              isolation = FetchIsolation.LOG_END,
-              minOneMessage = true)
+            val fetchDataInfo = log.read(currOffset, config.transactionLogLoadBufferSize, FetchIsolation.LOG_END, true)
 
             readAtLeastOneRecord = fetchDataInfo.records.sizeInBytes > 0