You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ij...@apache.org on 2023/10/16 13:37:41 UTC
[kafka] branch trunk updated: KAFKA-14481: Move LogSegment/LogSegments to storage module (#14529)
This is an automated email from the ASF dual-hosted git repository.
ijuma pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 1073d434ec9 KAFKA-14481: Move LogSegment/LogSegments to storage module (#14529)
1073d434ec9 is described below
commit 1073d434ec98e64afca1979cd18d3244b133e688
Author: Ismael Juma <is...@juma.me.uk>
AuthorDate: Mon Oct 16 06:37:30 2023 -0700
KAFKA-14481: Move LogSegment/LogSegments to storage module (#14529)
A few notes:
* Delete a few methods from `UnifiedLog` that were simply invoking the related method in `LogFileUtils`
* Fix `CoreUtils.swallow` to use the passed in `logging`
* Fix `LogCleanerParameterizedIntegrationTest` to close `log` before reopening
* Minor tweaks in `LogSegment` for readability
For broader context on this change, please check:
* KAFKA-14470: Move log layer to storage module
Reviewers: Divij Vaidya <di...@amazon.com>, Satish Duggana <sa...@apache.org>
---
.../kafka/common/record/AbstractRecords.java | 12 +
.../org/apache/kafka/common/record/Records.java | 8 +
.../java/org/apache/kafka/common/utils/Utils.java | 52 +-
.../org/apache/kafka/common/utils/UtilsTest.java | 77 ++
.../java/kafka/log/remote/RemoteLogManager.java | 8 +-
core/src/main/scala/kafka/log/LocalLog.scala | 57 +-
core/src/main/scala/kafka/log/LogCleaner.scala | 9 +-
.../main/scala/kafka/log/LogCleanerManager.scala | 14 +-
core/src/main/scala/kafka/log/LogLoader.scala | 64 +-
core/src/main/scala/kafka/log/LogManager.scala | 2 +-
core/src/main/scala/kafka/log/LogSegment.scala | 693 ----------------
core/src/main/scala/kafka/log/LogSegments.scala | 268 -------
core/src/main/scala/kafka/log/UnifiedLog.scala | 82 +-
core/src/main/scala/kafka/utils/CoreUtils.scala | 35 +-
.../kafka/log/remote/RemoteLogManagerTest.java | 39 +-
.../api/GroupCoordinatorIntegrationTest.scala | 4 +-
.../server/DynamicBrokerReconfigurationTest.scala | 2 +-
.../scala/kafka/raft/KafkaMetadataLogTest.scala | 5 +-
.../unit/kafka/cluster/PartitionLockTest.scala | 5 +-
.../scala/unit/kafka/cluster/PartitionTest.scala | 4 +-
.../log/AbstractLogCleanerIntegrationTest.scala | 5 +
.../test/scala/unit/kafka/log/LocalLogTest.scala | 40 +-
.../unit/kafka/log/LogCleanerIntegrationTest.scala | 8 +-
.../kafka/log/LogCleanerLagIntegrationTest.scala | 2 +-
.../unit/kafka/log/LogCleanerManagerTest.scala | 8 +-
.../LogCleanerParameterizedIntegrationTest.scala | 15 +-
.../test/scala/unit/kafka/log/LogCleanerTest.scala | 101 +--
.../scala/unit/kafka/log/LogConcurrencyTest.scala | 2 +-
.../test/scala/unit/kafka/log/LogLoaderTest.scala | 87 +-
.../test/scala/unit/kafka/log/LogManagerTest.scala | 16 +-
.../test/scala/unit/kafka/log/LogSegmentTest.scala | 121 ++-
.../scala/unit/kafka/log/LogSegmentsTest.scala | 65 +-
.../test/scala/unit/kafka/log/LogTestUtils.scala | 21 +-
.../test/scala/unit/kafka/log/UnifiedLogTest.scala | 64 +-
.../kafka/server/DynamicConfigChangeTest.scala | 2 +-
.../scala/unit/kafka/server/LogOffsetTest.scala | 12 +-
.../unit/kafka/server/ReplicaManagerTest.scala | 4 +-
.../server/epoch/LeaderEpochIntegrationTest.scala | 5 +-
.../scala/unit/kafka/utils/CoreUtilsTest.scala | 51 --
.../scala/unit/kafka/utils/SchedulerTest.scala | 8 +-
gradle/spotbugs-exclude.xml | 5 +-
.../kafka/storage/internals/log/LazyIndex.java | 3 +-
.../kafka/storage/internals/log/LogSegment.java | 889 +++++++++++++++++++++
.../log/LogSegmentOffsetOverflowException.java | 22 +-
.../kafka/storage/internals/log/LogSegments.java | 355 ++++++++
45 files changed, 1857 insertions(+), 1494 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/record/AbstractRecords.java b/clients/src/main/java/org/apache/kafka/common/record/AbstractRecords.java
index 265ef5b72b5..5fddaae4062 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/AbstractRecords.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/AbstractRecords.java
@@ -22,6 +22,7 @@ import org.apache.kafka.common.utils.Utils;
import java.nio.ByteBuffer;
import java.util.Iterator;
+import java.util.Optional;
public abstract class AbstractRecords implements Records {
@@ -44,6 +45,17 @@ public abstract class AbstractRecords implements Records {
return iterator.next();
}
+ @Override
+ public Optional<RecordBatch> lastBatch() {
+ Iterator<? extends RecordBatch> iterator = batches().iterator();
+
+ RecordBatch batch = null;
+ while (iterator.hasNext())
+ batch = iterator.next();
+
+ return Optional.ofNullable(batch);
+ }
+
/**
* Get an iterator over the deep records.
* @return An iterator over the records
diff --git a/clients/src/main/java/org/apache/kafka/common/record/Records.java b/clients/src/main/java/org/apache/kafka/common/record/Records.java
index c18679afd88..9b2352a9326 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/Records.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/Records.java
@@ -20,6 +20,7 @@ import org.apache.kafka.common.utils.AbstractIterator;
import org.apache.kafka.common.utils.Time;
import java.util.Iterator;
+import java.util.Optional;
/**
@@ -70,6 +71,13 @@ public interface Records extends TransferableRecords {
*/
AbstractIterator<? extends RecordBatch> batchIterator();
+ /**
+ * Return the last record batch if non-empty or an empty `Optional` otherwise.
+ *
+ * Note that this requires iterating over all the record batches and hence it's expensive.
+ */
+ Optional<RecordBatch> lastBatch();
+
/**
* Check whether all batches in this buffer have a certain magic value.
* @param magic The magic value to check
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
index bf9d0d8f16d..6a0913d3c2d 100644
--- a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
+++ b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
@@ -71,6 +71,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Properties;
import java.util.Set;
+import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;
@@ -1042,14 +1043,20 @@ public final class Utils {
if (exception != null)
throw exception;
}
- public static void swallow(final Logger log, final Level level, final String what, final Runnable code) {
+
+ @FunctionalInterface
+ public interface SwallowAction {
+ void run() throws Throwable;
+ }
+
+ public static void swallow(final Logger log, final Level level, final String what, final SwallowAction code) {
swallow(log, level, what, code, null);
}
/**
* Run the supplied code. If an exception is thrown, it is swallowed and registered to the firstException parameter.
*/
- public static void swallow(final Logger log, final Level level, final String what, final Runnable code,
+ public static void swallow(final Logger log, final Level level, final String what, final SwallowAction code,
final AtomicReference<Throwable> firstException) {
if (code != null) {
try {
@@ -1102,11 +1109,26 @@ public final class Utils {
* use a method reference from it.
*/
public static void closeQuietly(AutoCloseable closeable, String name) {
+ closeQuietly(closeable, name, log);
+ }
+
+ /**
+ * Closes {@code closeable} and if an exception is thrown, it is logged with the provided logger at the WARN level.
+ * <b>Be cautious when passing method references as an argument.</b> For example:
+ * <p>
+ * {@code closeQuietly(task::stop, "source task");}
+ * <p>
+ * Although this method gracefully handles null {@link AutoCloseable} objects, attempts to take a method
+ * reference from a null object will result in a {@link NullPointerException}. In the example code above,
+ * it would be the caller's responsibility to ensure that {@code task} was non-null before attempting to
+ * use a method reference from it.
+ */
+ public static void closeQuietly(AutoCloseable closeable, String name, Logger logger) {
if (closeable != null) {
try {
closeable.close();
} catch (Throwable t) {
- log.warn("Failed to close {} with type {}", name, closeable.getClass().getName(), t);
+ logger.warn("Failed to close {} with type {}", name, closeable.getClass().getName(), t);
}
}
}
@@ -1143,6 +1165,30 @@ public final class Utils {
for (AutoCloseable closeable : closeables) closeQuietly(closeable, name, firstException);
}
+ /**
+ * Invokes every function in `all` even if one or more functions throws an exception.
+ *
+ * If any of the functions throws an exception, the first one will be rethrown at the end with subsequent exceptions
+ * added as suppressed exceptions.
+ */
+ // Note that this is a generalised version of `closeAll`. We could potentially make it more general by
+ // changing the signature to `public <R> List<R> tryAll(all: List[Callable<R>])`
+ public static void tryAll(List<Callable<Void>> all) throws Throwable {
+ Throwable exception = null;
+ for (Callable call : all) {
+ try {
+ call.call();
+ } catch (Throwable t) {
+ if (exception != null)
+ exception.addSuppressed(t);
+ else
+ exception = t;
+ }
+ }
+ if (exception != null)
+ throw exception;
+ }
+
/**
* A cheap way to deterministically convert a number to a positive value. When the input is
* positive, the original value is returned. When the input number is negative, the returned
diff --git a/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java b/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
index 0c0bb6a7e53..467dd53ac76 100755
--- a/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
@@ -50,6 +50,7 @@ import java.util.Properties;
import java.util.Random;
import java.util.Set;
import java.util.TreeSet;
+import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
@@ -64,6 +65,7 @@ import static org.apache.kafka.common.utils.Utils.formatBytes;
import static org.apache.kafka.common.utils.Utils.getHost;
import static org.apache.kafka.common.utils.Utils.getPort;
import static org.apache.kafka.common.utils.Utils.intersection;
+import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkSet;
import static org.apache.kafka.common.utils.Utils.murmur2;
import static org.apache.kafka.common.utils.Utils.union;
@@ -952,4 +954,79 @@ public class UtilsTest {
assertEquals(expected, actual);
}
+ @Test
+ public void testTryAll() throws Throwable {
+ Map<String, Object> recorded = new HashMap<>();
+
+ Utils.tryAll(asList(
+ recordingCallable(recorded, "valid-0", null),
+ recordingCallable(recorded, null, new TestException("exception-1")),
+ recordingCallable(recorded, "valid-2", null),
+ recordingCallable(recorded, null, new TestException("exception-3"))
+ ));
+ Map<String, Object> expected = Utils.mkMap(
+ mkEntry("valid-0", "valid-0"),
+ mkEntry("exception-1", new TestException("exception-1")),
+ mkEntry("valid-2", "valid-2"),
+ mkEntry("exception-3", new TestException("exception-3"))
+ );
+ assertEquals(expected, recorded);
+
+ recorded.clear();
+ Utils.tryAll(asList(
+ recordingCallable(recorded, "valid-0", null),
+ recordingCallable(recorded, "valid-1", null)
+ ));
+ expected = Utils.mkMap(
+ mkEntry("valid-0", "valid-0"),
+ mkEntry("valid-1", "valid-1")
+ );
+ assertEquals(expected, recorded);
+
+ recorded.clear();
+ Utils.tryAll(asList(
+ recordingCallable(recorded, null, new TestException("exception-0")),
+ recordingCallable(recorded, null, new TestException("exception-1")))
+ );
+ expected = Utils.mkMap(
+ mkEntry("exception-0", new TestException("exception-0")),
+ mkEntry("exception-1", new TestException("exception-1"))
+ );
+ assertEquals(expected, recorded);
+ }
+
+ private Callable<Void> recordingCallable(Map<String, Object> recordingMap, String success, TestException failure) {
+ return () -> {
+ if (success == null)
+ recordingMap.put(failure.key, failure);
+ else if (failure == null)
+ recordingMap.put(success, success);
+ else
+ throw new IllegalArgumentException("Either `success` or `failure` must be null, but both are non-null.");
+
+ return null;
+ };
+ }
+
+ private class TestException extends Exception {
+ final String key;
+ TestException(String key) {
+ this.key = key;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (o == null || getClass() != o.getClass())
+ return false;
+ TestException that = (TestException) o;
+ return key.equals(that.key);
+ }
+
+ @Override
+ public int hashCode() {
+ return key.hashCode();
+ }
+ }
}
diff --git a/core/src/main/java/kafka/log/remote/RemoteLogManager.java b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
index ebbfa9b28d7..e3923afba1a 100644
--- a/core/src/main/java/kafka/log/remote/RemoteLogManager.java
+++ b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
@@ -19,7 +19,6 @@ package kafka.log.remote;
import com.yammer.metrics.core.Gauge;
import kafka.cluster.EndPoint;
import kafka.cluster.Partition;
-import kafka.log.LogSegment;
import kafka.log.UnifiedLog;
import kafka.server.BrokerTopicStats;
import kafka.server.KafkaConfig;
@@ -62,6 +61,7 @@ import org.apache.kafka.storage.internals.log.EpochEntry;
import org.apache.kafka.storage.internals.log.FetchDataInfo;
import org.apache.kafka.storage.internals.log.FetchIsolation;
import org.apache.kafka.storage.internals.log.LogOffsetMetadata;
+import org.apache.kafka.storage.internals.log.LogSegment;
import org.apache.kafka.storage.internals.log.OffsetIndex;
import org.apache.kafka.storage.internals.log.OffsetPosition;
import org.apache.kafka.storage.internals.log.RemoteIndexCache;
@@ -733,8 +733,8 @@ public class RemoteLogManager implements Closeable {
remoteLogMetadataManager.addRemoteLogSegmentMetadata(copySegmentStartedRlsm).get();
ByteBuffer leaderEpochsIndex = getLeaderEpochCheckpoint(log, -1, nextSegmentBaseOffset).readAsByteBuffer();
- LogSegmentData segmentData = new LogSegmentData(logFile.toPath(), toPathIfExists(segment.lazyOffsetIndex().get().file()),
- toPathIfExists(segment.lazyTimeIndex().get().file()), Optional.ofNullable(toPathIfExists(segment.txnIndex().file())),
+ LogSegmentData segmentData = new LogSegmentData(logFile.toPath(), toPathIfExists(segment.offsetIndex().file()),
+ toPathIfExists(segment.timeIndex().file()), Optional.ofNullable(toPathIfExists(segment.txnIndex().file())),
producerStateSnapshotFile.toPath(), leaderEpochsIndex);
brokerTopicStats.topicStats(log.topicPartition().topic()).remoteCopyRequestRate().mark();
brokerTopicStats.allTopicsStats().remoteCopyRequestRate().mark();
@@ -1374,7 +1374,7 @@ public class RemoteLogManager implements Closeable {
}
// Search in local segments
- collectAbortedTransactionInLocalSegments(startOffset, upperBoundOffset, accumulator, JavaConverters.asJavaIterator(log.logSegments().iterator()));
+ collectAbortedTransactionInLocalSegments(startOffset, upperBoundOffset, accumulator, log.logSegments().iterator());
}
private void collectAbortedTransactionInLocalSegments(long startOffset,
diff --git a/core/src/main/scala/kafka/log/LocalLog.scala b/core/src/main/scala/kafka/log/LocalLog.scala
index 6dfd92461c9..6f6679f1ebd 100644
--- a/core/src/main/scala/kafka/log/LocalLog.scala
+++ b/core/src/main/scala/kafka/log/LocalLog.scala
@@ -24,15 +24,17 @@ import org.apache.kafka.common.record.MemoryRecords
import org.apache.kafka.common.utils.{Time, Utils}
import org.apache.kafka.common.{KafkaException, TopicPartition}
import org.apache.kafka.server.util.Scheduler
-import org.apache.kafka.storage.internals.log.{AbortedTxn, FetchDataInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetMetadata, OffsetPosition}
+import org.apache.kafka.storage.internals.log.{AbortedTxn, FetchDataInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetMetadata, LogSegment, LogSegments, OffsetPosition}
import java.io.{File, IOException}
import java.nio.file.Files
+import java.util
import java.util.concurrent.atomic.AtomicLong
import java.util.regex.Pattern
import java.util.{Collections, Optional}
import scala.collection.mutable.{ArrayBuffer, ListBuffer}
import scala.collection.{Seq, immutable}
+import scala.compat.java8.OptionConverters._
import scala.jdk.CollectionConverters._
/**
@@ -171,9 +173,9 @@ class LocalLog(@volatile private var _dir: File,
val currentRecoveryPoint = recoveryPoint
if (currentRecoveryPoint <= offset) {
val segmentsToFlush = segments.values(currentRecoveryPoint, offset)
- segmentsToFlush.foreach(_.flush())
+ segmentsToFlush.forEach(_.flush())
// If there are any new segments, we need to flush the parent directory for crash consistency.
- if (segmentsToFlush.exists(_.baseOffset >= currentRecoveryPoint)) {
+ if (segmentsToFlush.stream().anyMatch(_.baseOffset >= currentRecoveryPoint)) {
// The directory might be renamed concurrently for topic deletion, which may cause NoSuchFileException here.
// Since the directory is to be deleted anyways, we just swallow NoSuchFileException and let it go.
Utils.flushDirIfExists(dir.toPath)
@@ -248,8 +250,8 @@ class LocalLog(@volatile private var _dir: File,
*/
private[log] def deleteAllSegments(): Iterable[LogSegment] = {
maybeHandleIOException(s"Error while deleting all segments for $topicPartition in dir ${dir.getParent}") {
- val deletableSegments = List[LogSegment]() ++ segments.values
- removeAndDeleteSegments(segments.values, asyncDelete = false, LogDeletion(this))
+ val deletableSegments = new util.ArrayList(segments.values).asScala
+ removeAndDeleteSegments(segments.values.asScala, asyncDelete = false, LogDeletion(this))
isMemoryMappedBufferClosed = true
deletableSegments
}
@@ -341,11 +343,11 @@ class LocalLog(@volatile private var _dir: File,
segmentToDelete.changeFileSuffixes("", LogFileUtils.DELETED_FILE_SUFFIX)
val newSegment = LogSegment.open(dir,
- baseOffset = newOffset,
+ newOffset,
config,
- time = time,
- initFileSize = config.initFileSize,
- preallocate = config.preallocate)
+ time,
+ config.initFileSize,
+ config.preallocate)
segments.add(newSegment)
reason.logReason(List(segmentToDelete))
@@ -394,7 +396,7 @@ class LocalLog(@volatile private var _dir: File,
var segmentOpt = segments.floorSegment(startOffset)
// return error on attempt to read beyond the log end offset
- if (startOffset > endOffset || segmentOpt.isEmpty)
+ if (startOffset > endOffset || !segmentOpt.isPresent)
throw new OffsetOutOfRangeException(s"Received request for offset $startOffset for partition $topicPartition, " +
s"but we only have log segments upto $endOffset.")
@@ -407,7 +409,7 @@ class LocalLog(@volatile private var _dir: File,
// but if that segment doesn't contain any messages with an offset greater than that
// continue to read from successive segments until we get some messages or we reach the end of the log
var fetchDataInfo: FetchDataInfo = null
- while (fetchDataInfo == null && segmentOpt.isDefined) {
+ while (fetchDataInfo == null && segmentOpt.isPresent) {
val segment = segmentOpt.get
val baseOffset = segment.baseOffset
@@ -435,8 +437,7 @@ class LocalLog(@volatile private var _dir: File,
}
private[log] def append(lastOffset: Long, largestTimestamp: Long, shallowOffsetOfMaxTimestamp: Long, records: MemoryRecords): Unit = {
- segments.activeSegment.append(largestOffset = lastOffset, largestTimestamp = largestTimestamp,
- shallowOffsetOfMaxTimestamp = shallowOffsetOfMaxTimestamp, records = records)
+ segments.activeSegment.append(lastOffset, largestTimestamp, shallowOffsetOfMaxTimestamp, records)
updateLogEndOffset(lastOffset + 1)
}
@@ -445,9 +446,8 @@ class LocalLog(@volatile private var _dir: File,
val fetchSize = fetchInfo.records.sizeInBytes
val startOffsetPosition = new OffsetPosition(fetchInfo.fetchOffsetMetadata.messageOffset,
fetchInfo.fetchOffsetMetadata.relativePositionInSegment)
- val upperBoundOffset = segment.fetchUpperBoundOffset(startOffsetPosition, fetchSize).orElse {
- segments.higherSegment(segment.baseOffset).map(_.baseOffset).getOrElse(logEndOffset)
- }
+ val upperBoundOffset = segment.fetchUpperBoundOffset(startOffsetPosition, fetchSize).orElse(
+ segments.higherSegment(segment.baseOffset).asScala.map(s => s.baseOffset).getOrElse(logEndOffset))
val abortedTransactions = ListBuffer.empty[FetchResponseData.AbortedTransaction]
def accumulator(abortedTxns: Seq[AbortedTxn]): Unit = abortedTransactions ++= abortedTxns.map(_.asAbortedTransaction)
@@ -478,7 +478,7 @@ class LocalLog(@volatile private var _dir: File,
val segmentEntry = segments.floorSegment(baseOffset)
val allAbortedTxns = ListBuffer.empty[AbortedTxn]
def accumulator(abortedTxns: Seq[AbortedTxn]): Unit = allAbortedTxns ++= abortedTxns
- segmentEntry.foreach(segment => collectAbortedTransactions(logStartOffset, upperBoundOffset, segment, accumulator))
+ segmentEntry.ifPresent(segment => collectAbortedTransactions(logStartOffset, upperBoundOffset, segment, accumulator))
allAbortedTxns.toList
}
@@ -529,15 +529,15 @@ class LocalLog(@volatile private var _dir: File,
Files.delete(file.toPath)
}
- segments.lastSegment.foreach(_.onBecomeInactiveSegment())
+ segments.lastSegment.ifPresent(_.onBecomeInactiveSegment())
}
val newSegment = LogSegment.open(dir,
- baseOffset = newOffset,
+ newOffset,
config,
- time = time,
- initFileSize = config.initFileSize,
- preallocate = config.preallocate)
+ time,
+ config.initFileSize,
+ config.preallocate)
segments.add(newSegment)
// We need to update the segment base offset and append position data of the metadata when log rolls.
@@ -560,7 +560,7 @@ class LocalLog(@volatile private var _dir: File,
maybeHandleIOException(s"Error while truncating the entire log for $topicPartition in dir ${dir.getParent}") {
debug(s"Truncate and start at offset $newOffset")
checkIfMemoryMappedBufferClosed()
- val segmentsToDelete = List[LogSegment]() ++ segments.values
+ val segmentsToDelete = new util.ArrayList(segments.values).asScala
if (segmentsToDelete.nonEmpty) {
removeAndDeleteSegments(segmentsToDelete.dropRight(1), asyncDelete = true, LogTruncation(this))
@@ -582,7 +582,7 @@ class LocalLog(@volatile private var _dir: File,
* @return the list of segments that were scheduled for deletion
*/
private[log] def truncateTo(targetOffset: Long): Iterable[LogSegment] = {
- val deletableSegments = List[LogSegment]() ++ segments.filter(segment => segment.baseOffset > targetOffset)
+ val deletableSegments = segments.filter(segment => segment.baseOffset > targetOffset).asScala
removeAndDeleteSegments(deletableSegments, asyncDelete = true, LogTruncation(this))
segments.activeSegment.truncateTo(targetOffset)
updateLogEndOffset(targetOffset)
@@ -777,7 +777,7 @@ object LocalLog extends Logging {
newSegments.foreach { splitSegment =>
splitSegment.onBecomeInactiveSegment()
splitSegment.flush()
- splitSegment.lastModified = segment.lastModified
+ splitSegment.setLastModified(segment.lastModified)
totalSizeOfNewSegments += splitSegment.log.sizeInBytes
}
// size of all the new segments combined must equal size of the original segment
@@ -945,9 +945,8 @@ object LocalLog extends Logging {
}
private[log] def createNewCleanedSegment(dir: File, logConfig: LogConfig, baseOffset: Long): LogSegment = {
- LogSegment.deleteIfExists(dir, baseOffset, fileSuffix = CleanedFileSuffix)
- LogSegment.open(dir, baseOffset, logConfig, Time.SYSTEM,
- fileSuffix = CleanedFileSuffix, initFileSize = logConfig.initFileSize, preallocate = logConfig.preallocate)
+ LogSegment.deleteIfExists(dir, baseOffset, CleanedFileSuffix)
+ LogSegment.open(dir, baseOffset, logConfig, Time.SYSTEM, false, logConfig.initFileSize, logConfig.preallocate, CleanedFileSuffix)
}
/**
@@ -958,7 +957,7 @@ object LocalLog extends Logging {
* @tparam T the type of object held within the iterator
* @return Some(iterator.next) if a next element exists, None otherwise.
*/
- private def nextOption[T](iterator: Iterator[T]): Option[T] = {
+ private def nextOption[T](iterator: util.Iterator[T]): Option[T] = {
if (iterator.hasNext)
Some(iterator.next())
else
diff --git a/core/src/main/scala/kafka/log/LogCleaner.scala b/core/src/main/scala/kafka/log/LogCleaner.scala
index 454cbaa2b35..0fa9a3b0488 100644
--- a/core/src/main/scala/kafka/log/LogCleaner.scala
+++ b/core/src/main/scala/kafka/log/LogCleaner.scala
@@ -34,7 +34,7 @@ import org.apache.kafka.common.record._
import org.apache.kafka.common.utils.{BufferSupplier, Time}
import org.apache.kafka.server.metrics.KafkaMetricsGroup
import org.apache.kafka.server.util.ShutdownableThread
-import org.apache.kafka.storage.internals.log.{AbortedTxn, CleanerConfig, LastRecord, LogDirFailureChannel, OffsetMap, SkimpyOffsetMap, TransactionIndex}
+import org.apache.kafka.storage.internals.log.{AbortedTxn, CleanerConfig, LastRecord, LogDirFailureChannel, LogSegment, LogSegmentOffsetOverflowException, OffsetMap, SkimpyOffsetMap, TransactionIndex}
import scala.jdk.CollectionConverters._
import scala.collection.mutable.ListBuffer
@@ -696,7 +696,7 @@ private[log] class Cleaner(val id: Int,
// update the modification date to retain the last modified date of the original files
val modified = segments.last.lastModified
- cleaned.lastModified = modified
+ cleaned.setLastModified(modified)
// swap in new segment
info(s"Swapping in cleaned segment $cleaned for segment(s) $segments in log $log")
@@ -812,10 +812,7 @@ private[log] class Cleaner(val id: Int,
val retained = MemoryRecords.readableRecords(outputBuffer)
// it's OK not to hold the Log's lock in this case, because this segment is only accessed by other threads
// after `Log.replaceSegments` (which acquires the lock) is called
- dest.append(largestOffset = result.maxOffset,
- largestTimestamp = result.maxTimestamp,
- shallowOffsetOfMaxTimestamp = result.shallowOffsetOfMaxTimestamp,
- records = retained)
+ dest.append(result.maxOffset, result.maxTimestamp, result.shallowOffsetOfMaxTimestamp, retained)
throttler.maybeThrottle(outputBuffer.limit())
}
diff --git a/core/src/main/scala/kafka/log/LogCleanerManager.scala b/core/src/main/scala/kafka/log/LogCleanerManager.scala
index a687c0facff..0e147108efd 100755
--- a/core/src/main/scala/kafka/log/LogCleanerManager.scala
+++ b/core/src/main/scala/kafka/log/LogCleanerManager.scala
@@ -20,7 +20,6 @@ package kafka.log
import java.io.File
import java.util.concurrent.TimeUnit
import java.util.concurrent.locks.ReentrantLock
-
import kafka.common.LogCleaningAbortedException
import kafka.server.checkpoints.OffsetCheckpointFile
import kafka.utils.CoreUtils._
@@ -31,6 +30,7 @@ import org.apache.kafka.common.utils.Time
import org.apache.kafka.storage.internals.log.LogDirFailureChannel
import org.apache.kafka.server.metrics.KafkaMetricsGroup
+import java.util.Comparator
import scala.collection.{Iterable, Seq, mutable}
import scala.jdk.CollectionConverters._
@@ -595,13 +595,9 @@ private[log] object LogCleanerManager extends Logging {
*/
def maxCompactionDelay(log: UnifiedLog, firstDirtyOffset: Long, now: Long) : Long = {
val dirtyNonActiveSegments = log.nonActiveLogSegmentsFrom(firstDirtyOffset)
- val firstBatchTimestamps = log.getFirstBatchTimestampForSegments(dirtyNonActiveSegments).filter(_ > 0)
+ val firstBatchTimestamps = log.getFirstBatchTimestampForSegments(dirtyNonActiveSegments).stream.filter(_ > 0)
- val earliestDirtySegmentTimestamp = {
- if (firstBatchTimestamps.nonEmpty)
- firstBatchTimestamps.min
- else Long.MaxValue
- }
+ val earliestDirtySegmentTimestamp = firstBatchTimestamps.min(Comparator.naturalOrder()).orElse(Long.MaxValue);
val maxCompactionLagMs = math.max(log.config.maxCompactionLagMs, 0L)
val cleanUntilTime = now - maxCompactionLagMs
@@ -662,7 +658,7 @@ private[log] object LogCleanerManager extends Logging {
if (minCompactionLagMs > 0) {
// dirty log segments
val dirtyNonActiveSegments = log.nonActiveLogSegmentsFrom(firstDirtyOffset)
- dirtyNonActiveSegments.find { s =>
+ dirtyNonActiveSegments.asScala.find { s =>
val isUncleanable = s.largestTimestamp > now - minCompactionLagMs
debug(s"Checking if log segment may be cleaned: log='${log.name}' segment.baseOffset=${s.baseOffset} " +
s"segment.largestTimestamp=${s.largestTimestamp}; now - compactionLag=${now - minCompactionLagMs}; " +
@@ -684,7 +680,7 @@ private[log] object LogCleanerManager extends Logging {
* @return the biggest uncleanable offset and the total amount of cleanable bytes
*/
def calculateCleanableBytes(log: UnifiedLog, firstDirtyOffset: Long, uncleanableOffset: Long): (Long, Long) = {
- val firstUncleanableSegment = log.nonActiveLogSegmentsFrom(uncleanableOffset).headOption.getOrElse(log.activeSegment)
+ val firstUncleanableSegment = log.nonActiveLogSegmentsFrom(uncleanableOffset).asScala.headOption.getOrElse(log.activeSegment)
val firstUncleanableOffset = firstUncleanableSegment.baseOffset
val cleanableBytes = log.logSegments(math.min(firstDirtyOffset, firstUncleanableOffset), firstUncleanableOffset).map(_.size.toLong).sum
diff --git a/core/src/main/scala/kafka/log/LogLoader.scala b/core/src/main/scala/kafka/log/LogLoader.scala
index 37210bb226c..ebfb3e66949 100644
--- a/core/src/main/scala/kafka/log/LogLoader.scala
+++ b/core/src/main/scala/kafka/log/LogLoader.scala
@@ -19,7 +19,6 @@ package kafka.log
import java.io.{File, IOException}
import java.nio.file.{Files, NoSuchFileException}
-import kafka.common.LogSegmentOffsetOverflowException
import kafka.log.UnifiedLog.{CleanedFileSuffix, SwapFileSuffix, isIndexFile, isLogFile, offsetFromFile}
import kafka.utils.Logging
import org.apache.kafka.common.TopicPartition
@@ -28,9 +27,11 @@ import org.apache.kafka.common.utils.{Time, Utils}
import org.apache.kafka.snapshot.Snapshots
import org.apache.kafka.server.util.Scheduler
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
-import org.apache.kafka.storage.internals.log.{CorruptIndexException, LoadedLogOffsets, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetMetadata, ProducerStateManager}
+import org.apache.kafka.storage.internals.log.{CorruptIndexException, LoadedLogOffsets, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetMetadata, LogSegment, LogSegmentOffsetOverflowException, LogSegments, ProducerStateManager}
+import java.util.Optional
import java.util.concurrent.{ConcurrentHashMap, ConcurrentMap}
+import scala.collection.mutable.ArrayBuffer
import scala.collection.{Set, mutable}
import scala.jdk.CollectionConverters._
@@ -76,7 +77,7 @@ class LogLoader(
segments: LogSegments,
logStartOffsetCheckpoint: Long,
recoveryPointCheckpoint: Long,
- leaderEpochCache: Option[LeaderEpochFileCache],
+ leaderEpochCache: Optional[LeaderEpochFileCache],
producerStateManager: ProducerStateManager,
numRemainingSegments: ConcurrentMap[String, Int] = new ConcurrentHashMap[String, Int],
isRemoteLogEnabled: Boolean = false,
@@ -111,10 +112,13 @@ class LogLoader(
swapFiles.filter(f => UnifiedLog.isLogFile(new File(Utils.replaceSuffix(f.getPath, SwapFileSuffix, "")))).foreach { f =>
val baseOffset = offsetFromFile(f)
val segment = LogSegment.open(f.getParentFile,
- baseOffset = baseOffset,
+ baseOffset,
config,
- time = time,
- fileSuffix = UnifiedLog.SwapFileSuffix)
+ time,
+ false,
+ 0,
+ false,
+ UnifiedLog.SwapFileSuffix)
info(s"Found log file ${f.getPath} from interrupted swap operation, which is recoverable from ${UnifiedLog.SwapFileSuffix} files by renaming.")
minSwapFileOffset = Math.min(segment.baseOffset, minSwapFileOffset)
maxSwapFileOffset = Math.max(segment.readNextOffset, maxSwapFileOffset)
@@ -170,24 +174,25 @@ class LogLoader(
if (segments.isEmpty) {
segments.add(
LogSegment.open(
- dir = dir,
- baseOffset = 0,
+ dir,
+ 0,
config,
- time = time,
- initFileSize = config.initFileSize))
+ time,
+ config.initFileSize,
+ false))
}
(0L, 0L)
}
}
- leaderEpochCache.foreach(_.truncateFromEnd(nextOffset))
+ leaderEpochCache.ifPresent(_.truncateFromEnd(nextOffset))
val newLogStartOffset = if (isRemoteLogEnabled) {
logStartOffsetCheckpoint
} else {
math.max(logStartOffsetCheckpoint, segments.firstSegment.get.baseOffset)
}
// The earliest leader epoch may not be flushed during a hard failure. Recover it here.
- leaderEpochCache.foreach(_.truncateFromStart(logStartOffsetCheckpoint))
+ leaderEpochCache.ifPresent(_.truncateFromStart(logStartOffsetCheckpoint))
// Any segment loading or recovery code must not use producerStateManager, so that we can build the full state here
// from scratch.
@@ -197,7 +202,7 @@ class LogLoader(
// Reload all snapshots into the ProducerStateManager cache, the intermediate ProducerStateManager used
// during log recovery may have deleted some files without the LogLoader.producerStateManager instance witnessing the
// deletion.
- producerStateManager.removeStraySnapshots(segments.baseOffsets.map(x => Long.box(x)).asJavaCollection)
+ producerStateManager.removeStraySnapshots(segments.baseOffsets)
UnifiedLog.rebuildProducerState(
producerStateManager,
segments,
@@ -313,7 +318,7 @@ class LogLoader(
if (isIndexFile(file)) {
// if it is an index file, make sure it has a corresponding .log file
val offset = offsetFromFile(file)
- val logFile = UnifiedLog.logFile(dir, offset)
+ val logFile = LogFileUtils.logFile(dir, offset)
if (!logFile.exists) {
warn(s"Found an orphaned index file ${file.getAbsolutePath}, with no corresponding log file.")
Files.deleteIfExists(file.toPath)
@@ -321,13 +326,16 @@ class LogLoader(
} else if (isLogFile(file)) {
// if it's a log file, load the corresponding log segment
val baseOffset = offsetFromFile(file)
- val timeIndexFileNewlyCreated = !UnifiedLog.timeIndexFile(dir, baseOffset).exists()
+ val timeIndexFileNewlyCreated = !LogFileUtils.timeIndexFile(dir, baseOffset).exists()
val segment = LogSegment.open(
- dir = dir,
- baseOffset = baseOffset,
+ dir,
+ baseOffset,
config,
- time = time,
- fileAlreadyExists = true)
+ time,
+ true,
+ 0,
+ false,
+ "")
try segment.sanityCheck(timeIndexFileNewlyCreated)
catch {
@@ -403,8 +411,8 @@ class LogLoader(
warn(s"Deleting all segments because logEndOffset ($logEndOffset) " +
s"is smaller than logStartOffset $logStartOffsetCheckpoint. " +
"This could happen if segment files were deleted from the file system.")
- removeAndDeleteSegmentsAsync(segments.values)
- leaderEpochCache.foreach(_.clearAndFlush())
+ removeAndDeleteSegmentsAsync(segments.values.asScala)
+ leaderEpochCache.ifPresent(_.clearAndFlush())
producerStateManager.truncateFullyAndStartAt(logStartOffsetCheckpoint)
None
}
@@ -439,7 +447,9 @@ class LogLoader(
// we had an invalid message, delete all remaining log
warn(s"Corruption found in segment ${segment.baseOffset}," +
s" truncating to offset ${segment.readNextOffset}")
- removeAndDeleteSegmentsAsync(unflushedIter.toList)
+ val unflushedRemaining = new ArrayBuffer[LogSegment]
+ unflushedIter.forEachRemaining(s => unflushedRemaining += s)
+ removeAndDeleteSegmentsAsync(unflushedRemaining)
truncated = true
// segment is truncated, so set remaining segments to 0
numRemainingSegments.put(threadName, 0)
@@ -456,12 +466,12 @@ class LogLoader(
// no existing segments, create a new mutable segment beginning at logStartOffset
segments.add(
LogSegment.open(
- dir = dir,
- baseOffset = logStartOffsetCheckpoint,
+ dir,
+ logStartOffsetCheckpoint,
config,
- time = time,
- initFileSize = config.initFileSize,
- preallocate = config.preallocate))
+ time,
+ config.initFileSize,
+ config.preallocate))
}
// Update the recovery point if there was a clean shutdown and did not perform any changes to
diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala
index 89e1be0b60b..e420b0df435 100755
--- a/core/src/main/scala/kafka/log/LogManager.scala
+++ b/core/src/main/scala/kafka/log/LogManager.scala
@@ -837,7 +837,7 @@ class LogManager(logDirs: Seq[File],
try {
logStartOffsetCheckpoints.get(logDir).foreach { checkpoint =>
val logStartOffsets = logsToCheckpoint.collect {
- case (tp, log) if log.logStartOffset > log.logSegments.head.baseOffset => tp -> log.logStartOffset
+ case (tp, log) if log.logStartOffset > log.logSegments.asScala.head.baseOffset => tp -> log.logStartOffset
}
checkpoint.write(logStartOffsets)
}
diff --git a/core/src/main/scala/kafka/log/LogSegment.scala b/core/src/main/scala/kafka/log/LogSegment.scala
deleted file mode 100644
index bc07df1ddc1..00000000000
--- a/core/src/main/scala/kafka/log/LogSegment.scala
+++ /dev/null
@@ -1,693 +0,0 @@
-/**
- * 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 kafka.log
-
-import com.yammer.metrics.core.Timer
-import kafka.common.LogSegmentOffsetOverflowException
-import kafka.utils._
-import org.apache.kafka.common.InvalidRecordException
-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, Utils}
-import org.apache.kafka.server.metrics.KafkaMetricsGroup
-import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
-import org.apache.kafka.storage.internals.log.{AbortedTxn, AppendOrigin, CompletedTxn, FetchDataInfo, LazyIndex, LogConfig, LogOffsetMetadata, OffsetIndex, OffsetPosition, ProducerStateManager, RollParams, TimeIndex, TimestampOffset, TransactionIndex, TxnIndexSearchResult}
-
-import java.io.{File, IOException}
-import java.nio.file.attribute.FileTime
-import java.nio.file.{Files, NoSuchFileException}
-import java.util.Optional
-import java.util.concurrent.TimeUnit
-import scala.compat.java8.OptionConverters._
-import scala.jdk.CollectionConverters._
-import scala.math._
-
-/**
- * A segment of the log. Each segment has two components: a log and an index. The log is a FileRecords containing
- * the actual messages. The index is an OffsetIndex that maps from logical offsets to physical file positions. Each
- * segment has a base offset which is an offset <= the least offset of any message in this segment and > any offset in
- * any previous segment.
- *
- * A segment with a base offset of [base_offset] would be stored in two files, a [base_offset].index and a [base_offset].log file.
- *
- * @param log The file records containing log entries
- * @param lazyOffsetIndex The offset index
- * @param lazyTimeIndex The timestamp index
- * @param txnIndex The transaction index
- * @param baseOffset A lower bound on the offsets in this segment
- * @param indexIntervalBytes The approximate number of bytes between entries in the index
- * @param rollJitterMs The maximum random jitter subtracted from the scheduled segment roll time
- * @param time The time instance
- */
-@nonthreadsafe
-class LogSegment private[log] (val log: FileRecords,
- val lazyOffsetIndex: LazyIndex[OffsetIndex],
- val lazyTimeIndex: LazyIndex[TimeIndex],
- val txnIndex: TransactionIndex,
- val baseOffset: Long,
- val indexIntervalBytes: Int,
- val rollJitterMs: Long,
- val time: Time) extends Logging {
-
- def offsetIndex: OffsetIndex = lazyOffsetIndex.get
-
- def timeIndex: TimeIndex = lazyTimeIndex.get
-
- def shouldRoll(rollParams: RollParams): Boolean = {
- val reachedRollMs = timeWaitedForRoll(rollParams.now, rollParams.maxTimestampInMessages) > rollParams.maxSegmentMs - rollJitterMs
- size > rollParams.maxSegmentBytes - rollParams.messagesSize ||
- (size > 0 && reachedRollMs) ||
- offsetIndex.isFull || timeIndex.isFull || !canConvertToRelativeOffset(rollParams.maxOffsetInMessages)
- }
-
- def resizeIndexes(size: Int): Unit = {
- offsetIndex.resize(size)
- timeIndex.resize(size)
- }
-
- def sanityCheck(timeIndexFileNewlyCreated: Boolean): Unit = {
- if (lazyOffsetIndex.file.exists) {
- // Resize the time index file to 0 if it is newly created.
- if (timeIndexFileNewlyCreated)
- timeIndex.resize(0)
- // Sanity checks for time index and offset index are skipped because
- // we will recover the segments above the recovery point in recoverLog()
- // in any case so sanity checking them here is redundant.
- txnIndex.sanityCheck()
- }
- else throw new NoSuchFileException(s"Offset index file ${lazyOffsetIndex.file.getAbsolutePath} does not exist")
- }
-
- private var created = time.milliseconds
-
- /* the number of bytes since we last added an entry in the offset index */
- private var bytesSinceLastIndexEntry = 0
-
- // The timestamp we used for time based log rolling and for ensuring max compaction delay
- // volatile for LogCleaner to see the update
- @volatile private var rollingBasedTimestamp: Option[Long] = None
-
- /* The maximum timestamp and offset we see so far */
- @volatile private var _maxTimestampAndOffsetSoFar: TimestampOffset = TimestampOffset.UNKNOWN
- def maxTimestampAndOffsetSoFar_= (timestampOffset: TimestampOffset): Unit = _maxTimestampAndOffsetSoFar = timestampOffset
- def maxTimestampAndOffsetSoFar: TimestampOffset = {
- if (_maxTimestampAndOffsetSoFar == TimestampOffset.UNKNOWN)
- _maxTimestampAndOffsetSoFar = timeIndex.lastEntry
- _maxTimestampAndOffsetSoFar
- }
-
- /* The maximum timestamp we see so far */
- def maxTimestampSoFar: Long = {
- maxTimestampAndOffsetSoFar.timestamp
- }
-
- def offsetOfMaxTimestampSoFar: Long = {
- maxTimestampAndOffsetSoFar.offset
- }
-
- /* Return the size in bytes of this log segment */
- def size: Int = log.sizeInBytes()
-
- /**
- * checks that the argument offset can be represented as an integer offset relative to the baseOffset.
- */
- def canConvertToRelativeOffset(offset: Long): Boolean = {
- offsetIndex.canAppendOffset(offset)
- }
-
- /**
- * Append the given messages starting with the given offset. Add
- * an entry to the index if needed.
- *
- * It is assumed this method is being called from within a lock.
- *
- * @param largestOffset The last offset in the message set
- * @param largestTimestamp The largest timestamp in the message set.
- * @param shallowOffsetOfMaxTimestamp The offset of the message that has the largest timestamp in the messages to append.
- * @param records The log entries to append.
- * @throws LogSegmentOffsetOverflowException if the largest offset causes index offset overflow
- */
- @nonthreadsafe
- def append(largestOffset: Long,
- largestTimestamp: Long,
- shallowOffsetOfMaxTimestamp: Long,
- records: MemoryRecords): Unit = {
- if (records.sizeInBytes > 0) {
- trace(s"Inserting ${records.sizeInBytes} bytes at end offset $largestOffset at position ${log.sizeInBytes} " +
- s"with largest timestamp $largestTimestamp at shallow offset $shallowOffsetOfMaxTimestamp")
- val physicalPosition = log.sizeInBytes()
- if (physicalPosition == 0)
- rollingBasedTimestamp = Some(largestTimestamp)
-
- ensureOffsetInRange(largestOffset)
-
- // append the messages
- val appendedBytes = log.append(records)
- trace(s"Appended $appendedBytes to ${log.file} at end offset $largestOffset")
- // Update the in memory max timestamp and corresponding offset.
- if (largestTimestamp > maxTimestampSoFar) {
- maxTimestampAndOffsetSoFar = new TimestampOffset(largestTimestamp, shallowOffsetOfMaxTimestamp)
- }
- // append an entry to the index (if needed)
- if (bytesSinceLastIndexEntry > indexIntervalBytes) {
- offsetIndex.append(largestOffset, physicalPosition)
- timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar)
- bytesSinceLastIndexEntry = 0
- }
- bytesSinceLastIndexEntry += records.sizeInBytes
- }
- }
-
- private def ensureOffsetInRange(offset: Long): Unit = {
- if (!canConvertToRelativeOffset(offset))
- throw new LogSegmentOffsetOverflowException(this, offset)
- }
-
- private def appendChunkFromFile(records: FileRecords, position: Int, bufferSupplier: BufferSupplier): Int = {
- var bytesToAppend = 0
- var maxTimestamp = Long.MinValue
- var offsetOfMaxTimestamp = Long.MinValue
- var maxOffset = Long.MinValue
- var readBuffer = bufferSupplier.get(1024 * 1024)
-
- def canAppend(batch: RecordBatch) =
- canConvertToRelativeOffset(batch.lastOffset) &&
- (bytesToAppend == 0 || bytesToAppend + batch.sizeInBytes < readBuffer.capacity)
-
- // find all batches that are valid to be appended to the current log segment and
- // determine the maximum offset and timestamp
- val nextBatches = records.batchesFrom(position).asScala.iterator
- for (batch <- nextBatches.takeWhile(canAppend)) {
- if (batch.maxTimestamp > maxTimestamp) {
- maxTimestamp = batch.maxTimestamp
- offsetOfMaxTimestamp = batch.lastOffset
- }
- maxOffset = batch.lastOffset
- bytesToAppend += batch.sizeInBytes
- }
-
- if (bytesToAppend > 0) {
- // Grow buffer if needed to ensure we copy at least one batch
- if (readBuffer.capacity < bytesToAppend)
- readBuffer = bufferSupplier.get(bytesToAppend)
-
- readBuffer.limit(bytesToAppend)
- records.readInto(readBuffer, position)
-
- append(maxOffset, maxTimestamp, offsetOfMaxTimestamp, MemoryRecords.readableRecords(readBuffer))
- }
-
- bufferSupplier.release(readBuffer)
- bytesToAppend
- }
-
- /**
- * Append records from a file beginning at the given position until either the end of the file
- * is reached or an offset is found which is too large to convert to a relative offset for the indexes.
- *
- * @return the number of bytes appended to the log (may be less than the size of the input if an
- * offset is encountered which would overflow this segment)
- */
- def appendFromFile(records: FileRecords, start: Int): Int = {
- var position = start
- val bufferSupplier: BufferSupplier = new BufferSupplier.GrowableBufferSupplier
- while (position < start + records.sizeInBytes) {
- val bytesAppended = appendChunkFromFile(records, position, bufferSupplier)
- if (bytesAppended == 0)
- return position - start
- position += bytesAppended
- }
- position - start
- }
-
- @nonthreadsafe
- def updateTxnIndex(completedTxn: CompletedTxn, lastStableOffset: Long): Unit = {
- if (completedTxn.isAborted) {
- trace(s"Writing aborted transaction $completedTxn to transaction index, last stable offset is $lastStableOffset")
- txnIndex.append(new AbortedTxn(completedTxn, lastStableOffset))
- }
- }
-
- private def updateProducerState(producerStateManager: ProducerStateManager, batch: RecordBatch): Unit = {
- if (batch.hasProducerId) {
- val producerId = batch.producerId
- val appendInfo = producerStateManager.prepareUpdate(producerId, AppendOrigin.REPLICATION)
- val maybeCompletedTxn = appendInfo.append(batch, Optional.empty())
- producerStateManager.update(appendInfo)
- maybeCompletedTxn.ifPresent(completedTxn => {
- val lastStableOffset = producerStateManager.lastStableOffset(completedTxn)
- updateTxnIndex(completedTxn, lastStableOffset)
- producerStateManager.completeTxn(completedTxn)
- })
- }
- producerStateManager.updateMapEndOffset(batch.lastOffset + 1)
- }
-
- /**
- * Find the physical file position for the first message with offset >= the requested offset.
- *
- * The startingFilePosition argument is an optimization that can be used if we already know a valid starting position
- * in the file higher than the greatest-lower-bound from the index.
- *
- * @param offset The offset we want to translate
- * @param startingFilePosition A lower bound on the file position from which to begin the search. This is purely an optimization and
- * when omitted, the search will begin at the position in the offset index.
- * @return The position in the log storing the message with the least offset >= the requested offset and the size of the
- * message or null if no message meets this criteria.
- */
- @threadsafe
- private[log] def translateOffset(offset: Long, startingFilePosition: Int = 0): LogOffsetPosition = {
- val mapping = offsetIndex.lookup(offset)
- log.searchForOffsetWithSize(offset, max(mapping.position, startingFilePosition))
- }
-
- /**
- * Read a message set from this segment beginning with the first offset >= startOffset. The message set will include
- * no more than maxSize bytes and will end before maxOffset if a maxOffset is specified.
- *
- * @param startOffset A lower bound on the first offset to include in the message set we read
- * @param maxSize The maximum number of bytes to include in the message set we read
- * @param maxPosition The maximum position in the log segment that should be exposed for read
- * @param minOneMessage If this is true, the first message will be returned even if it exceeds `maxSize` (if one exists)
- *
- * @return The fetched data and the offset metadata of the first message whose offset is >= startOffset,
- * or null if the startOffset is larger than the largest offset in this log
- */
- @threadsafe
- def read(startOffset: Long,
- maxSize: Int,
- maxPosition: Long = size,
- minOneMessage: Boolean = false): FetchDataInfo = {
- if (maxSize < 0)
- throw new IllegalArgumentException(s"Invalid max size $maxSize for log read from segment $log")
-
- val startOffsetAndSize = translateOffset(startOffset)
-
- // if the start position is already off the end of the log, return null
- if (startOffsetAndSize == null)
- return null
-
- val startPosition = startOffsetAndSize.position
- val offsetMetadata = new LogOffsetMetadata(startOffset, this.baseOffset, startPosition)
-
- val adjustedMaxSize =
- if (minOneMessage) math.max(maxSize, startOffsetAndSize.size)
- else maxSize
-
- // return a log segment but with zero size in the case below
- if (adjustedMaxSize == 0)
- return new FetchDataInfo(offsetMetadata, MemoryRecords.EMPTY)
-
- // calculate the length of the message set to read based on whether or not they gave us a maxOffset
- val fetchSize: Int = min((maxPosition - startPosition).toInt, adjustedMaxSize)
-
- new FetchDataInfo(offsetMetadata, log.slice(startPosition, fetchSize),
- adjustedMaxSize < startOffsetAndSize.size, Optional.empty())
- }
-
- def fetchUpperBoundOffset(startOffsetPosition: OffsetPosition, fetchSize: Int): Optional[Long] =
- offsetIndex.fetchUpperBoundOffset(startOffsetPosition, fetchSize).map(_.offset)
-
- /**
- * Run recovery on the given segment. This will rebuild the index from the log file and lop off any invalid bytes
- * from the end of the log and index.
- *
- * @param producerStateManager Producer state corresponding to the segment's base offset. This is needed to recover
- * the transaction index.
- * @param leaderEpochCache Optionally a cache for updating the leader epoch during recovery.
- * @return The number of bytes truncated from the log
- * @throws LogSegmentOffsetOverflowException if the log segment contains an offset that causes the index offset to overflow
- */
- @nonthreadsafe
- def recover(producerStateManager: ProducerStateManager, leaderEpochCache: Option[LeaderEpochFileCache] = None): Int = {
- offsetIndex.reset()
- timeIndex.reset()
- txnIndex.reset()
- var validBytes = 0
- var lastIndexEntry = 0
- maxTimestampAndOffsetSoFar = TimestampOffset.UNKNOWN
- try {
- for (batch <- log.batches.asScala) {
- batch.ensureValid()
- ensureOffsetInRange(batch.lastOffset)
-
- // The max timestamp is exposed at the batch level, so no need to iterate the records
- if (batch.maxTimestamp > maxTimestampSoFar) {
- maxTimestampAndOffsetSoFar = new TimestampOffset(batch.maxTimestamp, batch.lastOffset)
- }
-
- // Build offset index
- if (validBytes - lastIndexEntry > indexIntervalBytes) {
- offsetIndex.append(batch.lastOffset, validBytes)
- timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar)
- lastIndexEntry = validBytes
- }
- validBytes += batch.sizeInBytes()
-
- if (batch.magic >= RecordBatch.MAGIC_VALUE_V2) {
- leaderEpochCache.foreach { cache =>
- if (batch.partitionLeaderEpoch >= 0 && cache.latestEpoch.asScala.forall(batch.partitionLeaderEpoch > _))
- cache.assign(batch.partitionLeaderEpoch, batch.baseOffset)
- }
- updateProducerState(producerStateManager, batch)
- }
- }
- } catch {
- case e@ (_: CorruptRecordException | _: InvalidRecordException) =>
- warn("Found invalid messages in log segment %s at byte offset %d: %s. %s"
- .format(log.file.getAbsolutePath, validBytes, e.getMessage, e.getCause))
- }
- val truncated = log.sizeInBytes - validBytes
- if (truncated > 0)
- debug(s"Truncated $truncated invalid bytes at the end of segment ${log.file.getAbsoluteFile} during recovery")
-
- log.truncateTo(validBytes)
- offsetIndex.trimToValidSize()
- // A normally closed segment always appends the biggest timestamp ever seen into log segment, we do this as well.
- timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar, true)
- timeIndex.trimToValidSize()
- truncated
- }
-
- private def loadLargestTimestamp(): Unit = {
- // Get the last time index entry. If the time index is empty, it will return (-1, baseOffset)
- val lastTimeIndexEntry = timeIndex.lastEntry
- maxTimestampAndOffsetSoFar = lastTimeIndexEntry
-
- val offsetPosition = offsetIndex.lookup(lastTimeIndexEntry.offset)
- // Scan the rest of the messages to see if there is a larger timestamp after the last time index entry.
- val maxTimestampOffsetAfterLastEntry = log.largestTimestampAfter(offsetPosition.position)
- if (maxTimestampOffsetAfterLastEntry.timestamp > lastTimeIndexEntry.timestamp) {
- maxTimestampAndOffsetSoFar = new TimestampOffset(maxTimestampOffsetAfterLastEntry.timestamp, maxTimestampOffsetAfterLastEntry.offset)
- }
- }
-
- /**
- * Check whether the last offset of the last batch in this segment overflows the indexes.
- */
- def hasOverflow: Boolean = {
- val nextOffset = readNextOffset
- nextOffset > baseOffset && !canConvertToRelativeOffset(nextOffset - 1)
- }
-
- def collectAbortedTxns(fetchOffset: Long, upperBoundOffset: Long): TxnIndexSearchResult =
- txnIndex.collectAbortedTxns(fetchOffset, upperBoundOffset)
-
- override def toString: String = "LogSegment(baseOffset=" + baseOffset +
- ", size=" + size +
- ", lastModifiedTime=" + lastModified +
- ", largestRecordTimestamp=" + largestRecordTimestamp +
- ")"
-
- /**
- * Truncate off all index and log entries with offsets >= the given offset.
- * If the given offset is larger than the largest message in this segment, do nothing.
- *
- * @param offset The offset to truncate to
- * @return The number of log bytes truncated
- */
- @nonthreadsafe
- def truncateTo(offset: Long): Int = {
- // Do offset translation before truncating the index to avoid needless scanning
- // in case we truncate the full index
- val mapping = translateOffset(offset)
- offsetIndex.truncateTo(offset)
- timeIndex.truncateTo(offset)
- txnIndex.truncateTo(offset)
-
- // After truncation, reset and allocate more space for the (new currently active) index
- offsetIndex.resize(offsetIndex.maxIndexSize)
- timeIndex.resize(timeIndex.maxIndexSize)
-
- val bytesTruncated = if (mapping == null) 0 else log.truncateTo(mapping.position)
- if (log.sizeInBytes == 0) {
- created = time.milliseconds
- rollingBasedTimestamp = None
- }
-
- bytesSinceLastIndexEntry = 0
- if (maxTimestampSoFar >= 0)
- loadLargestTimestamp()
- bytesTruncated
- }
-
- /**
- * Calculate the offset that would be used for the next message to be append to this segment.
- * Note that this is expensive.
- */
- @threadsafe
- def readNextOffset: Long = {
- val fetchData = read(offsetIndex.lastOffset, log.sizeInBytes)
- if (fetchData == null)
- baseOffset
- else
- fetchData.records.batches.asScala.lastOption
- .map(_.nextOffset)
- .getOrElse(baseOffset)
- }
-
- /**
- * Flush this log segment to disk
- */
- @threadsafe
- def flush(): Unit = {
- LogFlushStats.logFlushTimer.time { () =>
- log.flush()
- offsetIndex.flush()
- timeIndex.flush()
- txnIndex.flush()
- }
- }
-
- /**
- * Update the directory reference for the log and indices in this segment. This would typically be called after a
- * directory is renamed.
- */
- def updateParentDir(dir: File): Unit = {
- log.updateParentDir(dir)
- lazyOffsetIndex.updateParentDir(dir)
- lazyTimeIndex.updateParentDir(dir)
- txnIndex.updateParentDir(dir)
- }
-
- /**
- * Change the suffix for the index and log files for this log segment
- * IOException from this method should be handled by the caller
- */
- def changeFileSuffixes(oldSuffix: String, newSuffix: String): Unit = {
- log.renameTo(new File(Utils.replaceSuffix(log.file.getPath, oldSuffix, newSuffix)))
- lazyOffsetIndex.renameTo(new File(Utils.replaceSuffix(lazyOffsetIndex.file.getPath, oldSuffix, newSuffix)))
- lazyTimeIndex.renameTo(new File(Utils.replaceSuffix(lazyTimeIndex.file.getPath, oldSuffix, newSuffix)))
- txnIndex.renameTo(new File(Utils.replaceSuffix(txnIndex.file.getPath, oldSuffix, newSuffix)))
- }
-
- def hasSuffix(suffix: String): Boolean = {
- log.file.getName.endsWith(suffix) &&
- lazyOffsetIndex.file.getName.endsWith(suffix) &&
- lazyTimeIndex.file.getName.endsWith(suffix) &&
- txnIndex.file.getName.endsWith(suffix)
- }
-
- /**
- * Append the largest time index entry to the time index and trim the log and indexes.
- *
- * The time index entry appended will be used to decide when to delete the segment.
- */
- def onBecomeInactiveSegment(): Unit = {
- timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar, true)
- offsetIndex.trimToValidSize()
- timeIndex.trimToValidSize()
- log.trim()
- }
-
- /**
- * If not previously loaded,
- * load the timestamp of the first message into memory.
- */
- private def loadFirstBatchTimestamp(): Unit = {
- if (rollingBasedTimestamp.isEmpty) {
- val iter = log.batches.iterator()
- if (iter.hasNext)
- rollingBasedTimestamp = Some(iter.next().maxTimestamp)
- }
- }
-
- /**
- * The time this segment has waited to be rolled.
- * If the first message batch has a timestamp we use its timestamp to determine when to roll a segment. A segment
- * is rolled if the difference between the new batch's timestamp and the first batch's timestamp exceeds the
- * segment rolling time.
- * If the first batch does not have a timestamp, we use the wall clock time to determine when to roll a segment. A
- * segment is rolled if the difference between the current wall clock time and the segment create time exceeds the
- * segment rolling time.
- */
- def timeWaitedForRoll(now: Long, messageTimestamp: Long): Long = {
- // Load the timestamp of the first message into memory
- loadFirstBatchTimestamp()
- rollingBasedTimestamp match {
- case Some(t) if t >= 0 => messageTimestamp - t
- case _ => now - created
- }
- }
-
- /**
- * @return the first batch timestamp if the timestamp is available. Otherwise return Long.MaxValue
- */
- def getFirstBatchTimestamp(): Long = {
- loadFirstBatchTimestamp()
- rollingBasedTimestamp match {
- case Some(t) if t >= 0 => t
- case _ => Long.MaxValue
- }
- }
-
- /**
- * Search the message offset based on timestamp and offset.
- *
- * This method returns an option of TimestampOffset. The returned value is determined using the following ordered list of rules:
- *
- * - If all the messages in the segment have smaller offsets, return None
- * - If all the messages in the segment have smaller timestamps, return None
- * - If all the messages in the segment have larger timestamps, or no message in the segment has a timestamp
- * the returned the offset will be max(the base offset of the segment, startingOffset) and the timestamp will be Message.NoTimestamp.
- * - Otherwise, return an option of TimestampOffset. The offset is the offset of the first message whose timestamp
- * is greater than or equals to the target timestamp and whose offset is greater than or equals to the startingOffset.
- *
- * This methods only returns None when 1) all messages' offset < startOffing or 2) the log is not empty but we did not
- * see any message when scanning the log from the indexed position. The latter could happen if the log is truncated
- * after we get the indexed position but before we scan the log from there. In this case we simply return None and the
- * caller will need to check on the truncated log and maybe retry or even do the search on another log segment.
- *
- * @param timestamp The timestamp to search for.
- * @param startingOffset The starting offset to search.
- * @return the timestamp and offset of the first message that meets the requirements. None will be returned if there is no such message.
- */
- def findOffsetByTimestamp(timestamp: Long, startingOffset: Long = baseOffset): Option[TimestampAndOffset] = {
- // Get the index entry with a timestamp less than or equal to the target timestamp
- val timestampOffset = timeIndex.lookup(timestamp)
- val position = offsetIndex.lookup(math.max(timestampOffset.offset, startingOffset)).position
-
- // Search the timestamp
- Option(log.searchForTimestamp(timestamp, position, startingOffset))
- }
-
- /**
- * Close this log segment
- */
- def close(): Unit = {
- if (_maxTimestampAndOffsetSoFar != TimestampOffset.UNKNOWN)
- CoreUtils.swallow(timeIndex.maybeAppend(maxTimestampSoFar, offsetOfMaxTimestampSoFar, true), this)
- CoreUtils.swallow(lazyOffsetIndex.close(), this)
- CoreUtils.swallow(lazyTimeIndex.close(), this)
- CoreUtils.swallow(log.close(), this)
- CoreUtils.swallow(txnIndex.close(), this)
- }
-
- /**
- * Close file handlers used by the log segment but don't write to disk. This is used when the disk may have failed
- */
- def closeHandlers(): Unit = {
- CoreUtils.swallow(lazyOffsetIndex.closeHandler(), this)
- CoreUtils.swallow(lazyTimeIndex.closeHandler(), this)
- CoreUtils.swallow(log.closeHandlers(), this)
- CoreUtils.swallow(txnIndex.close(), this)
- }
-
- /**
- * Delete this log segment from the filesystem.
- */
- def deleteIfExists(): Unit = {
- def delete(delete: () => Boolean, fileType: String, file: File, logIfMissing: Boolean): Unit = {
- try {
- if (delete())
- info(s"Deleted $fileType ${file.getAbsolutePath}.")
- else if (logIfMissing)
- info(s"Failed to delete $fileType ${file.getAbsolutePath} because it does not exist.")
- }
- catch {
- case e: IOException => throw new IOException(s"Delete of $fileType ${file.getAbsolutePath} failed.", e)
- }
- }
-
- CoreUtils.tryAll(Seq(
- () => delete(log.deleteIfExists _, "log", log.file, logIfMissing = true),
- () => delete(lazyOffsetIndex.deleteIfExists _, "offset index", lazyOffsetIndex.file, logIfMissing = true),
- () => delete(lazyTimeIndex.deleteIfExists _, "time index", lazyTimeIndex.file, logIfMissing = true),
- () => delete(txnIndex.deleteIfExists _, "transaction index", txnIndex.file, logIfMissing = false)
- ))
- }
-
- def deleted(): Boolean = {
- !log.file.exists() && !lazyOffsetIndex.file.exists() && !lazyTimeIndex.file.exists() && !txnIndex.file.exists()
- }
-
- /**
- * The last modified time of this log segment as a unix time stamp
- */
- def lastModified = log.file.lastModified
-
- /**
- * The largest timestamp this segment contains, if maxTimestampSoFar >= 0, otherwise None.
- */
- def largestRecordTimestamp: Option[Long] = if (maxTimestampSoFar >= 0) Some(maxTimestampSoFar) else None
-
- /**
- * The largest timestamp this segment contains.
- */
- def largestTimestamp = if (maxTimestampSoFar >= 0) maxTimestampSoFar else lastModified
-
- /**
- * Change the last modified time for this log segment
- */
- def lastModified_=(ms: Long) = {
- val fileTime = FileTime.fromMillis(ms)
- Files.setLastModifiedTime(log.file.toPath, fileTime)
- Files.setLastModifiedTime(lazyOffsetIndex.file.toPath, fileTime)
- Files.setLastModifiedTime(lazyTimeIndex.file.toPath, fileTime)
- }
-
-}
-
-object LogSegment {
-
- def open(dir: File, baseOffset: Long, config: LogConfig, time: Time, fileAlreadyExists: Boolean = false,
- initFileSize: Int = 0, preallocate: Boolean = false, fileSuffix: String = ""): LogSegment = {
- val maxIndexSize = config.maxIndexSize
- new LogSegment(
- FileRecords.open(UnifiedLog.logFile(dir, baseOffset, fileSuffix), fileAlreadyExists, initFileSize, preallocate),
- LazyIndex.forOffset(UnifiedLog.offsetIndexFile(dir, baseOffset, fileSuffix), baseOffset, maxIndexSize),
- LazyIndex.forTime(UnifiedLog.timeIndexFile(dir, baseOffset, fileSuffix), baseOffset, maxIndexSize),
- new TransactionIndex(baseOffset, UnifiedLog.transactionIndexFile(dir, baseOffset, fileSuffix)),
- baseOffset,
- indexIntervalBytes = config.indexInterval,
- rollJitterMs = config.randomSegmentJitter,
- time)
- }
-
- def deleteIfExists(dir: File, baseOffset: Long, fileSuffix: String = ""): Unit = {
- UnifiedLog.deleteFileIfExists(UnifiedLog.offsetIndexFile(dir, baseOffset, fileSuffix))
- UnifiedLog.deleteFileIfExists(UnifiedLog.timeIndexFile(dir, baseOffset, fileSuffix))
- UnifiedLog.deleteFileIfExists(UnifiedLog.transactionIndexFile(dir, baseOffset, fileSuffix))
- UnifiedLog.deleteFileIfExists(UnifiedLog.logFile(dir, baseOffset, fileSuffix))
- }
-}
-
-object LogFlushStats {
- private val metricsGroup = new KafkaMetricsGroup(LogFlushStats.getClass)
- val logFlushTimer: Timer = metricsGroup.newTimer("LogFlushRateAndTimeMs", TimeUnit.MILLISECONDS, TimeUnit.SECONDS)
-}
diff --git a/core/src/main/scala/kafka/log/LogSegments.scala b/core/src/main/scala/kafka/log/LogSegments.scala
deleted file mode 100644
index 564586dc9b4..00000000000
--- a/core/src/main/scala/kafka/log/LogSegments.scala
+++ /dev/null
@@ -1,268 +0,0 @@
-/**
- * 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 kafka.log
-
-import java.io.File
-import java.util.Map
-import java.util.concurrent.{ConcurrentNavigableMap, ConcurrentSkipListMap}
-
-import kafka.utils.threadsafe
-import org.apache.kafka.common.TopicPartition
-
-import scala.jdk.CollectionConverters._
-
-/**
- * This class encapsulates a thread-safe navigable map of LogSegment instances and provides the
- * required read and write behavior on the map.
- *
- * @param topicPartition the TopicPartition associated with the segments
- * (useful for logging purposes)
- */
-class LogSegments(topicPartition: TopicPartition) {
-
- /* the segments of the log with key being LogSegment base offset and value being a LogSegment */
- private val segments: ConcurrentNavigableMap[Long, LogSegment] = new ConcurrentSkipListMap[Long, LogSegment]
-
- /**
- * @return true if the segments are empty, false otherwise.
- */
- @threadsafe
- def isEmpty: Boolean = segments.isEmpty
-
- /**
- * @return true if the segments are non-empty, false otherwise.
- */
- @threadsafe
- def nonEmpty: Boolean = !isEmpty
-
- /**
- * Add the given segment, or replace an existing entry.
- *
- * @param segment the segment to add
- */
- @threadsafe
- def add(segment: LogSegment): LogSegment = this.segments.put(segment.baseOffset, segment)
-
- /**
- * Remove the segment at the provided offset.
- *
- * @param offset the offset to be removed
- */
- @threadsafe
- def remove(offset: Long): Unit = segments.remove(offset)
-
- /**
- * Clears all entries.
- */
- @threadsafe
- def clear(): Unit = segments.clear()
-
- /**
- * Close all segments.
- */
- def close(): Unit = values.foreach(_.close())
-
- /**
- * Close the handlers for all segments.
- */
- def closeHandlers(): Unit = values.foreach(_.closeHandlers())
-
- /**
- * Update the directory reference for the log and indices of all segments.
- *
- * @param dir the renamed directory
- */
- def updateParentDir(dir: File): Unit = values.foreach(_.updateParentDir(dir))
-
- /**
- * Take care! this is an O(n) operation, where n is the number of segments.
- *
- * @return The number of segments.
- *
- */
- @threadsafe
- def numberOfSegments: Int = segments.size
-
- /**
- * @return the base offsets of all segments
- */
- def baseOffsets: Iterable[Long] = segments.values().asScala.map(_.baseOffset)
-
- /**
- * @param offset the segment to be checked
- * @return true if a segment exists at the provided offset, false otherwise.
- */
- @threadsafe
- def contains(offset: Long): Boolean = segments.containsKey(offset)
-
- /**
- * Retrieves a segment at the specified offset.
- *
- * @param offset the segment to be retrieved
- *
- * @return the segment if it exists, otherwise None.
- */
- @threadsafe
- def get(offset: Long): Option[LogSegment] = Option(segments.get(offset))
-
- /**
- * @return an iterator to the log segments ordered from oldest to newest.
- */
- def values: Iterable[LogSegment] = segments.values.asScala
-
- /**
- * @return An iterator to all segments beginning with the segment that includes "from" and ending
- * with the segment that includes up to "to-1" or the end of the log (if to > end of log).
- */
- def values(from: Long, to: Long): Iterable[LogSegment] = {
- if (from == to) {
- // Handle non-segment-aligned empty sets
- List.empty[LogSegment]
- } else if (to < from) {
- throw new IllegalArgumentException(s"Invalid log segment range: requested segments in $topicPartition " +
- s"from offset $from which is greater than limit offset $to")
- } else {
- val view = Option(segments.floorKey(from)).map { floor =>
- segments.subMap(floor, to)
- }.getOrElse(segments.headMap(to))
- view.values.asScala
- }
- }
-
- def nonActiveLogSegmentsFrom(from: Long): Iterable[LogSegment] = {
- val activeSegment = lastSegment.get
- if (from > activeSegment.baseOffset)
- Seq.empty
- else
- values(from, activeSegment.baseOffset)
- }
-
- /**
- * @return the entry associated with the greatest offset less than or equal to the given offset,
- * if it exists.
- */
- @threadsafe
- private def floorEntry(offset: Long): Option[Map.Entry[Long, LogSegment]] = Option(segments.floorEntry(offset))
-
- /**
- * @return the log segment with the greatest offset less than or equal to the given offset,
- * if it exists.
- */
- @threadsafe
- def floorSegment(offset: Long): Option[LogSegment] = floorEntry(offset).map(_.getValue)
-
- /**
- * @return the entry associated with the greatest offset strictly less than the given offset,
- * if it exists.
- */
- @threadsafe
- private def lowerEntry(offset: Long): Option[Map.Entry[Long, LogSegment]] = Option(segments.lowerEntry(offset))
-
- /**
- * @return the log segment with the greatest offset strictly less than the given offset,
- * if it exists.
- */
- @threadsafe
- def lowerSegment(offset: Long): Option[LogSegment] = lowerEntry(offset).map(_.getValue)
-
- /**
- * @return the entry associated with the smallest offset strictly greater than the given offset,
- * if it exists.
- */
- @threadsafe
- def higherEntry(offset: Long): Option[Map.Entry[Long, LogSegment]] = Option(segments.higherEntry(offset))
-
- /**
- * @return the log segment with the smallest offset strictly greater than the given offset,
- * if it exists.
- */
- @threadsafe
- def higherSegment(offset: Long): Option[LogSegment] = higherEntry(offset).map(_.getValue)
-
- /**
- * @return the entry associated with the smallest offset, if it exists.
- */
- @threadsafe
- def firstEntry: Option[Map.Entry[Long, LogSegment]] = Option(segments.firstEntry)
-
- /**
- * @return the log segment associated with the smallest offset, if it exists.
- */
- @threadsafe
- def firstSegment: Option[LogSegment] = firstEntry.map(_.getValue)
-
- /**
- * @return the base offset of the log segment associated with the smallest offset, if it exists
- */
- private[log] def firstSegmentBaseOffset: Option[Long] = firstSegment.map(_.baseOffset)
-
- /**
- * @return the entry associated with the greatest offset, if it exists.
- */
- @threadsafe
- def lastEntry: Option[Map.Entry[Long, LogSegment]] = Option(segments.lastEntry)
-
- /**
- * @return the log segment with the greatest offset, if it exists.
- */
- @threadsafe
- def lastSegment: Option[LogSegment] = lastEntry.map(_.getValue)
-
- /**
- * @return an iterable with log segments ordered from lowest base offset to highest,
- * each segment returned has a base offset strictly greater than the provided baseOffset.
- */
- def higherSegments(baseOffset: Long): Iterable[LogSegment] = {
- val view =
- Option(segments.higherKey(baseOffset)).map {
- higherOffset => segments.tailMap(higherOffset, true)
- }.getOrElse(collection.immutable.Map[Long, LogSegment]().asJava)
- view.values.asScala
- }
-
- /**
- * The active segment that is currently taking appends
- */
- def activeSegment = lastSegment.get
-
- def sizeInBytes: Long = LogSegments.sizeInBytes(values)
-
- /**
- * Returns an Iterable containing segments matching the provided predicate.
- *
- * @param predicate the predicate to be used for filtering segments.
- */
- def filter(predicate: LogSegment => Boolean): Iterable[LogSegment] = values.filter(predicate)
-}
-
-object LogSegments {
- /**
- * Calculate a log's size (in bytes) from the provided log segments.
- *
- * @param segments The log segments to calculate the size of
- * @return Sum of the log segments' sizes (in bytes)
- */
- def sizeInBytes(segments: Iterable[LogSegment]): Long =
- segments.map(_.size.toLong).sum
-
- def getFirstBatchTimestampForSegments(segments: Iterable[LogSegment]): Iterable[Long] = {
- segments.map {
- segment =>
- segment.getFirstBatchTimestamp()
- }
- }
-}
diff --git a/core/src/main/scala/kafka/log/UnifiedLog.scala b/core/src/main/scala/kafka/log/UnifiedLog.scala
index 601d444d0e6..9246143520f 100644
--- a/core/src/main/scala/kafka/log/UnifiedLog.scala
+++ b/core/src/main/scala/kafka/log/UnifiedLog.scala
@@ -40,12 +40,13 @@ import org.apache.kafka.server.record.BrokerCompressionType
import org.apache.kafka.server.util.Scheduler
import org.apache.kafka.storage.internals.checkpoint.LeaderEpochCheckpointFile
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
-import org.apache.kafka.storage.internals.log.{AbortedTxn, AppendOrigin, BatchMetadata, CompletedTxn, EpochEntry, FetchDataInfo, FetchIsolation, LastRecord, LeaderHwChange, LogAppendInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetMetadata, LogOffsetSnapshot, LogOffsetsListener, LogStartOffsetIncrementReason, LogValidator, ProducerAppendInfo, ProducerStateManager, ProducerStateManagerConfig, RollParams}
+import org.apache.kafka.storage.internals.log.{AbortedTxn, AppendOrigin, BatchMetadata, CompletedTxn, EpochEntry, FetchDataInfo, FetchIsolation, LastRecord, LeaderHwChange, LogAppendInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetMetadata, LogOffsetSnapshot, LogOffsetsListener, LogSegment, LogSegments, LogStartOffsetIncrementReason, LogValidator, ProducerAppendInfo, ProducerStateManager, ProducerStateManagerConfig, RollParams}
import java.io.{File, IOException}
import java.nio.file.Files
import java.util
import java.util.concurrent.{ConcurrentHashMap, ConcurrentMap}
+import java.util.stream.Collectors
import java.util.{Collections, Optional, OptionalInt, OptionalLong}
import scala.annotation.nowarn
import scala.collection.mutable.ListBuffer
@@ -165,7 +166,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
initializePartitionMetadata()
updateLogStartOffset(logStartOffset)
- updateLocalLogStartOffset(math.max(logStartOffset, localLog.segments.firstSegmentBaseOffset.getOrElse(0L)))
+ updateLocalLogStartOffset(math.max(logStartOffset, localLog.segments.firstSegmentBaseOffset.orElse(0L)))
if (!remoteLogEnabled())
logStartOffset = localLogStartOffset()
maybeIncrementFirstUnstableOffset()
@@ -1313,9 +1314,9 @@ class UnifiedLog(@volatile var logStartOffset: Long,
} else if (targetTimestamp == ListOffsetsRequest.MAX_TIMESTAMP) {
// Cache to avoid race conditions. `toBuffer` is faster than most alternatives and provides
// constant time access while being safe to use with concurrent collections unlike `toArray`.
- val segmentsCopy = logSegments.toBuffer
+ val segmentsCopy = logSegments.asScala.toBuffer
val latestTimestampSegment = segmentsCopy.maxBy(_.maxTimestampSoFar)
- val latestTimestampAndOffset = latestTimestampSegment.maxTimestampAndOffsetSoFar
+ val latestTimestampAndOffset = latestTimestampSegment.readMaxTimestampAndOffsetSoFar
Some(new TimestampAndOffset(latestTimestampAndOffset.timestamp,
latestTimestampAndOffset.offset,
@@ -1347,15 +1348,15 @@ class UnifiedLog(@volatile var logStartOffset: Long,
private def searchOffsetInLocalLog(targetTimestamp: Long, startOffset: Long): Option[TimestampAndOffset] = {
// Cache to avoid race conditions. `toBuffer` is faster than most alternatives and provides
// constant time access while being safe to use with concurrent collections unlike `toArray`.
- val segmentsCopy = logSegments.toBuffer
+ val segmentsCopy = logSegments.asScala.toBuffer
val targetSeg = segmentsCopy.find(_.largestTimestamp >= targetTimestamp)
- targetSeg.flatMap(_.findOffsetByTimestamp(targetTimestamp, startOffset))
+ targetSeg.flatMap(_.findOffsetByTimestamp(targetTimestamp, startOffset).asScala)
}
def legacyFetchOffsetsBefore(timestamp: Long, maxNumOffsets: Int): Seq[Long] = {
// Cache to avoid race conditions. `toBuffer` is faster than most alternatives and provides
// constant time access while being safe to use with concurrent collections unlike `toArray`.
- val allSegments = logSegments.toBuffer
+ val allSegments = logSegments.asScala.toBuffer
val lastSegmentHasSize = allSegments.last.size > 0
val offsetTimeArray =
@@ -1463,7 +1464,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
// remove the segments for lookups
localLog.removeAndDeleteSegments(segmentsToDelete, asyncDelete = true, reason)
deleteProducerSnapshots(deletable, asyncDelete = true)
- incrementStartOffset(localLog.segments.firstSegmentBaseOffset.get, LogStartOffsetIncrementReason.SegmentDeletion)
+ incrementStartOffset(localLog.segments.firstSegmentBaseOffset.getAsLong, LogStartOffsetIncrementReason.SegmentDeletion)
}
numToDelete
}
@@ -1531,7 +1532,8 @@ class UnifiedLog(@volatile var logStartOffset: Long,
/**
* The log size in bytes for all segments that are only in local log but not yet in remote log.
*/
- def onlyLocalLogSegmentsSize: Long = UnifiedLog.sizeInBytes(logSegments.filter(_.baseOffset >= highestOffsetInRemoteStorage))
+ def onlyLocalLogSegmentsSize: Long =
+ UnifiedLog.sizeInBytes(logSegments.stream.filter(_.baseOffset >= highestOffsetInRemoteStorage).collect(Collectors.toList[LogSegment]))
/**
* The offset of the next message that will be appended to the log
@@ -1719,7 +1721,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
info(s"Truncating to offset $targetOffset")
lock synchronized {
localLog.checkIfMemoryMappedBufferClosed()
- if (localLog.segments.firstSegmentBaseOffset.get > targetOffset) {
+ if (localLog.segments.firstSegmentBaseOffset.getAsLong > targetOffset) {
truncateFullyAndStartAt(targetOffset)
} else {
val deletedSegments = localLog.truncateTo(targetOffset)
@@ -1770,17 +1772,17 @@ class UnifiedLog(@volatile var logStartOffset: Long,
/**
* All the log segments in this log ordered from oldest to newest
*/
- def logSegments: Iterable[LogSegment] = localLog.segments.values
+ def logSegments: util.Collection[LogSegment] = localLog.segments.values
/**
* Get all segments beginning with the segment that includes "from" and ending with the segment
* that includes up to "to-1" or the end of the log (if to > logEndOffset).
*/
def logSegments(from: Long, to: Long): Iterable[LogSegment] = lock synchronized {
- localLog.segments.values(from, to)
+ localLog.segments.values(from, to).asScala
}
- def nonActiveLogSegmentsFrom(from: Long): Iterable[LogSegment] = lock synchronized {
+ def nonActiveLogSegmentsFrom(from: Long): util.Collection[LogSegment] = lock synchronized {
localLog.segments.nonActiveLogSegmentsFrom(from)
}
@@ -1814,7 +1816,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
* Currently, it is used by LogCleaner threads on log compact non-active segments only with LogCleanerManager's lock
* to ensure no other logcleaner threads and retention thread can work on the same segment.
*/
- private[log] def getFirstBatchTimestampForSegments(segments: Iterable[LogSegment]): Iterable[Long] = {
+ private[log] def getFirstBatchTimestampForSegments(segments: util.Collection[LogSegment]): util.Collection[java.lang.Long] = {
LogSegments.getFirstBatchTimestampForSegments(segments)
}
@@ -1926,7 +1928,7 @@ object UnifiedLog extends Logging {
segments,
logStartOffset,
recoveryPoint,
- leaderEpochCache,
+ leaderEpochCache.asJava,
producerStateManager,
numRemainingSegments,
isRemoteLogEnabled,
@@ -1945,26 +1947,17 @@ object UnifiedLog extends Logging {
logOffsetsListener)
}
- def logFile(dir: File, offset: Long, suffix: String = ""): File = LogFileUtils.logFile(dir, offset, suffix)
-
def logDeleteDirName(topicPartition: TopicPartition): String = LocalLog.logDeleteDirName(topicPartition)
def logFutureDirName(topicPartition: TopicPartition): String = LocalLog.logFutureDirName(topicPartition)
def logDirName(topicPartition: TopicPartition): String = LocalLog.logDirName(topicPartition)
- def offsetIndexFile(dir: File, offset: Long, suffix: String = ""): File = LogFileUtils.offsetIndexFile(dir, offset, suffix)
-
- def timeIndexFile(dir: File, offset: Long, suffix: String = ""): File = LogFileUtils.timeIndexFile(dir, offset, suffix)
-
- def deleteFileIfExists(file: File, suffix: String = ""): Unit =
- Files.deleteIfExists(new File(file.getPath + suffix).toPath)
-
def transactionIndexFile(dir: File, offset: Long, suffix: String = ""): File = LogFileUtils.transactionIndexFile(dir, offset, suffix)
def offsetFromFile(file: File): Long = LogFileUtils.offsetFromFile(file)
- def sizeInBytes(segments: Iterable[LogSegment]): Long = LogSegments.sizeInBytes(segments)
+ def sizeInBytes(segments: util.Collection[LogSegment]): Long = LogSegments.sizeInBytes(segments)
def parseTopicPartitionName(dir: File): TopicPartition = LocalLog.parseTopicPartitionName(dir)
@@ -2103,7 +2096,7 @@ object UnifiedLog extends Logging {
val offsetsToSnapshot =
if (segments.nonEmpty) {
val lastSegmentBaseOffset = segments.lastSegment.get.baseOffset
- val nextLatestSegmentBaseOffset = segments.lowerSegment(lastSegmentBaseOffset).map(_.baseOffset)
+ val nextLatestSegmentBaseOffset = segments.lowerSegment(lastSegmentBaseOffset).asScala.map(_.baseOffset)
Seq(nextLatestSegmentBaseOffset, Some(lastSegmentBaseOffset), Some(lastOffset))
} else {
Seq(Some(lastOffset))
@@ -2147,14 +2140,14 @@ object UnifiedLog extends Logging {
if (lastOffset > producerStateManager.mapEndOffset && !isEmptyBeforeTruncation) {
val segmentOfLastOffset = segments.floorSegment(lastOffset)
- segments.values(producerStateManager.mapEndOffset, lastOffset).foreach { segment =>
+ segments.values(producerStateManager.mapEndOffset, lastOffset).forEach { segment =>
val startOffset = Utils.max(segment.baseOffset, producerStateManager.mapEndOffset, logStartOffset)
producerStateManager.updateMapEndOffset(startOffset)
if (offsetsToSnapshot.contains(Some(segment.baseOffset)))
producerStateManager.takeSnapshot()
- val maxPosition = if (segmentOfLastOffset.contains(segment)) {
+ val maxPosition = if (segmentOfLastOffset.isPresent && segmentOfLastOffset.get == segment) {
Option(segment.translateOffset(lastOffset))
.map(_.position)
.getOrElse(segment.size)
@@ -2162,9 +2155,7 @@ object UnifiedLog extends Logging {
segment.size
}
- val fetchDataInfo = segment.read(startOffset,
- maxSize = Int.MaxValue,
- maxPosition = maxPosition)
+ val fetchDataInfo = segment.read(startOffset, Int.MaxValue, maxPosition)
if (fetchDataInfo != null)
loadProducersFromRecords(producerStateManager, fetchDataInfo.records)
}
@@ -2264,21 +2255,20 @@ case class RetentionMsBreach(log: UnifiedLog, remoteLogEnabled: Boolean) extends
override def logReason(toDelete: List[LogSegment]): Unit = {
val retentionMs = UnifiedLog.localRetentionMs(log.config, remoteLogEnabled)
toDelete.foreach { segment =>
- segment.largestRecordTimestamp match {
- case Some(_) =>
- if (remoteLogEnabled)
- log.info(s"Deleting segment $segment due to local log retention time ${retentionMs}ms breach based on the largest " +
- s"record timestamp in the segment")
- else
- log.info(s"Deleting segment $segment due to log retention time ${retentionMs}ms breach based on the largest " +
- s"record timestamp in the segment")
- case None =>
- if (remoteLogEnabled)
- log.info(s"Deleting segment $segment due to local log retention time ${retentionMs}ms breach based on the " +
- s"last modified time of the segment")
- else
- log.info(s"Deleting segment $segment due to log retention time ${retentionMs}ms breach based on the " +
- s"last modified time of the segment")
+ if (segment.largestRecordTimestamp.isPresent)
+ if (remoteLogEnabled)
+ log.info(s"Deleting segment $segment due to local log retention time ${retentionMs}ms breach based on the largest " +
+ s"record timestamp in the segment")
+ else
+ log.info(s"Deleting segment $segment due to log retention time ${retentionMs}ms breach based on the largest " +
+ s"record timestamp in the segment")
+ else {
+ if (remoteLogEnabled)
+ log.info(s"Deleting segment $segment due to local log retention time ${retentionMs}ms breach based on the " +
+ s"last modified time of the segment")
+ else
+ log.info(s"Deleting segment $segment due to log retention time ${retentionMs}ms breach based on the " +
+ s"last modified time of the segment")
}
}
}
diff --git a/core/src/main/scala/kafka/utils/CoreUtils.scala b/core/src/main/scala/kafka/utils/CoreUtils.scala
index 8b960806042..9307a5d3996 100755
--- a/core/src/main/scala/kafka/utils/CoreUtils.scala
+++ b/core/src/main/scala/kafka/utils/CoreUtils.scala
@@ -61,16 +61,17 @@ object CoreUtils {
* @param logging The logging instance to use for logging the thrown exception.
* @param logLevel The log level to use for logging.
*/
+ @noinline // inlining this method is not typically useful and it triggers spurious spotbugs warnings
def swallow(action: => Unit, logging: Logging, logLevel: Level = Level.WARN): Unit = {
try {
action
} catch {
case e: Throwable => logLevel match {
- case Level.ERROR => logger.error(e.getMessage, e)
- case Level.WARN => logger.warn(e.getMessage, e)
- case Level.INFO => logger.info(e.getMessage, e)
- case Level.DEBUG => logger.debug(e.getMessage, e)
- case Level.TRACE => logger.trace(e.getMessage, e)
+ case Level.ERROR => logging.error(e.getMessage, e)
+ case Level.WARN => logging.warn(e.getMessage, e)
+ case Level.INFO => logging.info(e.getMessage, e)
+ case Level.DEBUG => logging.debug(e.getMessage, e)
+ case Level.TRACE => logging.trace(e.getMessage, e)
}
}
}
@@ -81,30 +82,6 @@ object CoreUtils {
*/
def delete(files: Seq[String]): Unit = files.foreach(f => Utils.delete(new File(f)))
- /**
- * Invokes every function in `all` even if one or more functions throws an exception.
- *
- * If any of the functions throws an exception, the first one will be rethrown at the end with subsequent exceptions
- * added as suppressed exceptions.
- */
- // Note that this is a generalised version of `Utils.closeAll`. We could potentially make it more general by
- // changing the signature to `def tryAll[R](all: Seq[() => R]): Seq[R]`
- def tryAll(all: Seq[() => Unit]): Unit = {
- var exception: Throwable = null
- all.foreach { element =>
- try element.apply()
- catch {
- case e: Throwable =>
- if (exception != null)
- exception.addSuppressed(e)
- else
- exception = e
- }
- }
- if (exception != null)
- throw exception
- }
-
/**
* Register the given mbean with the platform mbean server,
* unregistering any mbean that was there before. Note,
diff --git a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
index 07045ed811e..109ab874cd6 100644
--- a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
+++ b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
@@ -19,7 +19,6 @@ package kafka.log.remote;
import com.yammer.metrics.core.Gauge;
import kafka.cluster.EndPoint;
import kafka.cluster.Partition;
-import kafka.log.LogSegment;
import kafka.log.UnifiedLog;
import kafka.server.BrokerTopicStats;
import kafka.server.KafkaConfig;
@@ -65,6 +64,8 @@ import org.apache.kafka.storage.internals.log.FetchDataInfo;
import org.apache.kafka.storage.internals.log.FetchIsolation;
import org.apache.kafka.storage.internals.log.LazyIndex;
import org.apache.kafka.storage.internals.log.LogConfig;
+import org.apache.kafka.storage.internals.log.LogFileUtils;
+import org.apache.kafka.storage.internals.log.LogSegment;
import org.apache.kafka.storage.internals.log.OffsetIndex;
import org.apache.kafka.storage.internals.log.ProducerStateManager;
import org.apache.kafka.storage.internals.log.RemoteStorageFetchInfo;
@@ -428,13 +429,13 @@ public class RemoteLogManagerTest {
when(mockLog.lastStableOffset()).thenReturn(lastStableOffset);
when(mockLog.logEndOffset()).thenReturn(logEndOffset);
- LazyIndex idx = LazyIndex.forOffset(UnifiedLog.offsetIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1000);
- LazyIndex timeIdx = LazyIndex.forTime(UnifiedLog.timeIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1500);
+ OffsetIndex idx = LazyIndex.forOffset(LogFileUtils.offsetIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1000).get();
+ TimeIndex timeIdx = LazyIndex.forTime(LogFileUtils.timeIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1500).get();
File txnFile = UnifiedLog.transactionIndexFile(tempDir, oldSegmentStartOffset, "");
txnFile.createNewFile();
TransactionIndex txnIndex = new TransactionIndex(oldSegmentStartOffset, txnFile);
- when(oldSegment.lazyTimeIndex()).thenReturn(timeIdx);
- when(oldSegment.lazyOffsetIndex()).thenReturn(idx);
+ when(oldSegment.timeIndex()).thenReturn(timeIdx);
+ when(oldSegment.offsetIndex()).thenReturn(idx);
when(oldSegment.txnIndex()).thenReturn(txnIndex);
CompletableFuture<Void> dummyFuture = new CompletableFuture<>();
@@ -542,13 +543,13 @@ public class RemoteLogManagerTest {
when(mockLog.lastStableOffset()).thenReturn(lastStableOffset);
when(mockLog.logEndOffset()).thenReturn(logEndOffset);
- LazyIndex idx = LazyIndex.forOffset(UnifiedLog.offsetIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1000);
- LazyIndex timeIdx = LazyIndex.forTime(UnifiedLog.timeIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1500);
+ OffsetIndex idx = LazyIndex.forOffset(LogFileUtils.offsetIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1000).get();
+ TimeIndex timeIdx = LazyIndex.forTime(LogFileUtils.timeIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1500).get();
File txnFile = UnifiedLog.transactionIndexFile(tempDir, oldSegmentStartOffset, "");
txnFile.createNewFile();
TransactionIndex txnIndex = new TransactionIndex(oldSegmentStartOffset, txnFile);
- when(oldSegment.lazyTimeIndex()).thenReturn(timeIdx);
- when(oldSegment.lazyOffsetIndex()).thenReturn(idx);
+ when(oldSegment.timeIndex()).thenReturn(timeIdx);
+ when(oldSegment.offsetIndex()).thenReturn(idx);
when(oldSegment.txnIndex()).thenReturn(txnIndex);
int customMetadataSizeLimit = 128;
@@ -628,13 +629,13 @@ public class RemoteLogManagerTest {
when(mockStateManager.fetchSnapshot(anyLong())).thenReturn(Optional.of(mockProducerSnapshotIndex));
when(mockLog.lastStableOffset()).thenReturn(250L);
- LazyIndex idx = LazyIndex.forOffset(UnifiedLog.offsetIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1000);
- LazyIndex timeIdx = LazyIndex.forTime(UnifiedLog.timeIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1500);
+ OffsetIndex idx = LazyIndex.forOffset(LogFileUtils.offsetIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1000).get();
+ TimeIndex timeIdx = LazyIndex.forTime(LogFileUtils.timeIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1500).get();
File txnFile = UnifiedLog.transactionIndexFile(tempDir, oldSegmentStartOffset, "");
txnFile.createNewFile();
TransactionIndex txnIndex = new TransactionIndex(oldSegmentStartOffset, txnFile);
- when(oldSegment.lazyTimeIndex()).thenReturn(timeIdx);
- when(oldSegment.lazyOffsetIndex()).thenReturn(idx);
+ when(oldSegment.timeIndex()).thenReturn(timeIdx);
+ when(oldSegment.offsetIndex()).thenReturn(idx);
when(oldSegment.txnIndex()).thenReturn(txnIndex);
CompletableFuture<Void> dummyFuture = new CompletableFuture<>();
@@ -706,13 +707,13 @@ public class RemoteLogManagerTest {
when(mockStateManager.fetchSnapshot(anyLong())).thenReturn(Optional.of(mockProducerSnapshotIndex));
when(mockLog.lastStableOffset()).thenReturn(250L);
- LazyIndex idx = LazyIndex.forOffset(UnifiedLog.offsetIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1000);
- LazyIndex timeIdx = LazyIndex.forTime(UnifiedLog.timeIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1500);
+ OffsetIndex idx = LazyIndex.forOffset(LogFileUtils.offsetIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1000).get();
+ TimeIndex timeIdx = LazyIndex.forTime(LogFileUtils.timeIndexFile(tempDir, oldSegmentStartOffset, ""), oldSegmentStartOffset, 1500).get();
File txnFile = UnifiedLog.transactionIndexFile(tempDir, oldSegmentStartOffset, "");
txnFile.createNewFile();
TransactionIndex txnIndex = new TransactionIndex(oldSegmentStartOffset, txnFile);
- when(oldSegment.lazyTimeIndex()).thenReturn(timeIdx);
- when(oldSegment.lazyOffsetIndex()).thenReturn(idx);
+ when(oldSegment.timeIndex()).thenReturn(timeIdx);
+ when(oldSegment.offsetIndex()).thenReturn(idx);
when(oldSegment.txnIndex()).thenReturn(txnIndex);
CompletableFuture<Void> dummyFuture = new CompletableFuture<>();
@@ -856,8 +857,8 @@ public class RemoteLogManagerTest {
}
private void verifyLogSegmentData(LogSegmentData logSegmentData,
- LazyIndex idx,
- LazyIndex timeIdx,
+ OffsetIndex idx,
+ TimeIndex timeIdx,
TransactionIndex txnIndex,
File tempFile,
File mockProducerSnapshotIndex,
diff --git a/core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala
index cb01eaa6610..25c3af9b22d 100644
--- a/core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/GroupCoordinatorIntegrationTest.scala
@@ -48,10 +48,10 @@ class GroupCoordinatorIntegrationTest extends KafkaServerTestHarness {
def getGroupMetadataLogOpt: Option[UnifiedLog] =
logManager.getLog(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 0))
- TestUtils.waitUntilTrue(() => getGroupMetadataLogOpt.exists(_.logSegments.exists(_.log.batches.asScala.nonEmpty)),
+ TestUtils.waitUntilTrue(() => getGroupMetadataLogOpt.exists(_.logSegments.asScala.exists(_.log.batches.asScala.nonEmpty)),
"Commit message not appended in time")
- val logSegments = getGroupMetadataLogOpt.get.logSegments
+ val logSegments = getGroupMetadataLogOpt.get.logSegments.asScala
val incorrectCompressionCodecs = logSegments
.flatMap(_.log.batches.asScala.map(_.compressionType))
.filter(_ != offsetsTopicCompressionCodec)
diff --git a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
index 3bbfbbed78a..8eb27097371 100644
--- a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
+++ b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
@@ -672,7 +672,7 @@ class DynamicBrokerReconfigurationTest extends QuorumTestHarness with SaslSetup
consumerThread.waitForMatchingRecords(record => record.timestampType == TimestampType.LOG_APPEND_TIME)
// Verify that the new config is actually used for new segments of existing logs
- TestUtils.waitUntilTrue(() => log.logSegments.exists(_.size > 3000), "Log segment size increase not applied")
+ TestUtils.waitUntilTrue(() => log.logSegments.asScala.exists(_.size > 3000), "Log segment size increase not applied")
// Verify that overridden topic configs are not updated when broker default is updated
val log2 = servers.head.logManager.getLog(new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, 0))
diff --git a/core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala b/core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala
index 3b9c5f75cfb..51d31b11eac 100644
--- a/core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala
+++ b/core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala
@@ -40,6 +40,7 @@ import java.nio.ByteBuffer
import java.nio.file.{Files, Path}
import java.util
import java.util.{Collections, Optional, Properties}
+import scala.jdk.CollectionConverters._
final class KafkaMetadataLogTest {
import KafkaMetadataLogTest._
@@ -974,7 +975,7 @@ final class KafkaMetadataLogTest {
// The clean up code requires that there are at least two snapshots
// Generate first snapshots that includes the first segment by using the base offset of the second segment
val snapshotId1 = new OffsetAndEpoch(
- log.log.logSegments.drop(1).head.baseOffset,
+ log.log.logSegments.asScala.drop(1).head.baseOffset,
1
)
TestUtils.resource(log.storeSnapshot(snapshotId1).get()) { snapshot =>
@@ -982,7 +983,7 @@ final class KafkaMetadataLogTest {
}
// Generate second snapshots that includes the second segment by using the base offset of the third segment
val snapshotId2 = new OffsetAndEpoch(
- log.log.logSegments.drop(2).head.baseOffset,
+ log.log.logSegments.asScala.drop(2).head.baseOffset,
1
)
TestUtils.resource(log.storeSnapshot(snapshotId2).get()) { snapshot =>
diff --git a/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala b/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala
index 98c2a60bcfc..113c72abc50 100644
--- a/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala
+++ b/core/src/test/scala/unit/kafka/cluster/PartitionLockTest.scala
@@ -37,12 +37,13 @@ import org.apache.kafka.common.{TopicPartition, Uuid}
import org.apache.kafka.server.common.MetadataVersion
import org.apache.kafka.server.util.MockTime
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
-import org.apache.kafka.storage.internals.log.{AppendOrigin, CleanerConfig, FetchIsolation, FetchParams, LogAppendInfo, LogConfig, LogDirFailureChannel, ProducerStateManager, ProducerStateManagerConfig}
+import org.apache.kafka.storage.internals.log.{AppendOrigin, CleanerConfig, FetchIsolation, FetchParams, LogAppendInfo, LogConfig, LogDirFailureChannel, LogSegments, ProducerStateManager, ProducerStateManagerConfig}
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import org.mockito.ArgumentMatchers
import org.mockito.Mockito.{mock, when}
+import scala.compat.java8.OptionConverters._
import scala.concurrent.duration._
import scala.jdk.CollectionConverters._
@@ -317,7 +318,7 @@ class PartitionLockTest extends Logging {
segments,
0L,
0L,
- leaderEpochCache,
+ leaderEpochCache.asJava,
producerStateManager
).load()
val localLog = new LocalLog(log.dir, log.config, segments, offsets.recoveryPoint,
diff --git a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala
index 43b96d7037a..d211194777c 100644
--- a/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala
+++ b/core/src/test/scala/unit/kafka/cluster/PartitionTest.scala
@@ -56,7 +56,7 @@ import org.apache.kafka.server.common.MetadataVersion.IBP_2_6_IV0
import org.apache.kafka.server.metrics.KafkaYammerMetrics
import org.apache.kafka.server.util.{KafkaScheduler, MockTime}
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
-import org.apache.kafka.storage.internals.log.{AppendOrigin, CleanerConfig, EpochEntry, FetchIsolation, FetchParams, LogAppendInfo, LogDirFailureChannel, LogOffsetMetadata, LogReadInfo, LogStartOffsetIncrementReason, ProducerStateManager, ProducerStateManagerConfig}
+import org.apache.kafka.storage.internals.log.{AppendOrigin, CleanerConfig, EpochEntry, FetchIsolation, FetchParams, LogAppendInfo, LogDirFailureChannel, LogOffsetMetadata, LogReadInfo, LogSegments, LogStartOffsetIncrementReason, ProducerStateManager, ProducerStateManagerConfig}
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.ValueSource
@@ -451,7 +451,7 @@ class PartitionTest extends AbstractPartitionTest {
segments = segments,
logStartOffsetCheckpoint = 0L,
recoveryPointCheckpoint = 0L,
- leaderEpochCache,
+ leaderEpochCache.asJava,
producerStateManager
).load()
val localLog = new LocalLog(log.dir, log.config, segments, offsets.recoveryPoint,
diff --git a/core/src/test/scala/unit/kafka/log/AbstractLogCleanerIntegrationTest.scala b/core/src/test/scala/unit/kafka/log/AbstractLogCleanerIntegrationTest.scala
index 347cb5e6ecb..5cb66210376 100644
--- a/core/src/test/scala/unit/kafka/log/AbstractLogCleanerIntegrationTest.scala
+++ b/core/src/test/scala/unit/kafka/log/AbstractLogCleanerIntegrationTest.scala
@@ -164,4 +164,9 @@ abstract class AbstractLogCleanerIntegrationTest {
magicValue = messageFormatVersion)
(value, messageSet)
}
+
+ def closeLog(log: UnifiedLog): Unit = {
+ log.close()
+ logs -= log
+ }
}
diff --git a/core/src/test/scala/unit/kafka/log/LocalLogTest.scala b/core/src/test/scala/unit/kafka/log/LocalLogTest.scala
index 8556bc74fa5..924b8920bee 100644
--- a/core/src/test/scala/unit/kafka/log/LocalLogTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LocalLogTest.scala
@@ -29,7 +29,7 @@ import org.apache.kafka.common.errors.KafkaStorageException
import org.apache.kafka.common.record.{CompressionType, MemoryRecords, Record, SimpleRecord}
import org.apache.kafka.common.utils.{Time, Utils}
import org.apache.kafka.server.util.{MockTime, Scheduler}
-import org.apache.kafka.storage.internals.log.{FetchDataInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetMetadata}
+import org.apache.kafka.storage.internals.log.{FetchDataInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetMetadata, LogSegment, LogSegments}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.function.Executable
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
@@ -124,7 +124,7 @@ class LocalLogTest {
log.roll()
assertEquals(2, log.segments.numberOfSegments)
assertFalse(logDir.listFiles.isEmpty)
- val segmentsBeforeDelete = List[LogSegment]() ++ log.segments.values
+ val segmentsBeforeDelete = log.segments.values.asScala.toVector
val deletedSegments = log.deleteAllSegments()
assertTrue(log.segments.isEmpty)
assertEquals(segmentsBeforeDelete, deletedSegments)
@@ -181,7 +181,7 @@ class LocalLogTest {
assertEquals(newLogDir, log.dir)
assertEquals(newLogDir.getParent, log.parentDir)
assertEquals(newLogDir.getParent, log.dir.getParent)
- log.segments.values.foreach(segment => assertEquals(newLogDir.getPath, segment.log.file().getParentFile.getPath))
+ log.segments.values.forEach(segment => assertEquals(newLogDir.getPath, segment.log.file().getParentFile.getPath))
assertEquals(2, log.segments.numberOfSegments)
}
@@ -278,7 +278,7 @@ class LocalLogTest {
def deletedSegments: Iterable[LogSegment] = _deletedSegments
}
val reason = new TestDeletionReason()
- val toDelete = List[LogSegment]() ++ log.segments.values
+ val toDelete = log.segments.values.asScala.toVector
log.removeAndDeleteSegments(toDelete, asyncDelete = asyncDelete, reason)
if (asyncDelete) {
mockTime.sleep(log.config.fileDeleteDelayMs + 1)
@@ -307,7 +307,7 @@ class LocalLogTest {
assertEquals(10L, log.segments.numberOfSegments)
- val toDelete = List[LogSegment]() ++ log.segments.values
+ val toDelete = log.segments.values.asScala.toVector
LocalLog.deleteSegmentFiles(toDelete, asyncDelete = asyncDelete, log.dir, log.topicPartition, log.config, log.scheduler, log.logDirFailureChannel, "")
if (asyncDelete) {
toDelete.foreach {
@@ -343,14 +343,14 @@ class LocalLogTest {
{
val deletable = log.deletableSegments(
(segment: LogSegment, _: Option[LogSegment]) => segment.baseOffset <= 5)
- val expected = log.segments.nonActiveLogSegmentsFrom(0L).filter(segment => segment.baseOffset <= 5).toList
+ val expected = log.segments.nonActiveLogSegmentsFrom(0L).asScala.filter(segment => segment.baseOffset <= 5).toList
assertEquals(6, expected.length)
assertEquals(expected, deletable.toList)
}
{
val deletable = log.deletableSegments((_: LogSegment, _: Option[LogSegment]) => true)
- val expected = log.segments.nonActiveLogSegmentsFrom(0L).toList
+ val expected = log.segments.nonActiveLogSegmentsFrom(0L).asScala.toList
assertEquals(9, expected.length)
assertEquals(expected, deletable.toList)
}
@@ -359,7 +359,7 @@ class LocalLogTest {
val record = new SimpleRecord(mockTime.milliseconds, "a".getBytes)
appendRecords(List(record), initialOffset = 9L)
val deletable = log.deletableSegments((_: LogSegment, _: Option[LogSegment]) => true)
- val expected = log.segments.values.toList
+ val expected = log.segments.values.asScala.toList
assertEquals(10, expected.length)
assertEquals(expected, deletable.toList)
}
@@ -380,14 +380,14 @@ class LocalLogTest {
(segment: LogSegment, nextSegmentOpt: Option[LogSegment]) => {
assertEquals(offset, segment.baseOffset)
val floorSegmentOpt = log.segments.floorSegment(offset)
- assertTrue(floorSegmentOpt.isDefined)
+ assertTrue(floorSegmentOpt.isPresent)
assertEquals(floorSegmentOpt.get, segment)
if (offset == log.logEndOffset) {
assertFalse(nextSegmentOpt.isDefined)
} else {
assertTrue(nextSegmentOpt.isDefined)
val higherSegmentOpt = log.segments.higherSegment(segment.baseOffset)
- assertTrue(higherSegmentOpt.isDefined)
+ assertTrue(higherSegmentOpt.isPresent)
assertEquals(segment.baseOffset + 1, higherSegmentOpt.get.baseOffset)
assertEquals(higherSegmentOpt.get, nextSegmentOpt.get)
}
@@ -395,7 +395,7 @@ class LocalLogTest {
true
})
assertEquals(10L, log.segments.numberOfSegments)
- assertEquals(log.segments.nonActiveLogSegmentsFrom(0L).toSeq, deletableSegments.toSeq)
+ assertEquals(log.segments.nonActiveLogSegmentsFrom(0L).asScala.toSeq, deletableSegments.toSeq)
}
@Test
@@ -430,7 +430,7 @@ class LocalLogTest {
}
assertEquals(5, log.segments.numberOfSegments)
assertNotEquals(10L, log.segments.activeSegment.baseOffset)
- val expected = List[LogSegment]() ++ log.segments.values
+ val expected = log.segments.values.asScala.toVector
val deleted = log.truncateFullyAndStartAt(10L)
assertEquals(expected, deleted)
assertEquals(1, log.segments.numberOfSegments)
@@ -452,10 +452,10 @@ class LocalLogTest {
assertEquals(5, log.segments.numberOfSegments)
assertEquals(12L, log.logEndOffset)
- val expected = List[LogSegment]() ++ log.segments.values(9L, log.logEndOffset + 1)
+ val expected = log.segments.values(9L, log.logEndOffset + 1).asScala.toVector
// Truncate to an offset before the base offset of the active segment
val deleted = log.truncateTo(7L)
- assertEquals(expected, deleted)
+ assertEquals(expected, deleted.toVector)
assertEquals(3, log.segments.numberOfSegments)
assertEquals(6L, log.segments.activeSegment.baseOffset)
assertEquals(0L, log.recoveryPoint)
@@ -479,7 +479,7 @@ class LocalLogTest {
}
def nonActiveBaseOffsetsFrom(startOffset: Long): Seq[Long] = {
- log.segments.nonActiveLogSegmentsFrom(startOffset).map(_.baseOffset).toSeq
+ log.segments.nonActiveLogSegmentsFrom(startOffset).asScala.map(_.baseOffset).toSeq
}
assertEquals(5L, log.segments.activeSegment.baseOffset)
@@ -726,12 +726,12 @@ class LocalLogTest {
time: Time = mockTime,
topicPartition: TopicPartition = topicPartition,
logDirFailureChannel: LogDirFailureChannel = logDirFailureChannel): LocalLog = {
- segments.add(LogSegment.open(dir = dir,
- baseOffset = 0L,
+ segments.add(LogSegment.open(dir,
+ 0L,
config,
- time = time,
- initFileSize = config.initFileSize,
- preallocate = config.preallocate))
+ time,
+ config.initFileSize,
+ config.preallocate))
new LocalLog(_dir = dir,
config = config,
segments = segments,
diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerIntegrationTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerIntegrationTest.scala
index 925ab79825c..b85014ca50a 100644
--- a/core/src/test/scala/unit/kafka/log/LogCleanerIntegrationTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogCleanerIntegrationTest.scala
@@ -56,7 +56,7 @@ class LogCleanerIntegrationTest extends AbstractLogCleanerIntegrationTest {
val log = cleaner.logs.get(tp)
writeDups(numKeys = 20, numDups = 3, log = log, codec = codec)
- val partitionFile = log.logSegments.last.log.file()
+ val partitionFile = log.logSegments.asScala.last.log.file()
val writer = new PrintWriter(partitionFile)
writer.write("jogeajgoea")
writer.close()
@@ -76,8 +76,8 @@ class LogCleanerIntegrationTest extends AbstractLogCleanerIntegrationTest {
val uncleanableBytesGauge = getGauge[Long]("uncleanable-bytes", uncleanableDirectory)
TestUtils.waitUntilTrue(() => uncleanablePartitionsCountGauge.value() == 2, "There should be 2 uncleanable partitions", 2000L)
- val expectedTotalUncleanableBytes = LogCleanerManager.calculateCleanableBytes(log, 0, log.logSegments.last.baseOffset)._2 +
- LogCleanerManager.calculateCleanableBytes(log2, 0, log2.logSegments.last.baseOffset)._2
+ val expectedTotalUncleanableBytes = LogCleanerManager.calculateCleanableBytes(log, 0, log.logSegments.asScala.last.baseOffset)._2 +
+ LogCleanerManager.calculateCleanableBytes(log2, 0, log2.logSegments.asScala.last.baseOffset)._2
TestUtils.waitUntilTrue(() => uncleanableBytesGauge.value() == expectedTotalUncleanableBytes,
s"There should be $expectedTotalUncleanableBytes uncleanable bytes", 1000L)
@@ -192,7 +192,7 @@ class LogCleanerIntegrationTest extends AbstractLogCleanerIntegrationTest {
}
private def readFromLog(log: UnifiedLog): Iterable[(Int, Int)] = {
- for (segment <- log.logSegments; record <- segment.log.records.asScala) yield {
+ for (segment <- log.logSegments.asScala; record <- segment.log.records.asScala) yield {
val key = TestUtils.readString(record.key).toInt
val value = TestUtils.readString(record.value).toInt
key -> value
diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerLagIntegrationTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerLagIntegrationTest.scala
index b2cc21933a2..4ccbb3880a4 100644
--- a/core/src/test/scala/unit/kafka/log/LogCleanerLagIntegrationTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogCleanerLagIntegrationTest.scala
@@ -95,7 +95,7 @@ class LogCleanerLagIntegrationTest extends AbstractLogCleanerIntegrationTest wit
}
private def readFromLog(log: UnifiedLog): Iterable[(Int, Int)] = {
- for (segment <- log.logSegments; record <- segment.log.records.asScala) yield {
+ for (segment <- log.logSegments.asScala; record <- segment.log.records.asScala) yield {
val key = TestUtils.readString(record.key).toInt
val value = TestUtils.readString(record.value).toInt
key -> value
diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala
index f11b315faaf..9656ae67a6e 100644
--- a/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogCleanerManagerTest.scala
@@ -27,11 +27,13 @@ import org.apache.kafka.common.config.TopicConfig
import org.apache.kafka.common.record._
import org.apache.kafka.common.utils.Utils
import org.apache.kafka.server.util.MockTime
-import org.apache.kafka.storage.internals.log.{AppendOrigin, LogConfig, LogDirFailureChannel, LogStartOffsetIncrementReason, ProducerStateManager, ProducerStateManagerConfig}
+import org.apache.kafka.storage.internals.log.{AppendOrigin, LogConfig, LogDirFailureChannel, LogSegment, LogSegments, LogStartOffsetIncrementReason, ProducerStateManager, ProducerStateManagerConfig}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, Test}
+import java.util
import scala.collection.mutable
+import scala.compat.java8.OptionConverters._
/**
* Unit tests for the log cleaning logic
@@ -117,7 +119,7 @@ class LogCleanerManagerTest extends Logging {
segments,
0L,
0L,
- leaderEpochCache,
+ leaderEpochCache.asJava,
producerStateManager
).load()
val localLog = new LocalLog(tpDir, config, segments, offsets.recoveryPoint,
@@ -127,7 +129,7 @@ class LogCleanerManagerTest extends Logging {
producerIdExpirationCheckIntervalMs, leaderEpochCache,
producerStateManager, _topicId = None, keepPartitionMetadataFile = true) {
// Throw an error in getFirstBatchTimestampForSegments since it is called in grabFilthiestLog()
- override def getFirstBatchTimestampForSegments(segments: Iterable[LogSegment]): Iterable[Long] =
+ override def getFirstBatchTimestampForSegments(segments: util.Collection[LogSegment]): util.Collection[java.lang.Long] =
throw new IllegalStateException("Error!")
}
diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerParameterizedIntegrationTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerParameterizedIntegrationTest.scala
index 6946d8199a5..8729045db7e 100755
--- a/core/src/test/scala/unit/kafka/log/LogCleanerParameterizedIntegrationTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogCleanerParameterizedIntegrationTest.scala
@@ -62,7 +62,7 @@ class LogCleanerParameterizedIntegrationTest extends AbstractLogCleanerIntegrati
val firstDirty = log.activeSegment.baseOffset
checkLastCleaned("log", 0, firstDirty)
- val compactedSize = log.logSegments.map(_.size).sum
+ val compactedSize = log.logSegments.asScala.map(_.size).sum
assertTrue(startSize > compactedSize, s"log should have been compacted: startSize=$startSize compactedSize=$compactedSize")
checkLogAfterAppendingDups(log, startSize, appends)
@@ -111,7 +111,7 @@ class LogCleanerParameterizedIntegrationTest extends AbstractLogCleanerIntegrati
// should compact the log
checkLastCleaned("log", 0, firstDirty)
- val compactedSize = log.logSegments.map(_.size).sum
+ val compactedSize = log.logSegments.asScala.map(_.size).sum
assertTrue(startSize > compactedSize, s"log should have been compacted: startSize=$startSize compactedSize=$compactedSize")
(log, messages)
}
@@ -120,12 +120,13 @@ class LogCleanerParameterizedIntegrationTest extends AbstractLogCleanerIntegrati
// Set the last modified time to an old value to force deletion of old segments
val endOffset = log.logEndOffset
- log.logSegments.foreach(_.lastModified = time.milliseconds - (2 * retentionMs))
+ log.logSegments.forEach(_.setLastModified(time.milliseconds - (2 * retentionMs)))
TestUtils.waitUntilTrue(() => log.logStartOffset == endOffset,
"Timed out waiting for deletion of old segments")
assertEquals(1, log.numberOfSegments)
cleaner.shutdown()
+ closeLog(log)
// run the cleaner again to make sure if there are no issues post deletion
val (log2, messages) = runCleanerAndCheckCompacted(20)
@@ -162,7 +163,7 @@ class LogCleanerParameterizedIntegrationTest extends AbstractLogCleanerIntegrati
val firstDirty = log.activeSegment.baseOffset
checkLastCleaned("log", 0, firstDirty)
- val compactedSize = log.logSegments.map(_.size).sum
+ val compactedSize = log.logSegments.asScala.map(_.size).sum
assertTrue(startSize > compactedSize, s"log should have been compacted: startSize=$startSize compactedSize=$compactedSize")
checkLogAfterAppendingDups(log, startSize, appends)
@@ -220,7 +221,7 @@ class LogCleanerParameterizedIntegrationTest extends AbstractLogCleanerIntegrati
assertTrue(firstDirty > appendsV0.size) // ensure we clean data from V0 and V1
checkLastCleaned("log", 0, firstDirty)
- val compactedSize = log.logSegments.map(_.size).sum
+ val compactedSize = log.logSegments.asScala.map(_.size).sum
assertTrue(startSize > compactedSize, s"log should have been compacted: startSize=$startSize compactedSize=$compactedSize")
checkLogAfterAppendingDups(log, startSize, appends)
@@ -274,7 +275,7 @@ class LogCleanerParameterizedIntegrationTest extends AbstractLogCleanerIntegrati
assertEquals(2, cleaner.cleanerCount)
checkLastCleaned("log", 0, firstDirty)
- val compactedSize = log.logSegments.map(_.size).sum
+ val compactedSize = log.logSegments.asScala.map(_.size).sum
assertTrue(startSize > compactedSize, s"log should have been compacted: startSize=$startSize compactedSize=$compactedSize")
}
@@ -298,7 +299,7 @@ class LogCleanerParameterizedIntegrationTest extends AbstractLogCleanerIntegrati
}
private def readFromLog(log: UnifiedLog): Iterable[(Int, String, Long)] = {
- for (segment <- log.logSegments; deepLogEntry <- segment.log.records.asScala) yield {
+ for (segment <- log.logSegments.asScala; deepLogEntry <- segment.log.records.asScala) yield {
val key = TestUtils.readString(deepLogEntry.key).toInt
val value = TestUtils.readString(deepLogEntry.value)
(key, value, deepLogEntry.offset)
diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
index 7a221ad3310..9e87c336064 100644
--- a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
@@ -27,7 +27,7 @@ import org.apache.kafka.common.record._
import org.apache.kafka.common.utils.Utils
import org.apache.kafka.server.metrics.KafkaMetricsGroup
import org.apache.kafka.server.util.MockTime
-import org.apache.kafka.storage.internals.log.{AbortedTxn, AppendOrigin, CleanerConfig, LogAppendInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogStartOffsetIncrementReason, OffsetMap, ProducerStateManager, ProducerStateManagerConfig}
+import org.apache.kafka.storage.internals.log.{AbortedTxn, AppendOrigin, CleanerConfig, LogAppendInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogSegment, LogSegments, LogStartOffsetIncrementReason, OffsetMap, ProducerStateManager, ProducerStateManagerConfig}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, Test}
import org.mockito.ArgumentMatchers
@@ -41,6 +41,7 @@ import java.nio.file.Paths
import java.util.Properties
import java.util.concurrent.{CountDownLatch, TimeUnit}
import scala.collection._
+import scala.compat.java8.OptionConverters._
import scala.jdk.CollectionConverters._
/**
@@ -138,7 +139,7 @@ class LogCleanerTest {
keys.foreach(k => map.put(key(k), Long.MaxValue))
// clean the log
- val segments = log.logSegments.take(3).toSeq
+ val segments = log.logSegments.asScala.take(3).toSeq
val stats = new CleanerStats()
val expectedBytesRead = segments.map(_.size).sum
val shouldRemain = LogTestUtils.keysInLog(log).filterNot(keys.contains)
@@ -177,7 +178,7 @@ class LogCleanerTest {
logSegments,
0L,
0L,
- leaderEpochCache,
+ leaderEpochCache.asJava,
producerStateManager
).load()
val localLog = new LocalLog(dir, config, logSegments, offsets.recoveryPoint,
@@ -220,7 +221,7 @@ class LogCleanerTest {
assertEquals(3, log.numberOfSegments)
// Remember reference to the first log and determine its file name expected for async deletion
- val firstLogFile = log.logSegments.head.log
+ val firstLogFile = log.logSegments.asScala.head.log
val expectedFileName = Utils.replaceSuffix(firstLogFile.file.getPath, "", LogFileUtils.DELETED_FILE_SUFFIX)
// Clean the log. This should trigger replaceSegments() and deleteOldSegments();
@@ -898,7 +899,7 @@ class LogCleanerTest {
// clean the log
val stats = new CleanerStats()
- cleaner.cleanSegments(log, Seq(log.logSegments.head), map, 0L, stats, new CleanedTransactionMetadata, -1)
+ cleaner.cleanSegments(log, Seq(log.logSegments.asScala.head), map, 0L, stats, new CleanedTransactionMetadata, -1)
val shouldRemain = LogTestUtils.keysInLog(log).filterNot(keys.contains)
assertEquals(shouldRemain, LogTestUtils.keysInLog(log))
}
@@ -911,7 +912,7 @@ class LogCleanerTest {
val (log, offsetMap) = createLogWithMessagesLargerThanMaxSize(largeMessageSize = 1024 * 1024)
val cleaner = makeCleaner(Int.MaxValue, maxMessageSize=1024)
- cleaner.cleanSegments(log, Seq(log.logSegments.head), offsetMap, 0L, new CleanerStats, new CleanedTransactionMetadata, -1)
+ cleaner.cleanSegments(log, Seq(log.logSegments.asScala.head), offsetMap, 0L, new CleanerStats, new CleanedTransactionMetadata, -1)
val shouldRemain = LogTestUtils.keysInLog(log).filter(k => !offsetMap.map.containsKey(k.toString))
assertEquals(shouldRemain, LogTestUtils.keysInLog(log))
}
@@ -923,14 +924,14 @@ class LogCleanerTest {
@Test
def testMessageLargerThanMaxMessageSizeWithCorruptHeader(): Unit = {
val (log, offsetMap) = createLogWithMessagesLargerThanMaxSize(largeMessageSize = 1024 * 1024)
- val file = new RandomAccessFile(log.logSegments.head.log.file, "rw")
+ val file = new RandomAccessFile(log.logSegments.asScala.head.log.file, "rw")
file.seek(Records.MAGIC_OFFSET)
file.write(0xff)
file.close()
val cleaner = makeCleaner(Int.MaxValue, maxMessageSize=1024)
assertThrows(classOf[CorruptRecordException], () =>
- cleaner.cleanSegments(log, Seq(log.logSegments.head), offsetMap, 0L, new CleanerStats, new CleanedTransactionMetadata, -1)
+ cleaner.cleanSegments(log, Seq(log.logSegments.asScala.head), offsetMap, 0L, new CleanerStats, new CleanedTransactionMetadata, -1)
)
}
@@ -941,13 +942,13 @@ class LogCleanerTest {
@Test
def testCorruptMessageSizeLargerThanBytesAvailable(): Unit = {
val (log, offsetMap) = createLogWithMessagesLargerThanMaxSize(largeMessageSize = 1024 * 1024)
- val file = new RandomAccessFile(log.logSegments.head.log.file, "rw")
+ val file = new RandomAccessFile(log.logSegments.asScala.head.log.file, "rw")
file.setLength(1024)
file.close()
val cleaner = makeCleaner(Int.MaxValue, maxMessageSize=1024)
assertThrows(classOf[CorruptRecordException], () =>
- cleaner.cleanSegments(log, Seq(log.logSegments.head), offsetMap, 0L, new CleanerStats, new CleanedTransactionMetadata, -1)
+ cleaner.cleanSegments(log, Seq(log.logSegments.asScala.head), offsetMap, 0L, new CleanerStats, new CleanedTransactionMetadata, -1)
)
}
@@ -1188,7 +1189,7 @@ class LogCleanerTest {
// the last (active) segment has just one message
- def distinctValuesBySegment = log.logSegments.map(s => s.log.records.asScala.map(record => TestUtils.readString(record.value)).toSet.size).toSeq
+ def distinctValuesBySegment = log.logSegments.asScala.map(s => s.log.records.asScala.map(record => TestUtils.readString(record.value)).toSet.size).toSeq
val disctinctValuesBySegmentBeforeClean = distinctValuesBySegment
assertTrue(distinctValuesBySegment.reverse.tail.forall(_ > N),
@@ -1236,7 +1237,7 @@ class LogCleanerTest {
log.appendAsLeader(createRecords, leaderEpoch = 0)
// segments [0,1] are clean; segments [2, 3] are cleanable; segments [4,5] are uncleanable
- val segs = log.logSegments.toSeq
+ val segs = log.logSegments.asScala.toSeq
val logToClean = LogToClean(new TopicPartition("test", 0), log, segs(2).baseOffset, segs(4).baseOffset)
val expectedCleanSize = segs.take(2).map(_.size).sum
@@ -1285,27 +1286,27 @@ class LogCleanerTest {
}
private def batchBaseOffsetsInLog(log: UnifiedLog): Iterable[Long] = {
- for (segment <- log.logSegments; batch <- segment.log.batches.asScala)
+ for (segment <- log.logSegments.asScala; batch <- segment.log.batches.asScala)
yield batch.baseOffset
}
def lastOffsetsPerBatchInLog(log: UnifiedLog): Iterable[Long] = {
- for (segment <- log.logSegments; batch <- segment.log.batches.asScala)
+ for (segment <- log.logSegments.asScala; batch <- segment.log.batches.asScala)
yield batch.lastOffset
}
def lastSequencesInLog(log: UnifiedLog): Map[Long, Int] = {
- (for (segment <- log.logSegments;
+ (for (segment <- log.logSegments.asScala;
batch <- segment.log.batches.asScala if !batch.isControlBatch && batch.hasProducerId)
yield batch.producerId -> batch.lastSequence).toMap
}
/* extract all the offsets from a log */
def offsetsInLog(log: UnifiedLog): Iterable[Long] =
- log.logSegments.flatMap(s => s.log.records.asScala.filter(_.hasValue).filter(_.hasKey).map(m => m.offset))
+ log.logSegments.asScala.flatMap(s => s.log.records.asScala.filter(_.hasValue).filter(_.hasKey).map(m => m.offset))
def unkeyedMessageCountInLog(log: UnifiedLog) =
- log.logSegments.map(s => s.log.records.asScala.filter(_.hasValue).count(m => !m.hasKey)).sum
+ log.logSegments.asScala.map(s => s.log.records.asScala.filter(_.hasValue).count(m => !m.hasKey)).sum
def abortCheckDone(topicPartition: TopicPartition): Unit = {
throw new LogCleaningAbortedException()
@@ -1330,7 +1331,7 @@ class LogCleanerTest {
val map = new FakeOffsetMap(Int.MaxValue)
keys.foreach(k => map.put(key(k), Long.MaxValue))
assertThrows(classOf[LogCleaningAbortedException], () =>
- cleaner.cleanSegments(log, log.logSegments.take(3).toSeq, map, 0L, new CleanerStats(),
+ cleaner.cleanSegments(log, log.logSegments.asScala.take(3).toSeq, map, 0L, new CleanerStats(),
new CleanedTransactionMetadata, -1)
)
}
@@ -1355,17 +1356,17 @@ class LogCleanerTest {
}
// grouping by very large values should result in a single group with all the segments in it
- var groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize = Int.MaxValue, log.logEndOffset)
+ var groups = cleaner.groupSegmentsBySize(log.logSegments.asScala, maxSize = Int.MaxValue, maxIndexSize = Int.MaxValue, log.logEndOffset)
assertEquals(1, groups.size)
assertEquals(log.numberOfSegments, groups.head.size)
checkSegmentOrder(groups)
// grouping by very small values should result in all groups having one entry
- groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = 1, maxIndexSize = Int.MaxValue, log.logEndOffset)
+ groups = cleaner.groupSegmentsBySize(log.logSegments.asScala, maxSize = 1, maxIndexSize = Int.MaxValue, log.logEndOffset)
assertEquals(log.numberOfSegments, groups.size)
assertTrue(groups.forall(_.size == 1), "All groups should be singletons.")
checkSegmentOrder(groups)
- groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize = 1, log.logEndOffset)
+ groups = cleaner.groupSegmentsBySize(log.logSegments.asScala, maxSize = Int.MaxValue, maxIndexSize = 1, log.logEndOffset)
assertEquals(log.numberOfSegments, groups.size)
assertTrue(groups.forall(_.size == 1), "All groups should be singletons.")
checkSegmentOrder(groups)
@@ -1373,14 +1374,14 @@ class LogCleanerTest {
val groupSize = 3
// check grouping by log size
- val logSize = log.logSegments.take(groupSize).map(_.size).sum.toInt + 1
- groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = logSize, maxIndexSize = Int.MaxValue, log.logEndOffset)
+ val logSize = log.logSegments.asScala.take(groupSize).map(_.size).sum.toInt + 1
+ groups = cleaner.groupSegmentsBySize(log.logSegments.asScala, maxSize = logSize, maxIndexSize = Int.MaxValue, log.logEndOffset)
checkSegmentOrder(groups)
assertTrue(groups.dropRight(1).forall(_.size == groupSize), "All but the last group should be the target size.")
// check grouping by index size
- val indexSize = log.logSegments.take(groupSize).map(_.offsetIndex.sizeInBytes).sum + 1
- groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize = indexSize, log.logEndOffset)
+ val indexSize = log.logSegments.asScala.take(groupSize).map(_.offsetIndex.sizeInBytes).sum + 1
+ groups = cleaner.groupSegmentsBySize(log.logSegments.asScala, maxSize = Int.MaxValue, maxIndexSize = indexSize, log.logEndOffset)
checkSegmentOrder(groups)
assertTrue(groups.dropRight(1).forall(_.size == groupSize),
"All but the last group should be the target size.")
@@ -1413,17 +1414,17 @@ class LogCleanerTest {
val notCleanableSegments = 1
assertEquals(totalSegments, log.numberOfSegments)
- var groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize = Int.MaxValue, firstUncleanableOffset)
+ var groups = cleaner.groupSegmentsBySize(log.logSegments.asScala, maxSize = Int.MaxValue, maxIndexSize = Int.MaxValue, firstUncleanableOffset)
//because index file uses 4 byte relative index offset and current segments all none empty,
//segments will not group even their size is very small.
assertEquals(totalSegments - notCleanableSegments, groups.size)
//do clean to clean first 2 segments to empty
cleaner.clean(LogToClean(log.topicPartition, log, 0, firstUncleanableOffset))
assertEquals(totalSegments, log.numberOfSegments)
- assertEquals(0, log.logSegments.head.size)
+ assertEquals(0, log.logSegments.asScala.head.size)
//after clean we got 2 empty segment, they will group together this time
- groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize = Int.MaxValue, firstUncleanableOffset)
+ groups = cleaner.groupSegmentsBySize(log.logSegments.asScala, maxSize = Int.MaxValue, maxIndexSize = Int.MaxValue, firstUncleanableOffset)
val noneEmptySegment = 1
assertEquals(noneEmptySegment + 1, groups.size)
@@ -1459,14 +1460,14 @@ class LogCleanerTest {
assertEquals(Int.MaxValue, log.activeSegment.offsetIndex.lastOffset)
// grouping should result in a single group with maximum relative offset of Int.MaxValue
- var groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize = Int.MaxValue, log.logEndOffset)
+ var groups = cleaner.groupSegmentsBySize(log.logSegments.asScala, maxSize = Int.MaxValue, maxIndexSize = Int.MaxValue, log.logEndOffset)
assertEquals(1, groups.size)
// append another message, making last offset of second segment > Int.MaxValue
log.appendAsLeader(TestUtils.singletonRecords(value = "hello".getBytes, key = "hello".getBytes), leaderEpoch = 0)
// grouping should not group the two segments to ensure that maximum relative offset in each group <= Int.MaxValue
- groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize = Int.MaxValue, log.logEndOffset)
+ groups = cleaner.groupSegmentsBySize(log.logSegments.asScala, maxSize = Int.MaxValue, maxIndexSize = Int.MaxValue, log.logEndOffset)
assertEquals(2, groups.size)
checkSegmentOrder(groups)
@@ -1474,7 +1475,7 @@ class LogCleanerTest {
while (log.numberOfSegments < 4)
log.appendAsLeader(TestUtils.singletonRecords(value = "hello".getBytes, key = "hello".getBytes), leaderEpoch = 0)
- groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize = Int.MaxValue, log.logEndOffset)
+ groups = cleaner.groupSegmentsBySize(log.logSegments.asScala, maxSize = Int.MaxValue, maxIndexSize = Int.MaxValue, log.logEndOffset)
assertEquals(log.numberOfSegments - 1, groups.size)
for (group <- groups)
assertTrue(group.last.offsetIndex.lastOffset - group.head.offsetIndex.baseOffset <= Int.MaxValue,
@@ -1513,11 +1514,11 @@ class LogCleanerTest {
log.appendAsFollower(record4)
assertTrue(log.logEndOffset - 1 - log.logStartOffset > Int.MaxValue, "Actual offset range should be > Int.MaxValue")
- assertTrue(log.logSegments.last.offsetIndex.lastOffset - log.logStartOffset <= Int.MaxValue,
+ assertTrue(log.logSegments.asScala.last.offsetIndex.lastOffset - log.logStartOffset <= Int.MaxValue,
"index.lastOffset is reporting the wrong last offset")
// grouping should result in two groups because the second segment takes the offset range > MaxInt
- val groups = cleaner.groupSegmentsBySize(log.logSegments, maxSize = Int.MaxValue, maxIndexSize = Int.MaxValue, log.logEndOffset)
+ val groups = cleaner.groupSegmentsBySize(log.logSegments.asScala, maxSize = Int.MaxValue, maxIndexSize = Int.MaxValue, log.logEndOffset)
assertEquals(2, groups.size)
for (group <- groups)
@@ -1556,7 +1557,7 @@ class LogCleanerTest {
assertEquals(end - start, stats.mapMessagesRead)
}
- val segments = log.logSegments.toSeq
+ val segments = log.logSegments.asScala.toSeq
checkRange(map, 0, segments(1).baseOffset.toInt)
checkRange(map, segments(1).baseOffset.toInt, segments(3).baseOffset.toInt)
checkRange(map, segments(3).baseOffset.toInt, log.logEndOffset.toInt)
@@ -1598,7 +1599,7 @@ class LogCleanerTest {
assertFalse(LogTestUtils.hasOffsetOverflow(log))
// Clean each segment now that split is complete.
- for (segmentToClean <- log.logSegments)
+ for (segmentToClean <- log.logSegments.asScala)
cleaner.cleanSegments(log, List(segmentToClean), offsetMap, 0L, new CleanerStats(),
new CleanedTransactionMetadata, -1)
assertEquals(expectedKeysAfterCleaning, LogTestUtils.keysInLog(log))
@@ -1640,7 +1641,7 @@ class LogCleanerTest {
offsetMap.put(key(k), Long.MaxValue)
// clean the log
- cleaner.cleanSegments(log, log.logSegments.take(9).toSeq, offsetMap, 0L, new CleanerStats(),
+ cleaner.cleanSegments(log, log.logSegments.asScala.take(9).toSeq, offsetMap, 0L, new CleanerStats(),
new CleanedTransactionMetadata, -1)
// clear scheduler so that async deletes don't run
time.scheduler.clear()
@@ -1649,14 +1650,14 @@ class LogCleanerTest {
// 1) Simulate recovery just after .cleaned file is created, before rename to .swap
// On recovery, clean operation is aborted. All messages should be present in the log
- log.logSegments.head.changeFileSuffixes("", UnifiedLog.CleanedFileSuffix)
+ log.logSegments.asScala.head.changeFileSuffixes("", UnifiedLog.CleanedFileSuffix)
for (file <- dir.listFiles if file.getName.endsWith(LogFileUtils.DELETED_FILE_SUFFIX)) {
Utils.atomicMoveWithFallback(file.toPath, Paths.get(Utils.replaceSuffix(file.getPath, LogFileUtils.DELETED_FILE_SUFFIX, "")), false)
}
log = recoverAndCheck(config, allKeys)
// clean again
- cleaner.cleanSegments(log, log.logSegments.take(9).toSeq, offsetMap, 0L, new CleanerStats(),
+ cleaner.cleanSegments(log, log.logSegments.asScala.take(9).toSeq, offsetMap, 0L, new CleanerStats(),
new CleanedTransactionMetadata, -1)
// clear scheduler so that async deletes don't run
time.scheduler.clear()
@@ -1665,15 +1666,15 @@ class LogCleanerTest {
// 2) Simulate recovery just after .cleaned file is created, and a subset of them are renamed to .swap
// On recovery, clean operation is aborted. All messages should be present in the log
- log.logSegments.head.changeFileSuffixes("", UnifiedLog.CleanedFileSuffix)
- log.logSegments.head.log.renameTo(new File(Utils.replaceSuffix(log.logSegments.head.log.file.getPath, UnifiedLog.CleanedFileSuffix, UnifiedLog.SwapFileSuffix)))
+ log.logSegments.asScala.head.changeFileSuffixes("", UnifiedLog.CleanedFileSuffix)
+ log.logSegments.asScala.head.log.renameTo(new File(Utils.replaceSuffix(log.logSegments.asScala.head.log.file.getPath, UnifiedLog.CleanedFileSuffix, UnifiedLog.SwapFileSuffix)))
for (file <- dir.listFiles if file.getName.endsWith(LogFileUtils.DELETED_FILE_SUFFIX)) {
Utils.atomicMoveWithFallback(file.toPath, Paths.get(Utils.replaceSuffix(file.getPath, LogFileUtils.DELETED_FILE_SUFFIX, "")), false)
}
log = recoverAndCheck(config, allKeys)
// clean again
- cleaner.cleanSegments(log, log.logSegments.take(9).toSeq, offsetMap, 0L, new CleanerStats(),
+ cleaner.cleanSegments(log, log.logSegments.asScala.take(9).toSeq, offsetMap, 0L, new CleanerStats(),
new CleanedTransactionMetadata, -1)
// clear scheduler so that async deletes don't run
time.scheduler.clear()
@@ -1682,7 +1683,7 @@ class LogCleanerTest {
// 3) Simulate recovery just after swap file is created, before old segment files are
// renamed to .deleted. Clean operation is resumed during recovery.
- log.logSegments.head.changeFileSuffixes("", UnifiedLog.SwapFileSuffix)
+ log.logSegments.asScala.head.changeFileSuffixes("", UnifiedLog.SwapFileSuffix)
for (file <- dir.listFiles if file.getName.endsWith(LogFileUtils.DELETED_FILE_SUFFIX)) {
Utils.atomicMoveWithFallback(file.toPath, Paths.get(Utils.replaceSuffix(file.getPath, LogFileUtils.DELETED_FILE_SUFFIX, "")), false)
}
@@ -1695,7 +1696,7 @@ class LogCleanerTest {
}
for (k <- 1 until messageCount by 2)
offsetMap.put(key(k), Long.MaxValue)
- cleaner.cleanSegments(log, log.logSegments.take(9).toSeq, offsetMap, 0L, new CleanerStats(),
+ cleaner.cleanSegments(log, log.logSegments.asScala.take(9).toSeq, offsetMap, 0L, new CleanerStats(),
new CleanedTransactionMetadata, -1)
// clear scheduler so that async deletes don't run
time.scheduler.clear()
@@ -1703,7 +1704,7 @@ class LogCleanerTest {
// 4) Simulate recovery after swap file is created and old segments files are renamed
// to .deleted. Clean operation is resumed during recovery.
- log.logSegments.head.changeFileSuffixes("", UnifiedLog.SwapFileSuffix)
+ log.logSegments.asScala.head.changeFileSuffixes("", UnifiedLog.SwapFileSuffix)
log = recoverAndCheck(config, cleanedKeys)
// add some more messages and clean the log again
@@ -1713,7 +1714,7 @@ class LogCleanerTest {
}
for (k <- 1 until messageCount by 2)
offsetMap.put(key(k), Long.MaxValue)
- cleaner.cleanSegments(log, log.logSegments.take(9).toSeq, offsetMap, 0L, new CleanerStats(),
+ cleaner.cleanSegments(log, log.logSegments.asScala.take(9).toSeq, offsetMap, 0L, new CleanerStats(),
new CleanedTransactionMetadata, -1)
// clear scheduler so that async deletes don't run
time.scheduler.clear()
@@ -1721,7 +1722,7 @@ class LogCleanerTest {
// 5) Simulate recovery after a subset of swap files are renamed to regular files and old segments files are renamed
// to .deleted. Clean operation is resumed during recovery.
- log.logSegments.head.timeIndex.file.renameTo(new File(Utils.replaceSuffix(log.logSegments.head.timeIndex.file.getPath, "", UnifiedLog.SwapFileSuffix)))
+ log.logSegments.asScala.head.timeIndex.file.renameTo(new File(Utils.replaceSuffix(log.logSegments.asScala.head.timeIndex.file.getPath, "", UnifiedLog.SwapFileSuffix)))
log = recoverAndCheck(config, cleanedKeys)
// add some more messages and clean the log again
@@ -1731,7 +1732,7 @@ class LogCleanerTest {
}
for (k <- 1 until messageCount by 2)
offsetMap.put(key(k), Long.MaxValue)
- cleaner.cleanSegments(log, log.logSegments.take(9).toSeq, offsetMap, 0L, new CleanerStats(),
+ cleaner.cleanSegments(log, log.logSegments.asScala.take(9).toSeq, offsetMap, 0L, new CleanerStats(),
new CleanedTransactionMetadata, -1)
// clear scheduler so that async deletes don't run
time.scheduler.clear()
@@ -1828,7 +1829,7 @@ class LogCleanerTest {
cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0, log.activeSegment.baseOffset))
- for (segment <- log.logSegments; batch <- segment.log.batches.asScala; record <- batch.asScala) {
+ for (segment <- log.logSegments.asScala; batch <- segment.log.batches.asScala; record <- batch.asScala) {
assertTrue(record.hasMagic(batch.magic))
val value = TestUtils.readString(record.value).toLong
assertEquals(record.offset, value)
@@ -1875,7 +1876,7 @@ class LogCleanerTest {
log.roll()
cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 1, log.activeSegment.baseOffset))
- assertEquals(1, log.logSegments.head.log.batches.iterator.next().lastOffset,
+ assertEquals(1, log.logSegments.asScala.head.log.batches.iterator.next().lastOffset,
"The tombstone should be retained.")
// Append a message and roll out another log segment.
log.appendAsLeader(TestUtils.singletonRecords(value = "1".getBytes,
@@ -1883,7 +1884,7 @@ class LogCleanerTest {
timestamp = time.milliseconds()), leaderEpoch = 0)
log.roll()
cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 2, log.activeSegment.baseOffset))
- assertEquals(1, log.logSegments.head.log.batches.iterator.next().lastOffset,
+ assertEquals(1, log.logSegments.asScala.head.log.batches.iterator.next().lastOffset,
"The tombstone should be retained.")
}
diff --git a/core/src/test/scala/unit/kafka/log/LogConcurrencyTest.scala b/core/src/test/scala/unit/kafka/log/LogConcurrencyTest.scala
index 16700bcc7aa..7935cc761dc 100644
--- a/core/src/test/scala/unit/kafka/log/LogConcurrencyTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogConcurrencyTest.scala
@@ -161,7 +161,7 @@ class LogConcurrencyTest {
private def validateConsumedData(log: UnifiedLog, consumedBatches: Iterable[FetchedBatch]): Unit = {
val iter = consumedBatches.iterator
- log.logSegments.foreach { segment =>
+ log.logSegments.forEach { segment =>
segment.log.batches.forEach { batch =>
if (iter.hasNext) {
val consumedBatch = iter.next()
diff --git a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala b/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
index 577e4a6f735..b9b41259d4b 100644
--- a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
@@ -33,7 +33,7 @@ import org.apache.kafka.server.common.MetadataVersion
import org.apache.kafka.server.common.MetadataVersion.IBP_0_11_0_IV0
import org.apache.kafka.server.util.{MockTime, Scheduler}
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
-import org.apache.kafka.storage.internals.log.{AbortedTxn, CleanerConfig, EpochEntry, FetchDataInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetMetadata, LogStartOffsetIncrementReason, OffsetIndex, ProducerStateManager, ProducerStateManagerConfig, SnapshotFile}
+import org.apache.kafka.storage.internals.log.{AbortedTxn, CleanerConfig, EpochEntry, FetchDataInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetMetadata, LogSegment, LogSegments, LogStartOffsetIncrementReason, OffsetIndex, ProducerStateManager, ProducerStateManagerConfig, SnapshotFile}
import org.junit.jupiter.api.Assertions.{assertDoesNotThrow, assertEquals, assertFalse, assertNotEquals, assertThrows, assertTrue}
import org.junit.jupiter.api.function.Executable
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
@@ -154,7 +154,7 @@ class LogLoaderTest {
this.maxTransactionTimeoutMs, this.producerStateManagerConfig, time)
val logLoader = new LogLoader(logDir, topicPartition, config, time.scheduler, time,
logDirFailureChannel, hadCleanShutdown, segments, logStartOffset, logRecoveryPoint,
- leaderEpochCache, producerStateManager)
+ leaderEpochCache.asJava, producerStateManager)
val offsets = logLoader.load()
val localLog = new LocalLog(logDir, logConfig, segments, offsets.recoveryPoint,
offsets.nextOffsetMetadata, mockTime.scheduler, mockTime, topicPartition,
@@ -317,7 +317,7 @@ class LogLoaderTest {
}
assertTrue(log.logSegments.size >= 5)
- val segmentOffsets = log.logSegments.toVector.map(_.baseOffset)
+ val segmentOffsets = log.logSegments.asScala.toVector.map(_.baseOffset)
val activeSegmentOffset = segmentOffsets.last
// We want the recovery point to be past the segment offset and before the last 2 segments including a gap of
@@ -336,10 +336,10 @@ class LogLoaderTest {
if (logConfig.messageFormatVersion.isLessThan(IBP_0_11_0_IV0)) {
expectedSegmentsWithReads += activeSegmentOffset
- expectedSnapshotOffsets ++= log.logSegments.map(_.baseOffset).toVector.takeRight(2) :+ log.logEndOffset
+ expectedSnapshotOffsets ++= log.logSegments.asScala.map(_.baseOffset).toVector.takeRight(2) :+ log.logEndOffset
} else {
expectedSegmentsWithReads ++= segOffsetsBeforeRecovery ++ Set(activeSegmentOffset)
- expectedSnapshotOffsets ++= log.logSegments.map(_.baseOffset).toVector.takeRight(4) :+ log.logEndOffset
+ expectedSnapshotOffsets ++= log.logSegments.asScala.map(_.baseOffset).toVector.takeRight(4) :+ log.logEndOffset
}
def createLogWithInterceptedReads(recoveryPoint: Long): UnifiedLog = {
@@ -350,8 +350,7 @@ class LogLoaderTest {
// Intercept all segment read calls
val interceptedLogSegments = new LogSegments(topicPartition) {
override def add(segment: LogSegment): LogSegment = {
- val wrapper = new LogSegment(segment.log, segment.lazyOffsetIndex, segment.lazyTimeIndex, segment.txnIndex, segment.baseOffset,
- segment.indexIntervalBytes, segment.rollJitterMs, mockTime) {
+ val wrapper = new LogSegment(segment) {
override def read(startOffset: Long, maxSize: Int, maxPosition: Long, minOneMessage: Boolean): FetchDataInfo = {
segmentsWithReads += this
@@ -359,7 +358,7 @@ class LogLoaderTest {
}
override def recover(producerStateManager: ProducerStateManager,
- leaderEpochCache: Option[LeaderEpochFileCache]): Int = {
+ leaderEpochCache: Optional[LeaderEpochFileCache]): Int = {
recoveredSegments += this
super.recover(producerStateManager, leaderEpochCache)
}
@@ -381,7 +380,7 @@ class LogLoaderTest {
interceptedLogSegments,
0L,
recoveryPoint,
- leaderEpochCache,
+ leaderEpochCache.asJava,
producerStateManager)
val offsets = logLoader.load()
val localLog = new LocalLog(logDir, logConfig, interceptedLogSegments, offsets.recoveryPoint,
@@ -443,7 +442,7 @@ class LogLoaderTest {
segments,
0L,
0L,
- leaderEpochCache,
+ leaderEpochCache.asJava,
stateManager
).load()
val localLog = new LocalLog(logDir, config, segments, offsets.recoveryPoint,
@@ -552,7 +551,7 @@ class LogLoaderTest {
segments,
0L,
0L,
- leaderEpochCache,
+ leaderEpochCache.asJava,
stateManager
).load()
val localLog = new LocalLog(logDir, config, segments, offsets.recoveryPoint,
@@ -606,7 +605,7 @@ class LogLoaderTest {
segments,
0L,
0L,
- leaderEpochCache,
+ leaderEpochCache.asJava,
stateManager
).load()
val localLog = new LocalLog(logDir, config, segments, offsets.recoveryPoint,
@@ -659,7 +658,7 @@ class LogLoaderTest {
segments,
0L,
0L,
- leaderEpochCache,
+ leaderEpochCache.asJava,
stateManager
).load()
val localLog = new LocalLog(logDir, config, segments, offsets.recoveryPoint,
@@ -843,8 +842,8 @@ class LogLoaderTest {
var log = createLog(logDir, logConfig)
for(i <- 0 until numMessages)
log.appendAsLeader(TestUtils.singletonRecords(value = TestUtils.randomBytes(10), timestamp = mockTime.milliseconds + i * 10), leaderEpoch = 0)
- val indexFiles = log.logSegments.map(_.lazyOffsetIndex.file)
- val timeIndexFiles = log.logSegments.map(_.lazyTimeIndex.file)
+ val indexFiles = log.logSegments.asScala.map(_.offsetIndexFile)
+ val timeIndexFiles = log.logSegments.asScala.map(_.timeIndexFile)
log.close()
// delete all the index files
@@ -854,12 +853,12 @@ class LogLoaderTest {
// reopen the log
log = createLog(logDir, logConfig, lastShutdownClean = false)
assertEquals(numMessages, log.logEndOffset, "Should have %d messages when log is reopened".format(numMessages))
- assertTrue(log.logSegments.head.offsetIndex.entries > 0, "The index should have been rebuilt")
- assertTrue(log.logSegments.head.timeIndex.entries > 0, "The time index should have been rebuilt")
+ assertTrue(log.logSegments.asScala.head.offsetIndex.entries > 0, "The index should have been rebuilt")
+ assertTrue(log.logSegments.asScala.head.timeIndex.entries > 0, "The time index should have been rebuilt")
for(i <- 0 until numMessages) {
assertEquals(i, LogTestUtils.readLog(log, i, 100).records.batches.iterator.next().lastOffset)
if (i == 0)
- assertEquals(log.logSegments.head.baseOffset, log.fetchOffsetByTimestamp(mockTime.milliseconds + i * 10).get.offset)
+ assertEquals(log.logSegments.asScala.head.baseOffset, log.fetchOffsetByTimestamp(mockTime.milliseconds + i * 10).get.offset)
else
assertEquals(i, log.fetchOffsetByTimestamp(mockTime.milliseconds + i * 10).get.offset)
}
@@ -883,7 +882,7 @@ class LogLoaderTest {
for (i <- 0 until numMessages)
log.appendAsLeader(TestUtils.singletonRecords(value = TestUtils.randomBytes(10),
timestamp = mockTime.milliseconds + i * 10, magicValue = RecordBatch.MAGIC_VALUE_V1), leaderEpoch = 0)
- val timeIndexFiles = log.logSegments.map(_.lazyTimeIndex.file)
+ val timeIndexFiles = log.logSegments.asScala.map(_.timeIndexFile())
log.close()
// Delete the time index.
@@ -891,9 +890,9 @@ class LogLoaderTest {
// The rebuilt time index should be empty
log = createLog(logDir, logConfig, recoveryPoint = numMessages + 1, lastShutdownClean = false)
- for (segment <- log.logSegments.init) {
+ for (segment <- log.logSegments.asScala.init) {
assertEquals(0, segment.timeIndex.entries, "The time index should be empty")
- assertEquals(0, segment.lazyTimeIndex.file.length, "The time index file size should be 0")
+ assertEquals(0, segment.timeIndexFile().length, "The time index file size should be 0")
}
}
@@ -909,8 +908,8 @@ class LogLoaderTest {
var log = createLog(logDir, logConfig)
for(i <- 0 until numMessages)
log.appendAsLeader(TestUtils.singletonRecords(value = TestUtils.randomBytes(10), timestamp = mockTime.milliseconds + i * 10), leaderEpoch = 0)
- val indexFiles = log.logSegments.map(_.lazyOffsetIndex.file)
- val timeIndexFiles = log.logSegments.map(_.lazyTimeIndex.file)
+ val indexFiles = log.logSegments.asScala.map(_.offsetIndexFile())
+ val timeIndexFiles = log.logSegments.asScala.map(_.timeIndexFile())
log.close()
// corrupt all the index files
@@ -933,7 +932,7 @@ class LogLoaderTest {
for(i <- 0 until numMessages) {
assertEquals(i, LogTestUtils.readLog(log, i, 100).records.batches.iterator.next().lastOffset)
if (i == 0)
- assertEquals(log.logSegments.head.baseOffset, log.fetchOffsetByTimestamp(mockTime.milliseconds + i * 10).get.offset)
+ assertEquals(log.logSegments.asScala.head.baseOffset, log.fetchOffsetByTimestamp(mockTime.milliseconds + i * 10).get.offset)
else
assertEquals(i, log.fetchOffsetByTimestamp(mockTime.milliseconds + i * 10).get.offset)
}
@@ -945,10 +944,10 @@ class LogLoaderTest {
*/
@Test
def testBogusIndexSegmentsAreRemoved(): Unit = {
- val bogusIndex1 = UnifiedLog.offsetIndexFile(logDir, 0)
- val bogusTimeIndex1 = UnifiedLog.timeIndexFile(logDir, 0)
- val bogusIndex2 = UnifiedLog.offsetIndexFile(logDir, 5)
- val bogusTimeIndex2 = UnifiedLog.timeIndexFile(logDir, 5)
+ val bogusIndex1 = LogFileUtils.offsetIndexFile(logDir, 0)
+ val bogusTimeIndex1 = LogFileUtils.timeIndexFile(logDir, 0)
+ val bogusIndex2 = LogFileUtils.offsetIndexFile(logDir, 5)
+ val bogusTimeIndex2 = LogFileUtils.timeIndexFile(logDir, 5)
// The files remain absent until we first access it because we are doing lazy loading for time index and offset index
// files but in this test case we need to create these files in order to test we will remove them.
@@ -960,8 +959,8 @@ class LogLoaderTest {
val log = createLog(logDir, logConfig)
// Force the segment to access the index files because we are doing index lazy loading.
- log.logSegments.toSeq.head.offsetIndex
- log.logSegments.toSeq.head.timeIndex
+ log.logSegments.asScala.head.offsetIndex
+ log.logSegments.asScala.head.timeIndex
assertTrue(bogusIndex1.length > 0,
"The first index file should have been replaced with a larger file")
@@ -1033,18 +1032,18 @@ class LogLoaderTest {
val numMessages = 50 + TestUtils.random.nextInt(50)
for (_ <- 0 until numMessages)
log.appendAsLeader(createRecords, leaderEpoch = 0)
- val records = log.logSegments.flatMap(_.log.records.asScala.toList).toList
+ val records = log.logSegments.asScala.flatMap(_.log.records.asScala.toList).toList
log.close()
// corrupt index and log by appending random bytes
- TestUtils.appendNonsenseToFile(log.activeSegment.lazyOffsetIndex.file, TestUtils.random.nextInt(1024) + 1)
+ TestUtils.appendNonsenseToFile(log.activeSegment.offsetIndexFile, TestUtils.random.nextInt(1024) + 1)
TestUtils.appendNonsenseToFile(log.activeSegment.log.file, TestUtils.random.nextInt(1024) + 1)
// attempt recovery
log = createLog(logDir, logConfig, brokerTopicStats, 0L, recoveryPoint, lastShutdownClean = false)
assertEquals(numMessages, log.logEndOffset)
- val recovered = log.logSegments.flatMap(_.log.records.asScala.toList).toList
+ val recovered = log.logSegments.asScala.flatMap(_.log.records.asScala.toList).toList
assertEquals(records.size, recovered.size)
for (i <- records.indices) {
@@ -1209,7 +1208,7 @@ class LogLoaderTest {
// Running split again would throw an error
- for (segment <- recoveredLog.logSegments) {
+ for (segment <- recoveredLog.logSegments.asScala) {
assertThrows(classOf[IllegalArgumentException], () => log.splitOverflowedSegment(segment))
}
}
@@ -1438,7 +1437,7 @@ class LogLoaderTest {
LogTestUtils.appendEndTxnMarkerAsLeader(log, pid4, epoch, ControlRecordType.COMMIT, mockTime.milliseconds()) // 90
// delete all the offset and transaction index files to force recovery
- log.logSegments.foreach { segment =>
+ log.logSegments.forEach { segment =>
segment.offsetIndex.deleteIfExists()
segment.txnIndex.deleteIfExists()
}
@@ -1489,7 +1488,7 @@ class LogLoaderTest {
LogTestUtils.appendEndTxnMarkerAsLeader(log, pid4, epoch, ControlRecordType.COMMIT, mockTime.milliseconds()) // 90
// delete the last offset and transaction index files to force recovery
- val lastSegment = log.logSegments.last
+ val lastSegment = log.logSegments.asScala.last
val recoveryPoint = lastSegment.baseOffset
lastSegment.offsetIndex.deleteIfExists()
lastSegment.txnIndex.deleteIfExists()
@@ -1543,7 +1542,7 @@ class LogLoaderTest {
// delete the last offset and transaction index files to force recovery. this should force us to rebuild
// the producer state from the start of the log
- val lastSegment = log.logSegments.last
+ val lastSegment = log.logSegments.asScala.last
val recoveryPoint = lastSegment.baseOffset
lastSegment.offsetIndex.deleteIfExists()
lastSegment.txnIndex.deleteIfExists()
@@ -1574,7 +1573,7 @@ class LogLoaderTest {
assertTrue(log.logEndOffset > log.logStartOffset)
// Append garbage to a segment below the current log start offset
- val segmentToForceTruncation = log.logSegments.take(2).last
+ val segmentToForceTruncation = log.logSegments.asScala.take(2).last
val bw = new BufferedWriter(new FileWriter(segmentToForceTruncation.log.file))
bw.write("corruptRecord")
bw.close()
@@ -1590,11 +1589,11 @@ class LogLoaderTest {
assertEquals(startOffset, log.logStartOffset)
assertEquals(startOffset, log.logEndOffset)
// Validate that the remaining segment matches our expectations
- val onlySegment = log.logSegments.head
+ val onlySegment = log.logSegments.asScala.head
assertEquals(startOffset, onlySegment.baseOffset)
assertTrue(onlySegment.log.file().exists())
- assertTrue(onlySegment.lazyOffsetIndex.file.exists())
- assertTrue(onlySegment.lazyTimeIndex.file.exists())
+ assertTrue(onlySegment.offsetIndexFile().exists())
+ assertTrue(onlySegment.timeIndexFile().exists())
}
@Test
@@ -1639,7 +1638,7 @@ class LogLoaderTest {
// | |---> logEndOffset
// corrupt record inserted
//
- val segmentToForceTruncation = log.logSegments.take(2).last
+ val segmentToForceTruncation = log.logSegments.asScala.take(2).last
assertEquals(1, segmentToForceTruncation.baseOffset)
val bw = new BufferedWriter(new FileWriter(segmentToForceTruncation.log.file))
bw.write("corruptRecord")
@@ -1786,7 +1785,7 @@ class LogLoaderTest {
log.deleteOldSegments()
val segments = new LogSegments(topicPartition)
- log.logSegments.foreach(segment => segments.add(segment))
+ log.logSegments.forEach(segment => segments.add(segment))
assertEquals(5, segments.firstSegment.get.baseOffset)
val leaderEpochCache = UnifiedLog.maybeCreateLeaderEpochCache(logDir, topicPartition, logDirFailureChannel, logConfig.recordVersion, "")
@@ -1801,7 +1800,7 @@ class LogLoaderTest {
segments,
0L,
0L,
- leaderEpochCache,
+ leaderEpochCache.asJava,
stateManager,
isRemoteLogEnabled = isRemoteLogEnabled
).load()
diff --git a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
index 1d1ddf40686..7e35857c5f7 100755
--- a/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogManagerTest.scala
@@ -275,15 +275,15 @@ class LogManagerTest {
assertTrue(log.numberOfSegments > 1, "There should be more than one segment now.")
log.updateHighWatermark(log.logEndOffset)
- log.logSegments.foreach(_.log.file.setLastModified(time.milliseconds))
+ log.logSegments.forEach(_.log.file.setLastModified(time.milliseconds))
time.sleep(maxLogAgeMs + 1)
assertEquals(1, log.numberOfSegments, "Now there should only be only one segment in the index.")
time.sleep(log.config.fileDeleteDelayMs + 1)
- log.logSegments.foreach(s => {
- s.lazyOffsetIndex.get
- s.lazyTimeIndex.get
+ log.logSegments.forEach(s => {
+ s.offsetIndex()
+ s.timeIndex()
})
// there should be a log file, two indexes, one producer snapshot, and the leader epoch checkpoint
@@ -374,7 +374,7 @@ class LogManagerTest {
val numSegments = log.numberOfSegments
assertTrue(log.numberOfSegments > 1, "There should be more than one segment now.")
- log.logSegments.foreach(_.log.file.setLastModified(time.milliseconds))
+ log.logSegments.forEach(_.log.file.setLastModified(time.milliseconds))
time.sleep(maxLogAgeMs + 1)
assertEquals(numSegments, log.numberOfSegments, "number of segments shouldn't have changed")
@@ -511,12 +511,12 @@ class LogManagerTest {
val removedLog = logManager.asyncDelete(new TopicPartition(name, 0)).get
val removedSegment = removedLog.activeSegment
- val indexFilesAfterDelete = Seq(removedSegment.lazyOffsetIndex.file, removedSegment.lazyTimeIndex.file,
+ val indexFilesAfterDelete = Seq(removedSegment.offsetIndexFile, removedSegment.timeIndexFile,
removedSegment.txnIndex.file)
assertEquals(new File(removedLog.dir, logName), removedSegment.log.file)
- assertEquals(new File(removedLog.dir, indexName), removedSegment.lazyOffsetIndex.file)
- assertEquals(new File(removedLog.dir, timeIndexName), removedSegment.lazyTimeIndex.file)
+ assertEquals(new File(removedLog.dir, indexName), removedSegment.offsetIndexFile)
+ assertEquals(new File(removedLog.dir, timeIndexName), removedSegment.timeIndexFile)
assertEquals(new File(removedLog.dir, txnIndexName), removedSegment.txnIndex.file)
// Try to detect the case where a new index type was added and we forgot to update the pointer
diff --git a/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala b/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala
index 65088b7dae3..aacab5b624e 100644
--- a/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala
@@ -16,7 +16,6 @@
*/
package kafka.log
-import kafka.common.LogSegmentOffsetOverflowException
import kafka.utils.TestUtils
import kafka.utils.TestUtils.checkEquals
import org.apache.kafka.common.TopicPartition
@@ -25,7 +24,7 @@ import org.apache.kafka.common.record._
import org.apache.kafka.common.utils.{MockTime, Time, Utils}
import org.apache.kafka.storage.internals.checkpoint.LeaderEpochCheckpoint
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
-import org.apache.kafka.storage.internals.log.{BatchMetadata, EpochEntry, LogConfig, ProducerStateEntry, ProducerStateManager, ProducerStateManagerConfig, RollParams}
+import org.apache.kafka.storage.internals.log.{BatchMetadata, EpochEntry, LogConfig, LogFileUtils, LogSegment, LogSegmentOffsetOverflowException, ProducerStateEntry, ProducerStateManager, ProducerStateManagerConfig, RollParams}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import org.junit.jupiter.params.ParameterizedTest
@@ -33,7 +32,7 @@ import org.junit.jupiter.params.provider.CsvSource
import java.io.File
import java.util
-import java.util.OptionalLong
+import java.util.{Optional, OptionalLong}
import scala.collection._
import scala.jdk.CollectionConverters._
@@ -99,7 +98,7 @@ class LogSegmentTest {
@Test
def testReadOnEmptySegment(): Unit = {
val seg = createSegment(40)
- val read = seg.read(startOffset = 40, maxSize = 300)
+ val read = seg.read(40, 300)
assertNull(read, "Read beyond the last offset in the segment should be null")
}
@@ -112,7 +111,7 @@ class LogSegmentTest {
val seg = createSegment(40)
val ms = records(50, "hello", "there", "little", "bee")
seg.append(53, RecordBatch.NO_TIMESTAMP, -1L, ms)
- val read = seg.read(startOffset = 41, maxSize = 300).records
+ val read = seg.read(41, 300).records
checkEquals(ms.records.iterator, read.records.iterator)
}
@@ -124,7 +123,7 @@ class LogSegmentTest {
val seg = createSegment(40)
val ms = records(50, "hello", "there")
seg.append(51, RecordBatch.NO_TIMESTAMP, -1L, ms)
- val read = seg.read(startOffset = 52, maxSize = 200)
+ val read = seg.read(52, 200)
assertNull(read, "Read beyond the last offset in the segment should give null")
}
@@ -139,7 +138,7 @@ class LogSegmentTest {
seg.append(51, RecordBatch.NO_TIMESTAMP, -1L, ms)
val ms2 = records(60, "alpha", "beta")
seg.append(61, RecordBatch.NO_TIMESTAMP, -1L, ms2)
- val read = seg.read(startOffset = 55, maxSize = 200)
+ val read = seg.read(55, 200)
checkEquals(ms2.records.iterator, read.records.records.iterator)
}
@@ -264,17 +263,17 @@ class LogSegmentTest {
assertEquals(490, seg.largestTimestamp)
// Search for an indexed timestamp
- assertEquals(42, seg.findOffsetByTimestamp(420).get.offset)
- assertEquals(43, seg.findOffsetByTimestamp(421).get.offset)
+ assertEquals(42, seg.findOffsetByTimestamp(420, 0L).get.offset)
+ assertEquals(43, seg.findOffsetByTimestamp(421, 0L).get.offset)
// Search for an un-indexed timestamp
- assertEquals(43, seg.findOffsetByTimestamp(430).get.offset)
- assertEquals(44, seg.findOffsetByTimestamp(431).get.offset)
+ assertEquals(43, seg.findOffsetByTimestamp(430, 0L).get.offset)
+ assertEquals(44, seg.findOffsetByTimestamp(431, 0L).get.offset)
// Search beyond the last timestamp
- assertEquals(None, seg.findOffsetByTimestamp(491))
+ assertEquals(Optional.empty(), seg.findOffsetByTimestamp(491, 0L))
// Search before the first indexed timestamp
- assertEquals(41, seg.findOffsetByTimestamp(401).get.offset)
+ assertEquals(41, seg.findOffsetByTimestamp(401, 0L).get.offset)
// Search before the first timestamp
- assertEquals(40, seg.findOffsetByTimestamp(399).get.offset)
+ assertEquals(40, seg.findOffsetByTimestamp(399, 0L).get.offset)
}
/**
@@ -295,26 +294,26 @@ class LogSegmentTest {
def testChangeFileSuffixes(): Unit = {
val seg = createSegment(40)
val logFile = seg.log.file
- val indexFile = seg.lazyOffsetIndex.file
- val timeIndexFile = seg.lazyTimeIndex.file
+ val indexFile = seg.offsetIndexFile
+ val timeIndexFile = seg.timeIndexFile
// Ensure that files for offset and time indices have not been created eagerly.
- assertFalse(seg.lazyOffsetIndex.file.exists)
- assertFalse(seg.lazyTimeIndex.file.exists)
+ assertFalse(seg.offsetIndexFile.exists)
+ assertFalse(seg.timeIndexFile.exists)
seg.changeFileSuffixes("", ".deleted")
// Ensure that attempt to change suffixes for non-existing offset and time indices does not create new files.
- assertFalse(seg.lazyOffsetIndex.file.exists)
- assertFalse(seg.lazyTimeIndex.file.exists)
+ assertFalse(seg.offsetIndexFile.exists)
+ assertFalse(seg.timeIndexFile.exists)
// Ensure that file names are updated accordingly.
assertEquals(logFile.getAbsolutePath + ".deleted", seg.log.file.getAbsolutePath)
- assertEquals(indexFile.getAbsolutePath + ".deleted", seg.lazyOffsetIndex.file.getAbsolutePath)
- assertEquals(timeIndexFile.getAbsolutePath + ".deleted", seg.lazyTimeIndex.file.getAbsolutePath)
+ assertEquals(indexFile.getAbsolutePath + ".deleted", seg.offsetIndexFile.getAbsolutePath)
+ assertEquals(timeIndexFile.getAbsolutePath + ".deleted", seg.timeIndexFile.getAbsolutePath)
assertTrue(seg.log.file.exists)
// Ensure lazy creation of offset index file upon accessing it.
- seg.lazyOffsetIndex.get
- assertTrue(seg.lazyOffsetIndex.file.exists)
+ seg.offsetIndex()
+ assertTrue(seg.offsetIndexFile.exists)
// Ensure lazy creation of time index file upon accessing it.
- seg.lazyTimeIndex.get
- assertTrue(seg.lazyTimeIndex.file.exists)
+ seg.timeIndex()
+ assertTrue(seg.timeIndexFile.exists)
}
/**
@@ -326,11 +325,11 @@ class LogSegmentTest {
val seg = createSegment(0)
for(i <- 0 until 100)
seg.append(i, RecordBatch.NO_TIMESTAMP, -1L, records(i, i.toString))
- val indexFile = seg.lazyOffsetIndex.file
+ val indexFile = seg.offsetIndexFile
TestUtils.writeNonsenseToFile(indexFile, 5, indexFile.length.toInt)
- seg.recover(newProducerStateManager())
+ seg.recover(newProducerStateManager(), Optional.empty())
for(i <- 0 until 100) {
- val records = seg.read(i, 1, minOneMessage = true).records.records
+ val records = seg.read(i, 1, seg.size(), true).records.records
assertEquals(i, records.iterator.next().offset)
}
}
@@ -346,30 +345,28 @@ class LogSegmentTest {
val pid2 = 10L
// append transactional records from pid1
- segment.append(largestOffset = 101L, largestTimestamp = RecordBatch.NO_TIMESTAMP,
- shallowOffsetOfMaxTimestamp = 100L, records = MemoryRecords.withTransactionalRecords(100L, CompressionType.NONE,
+ segment.append(101L, RecordBatch.NO_TIMESTAMP,
+ 100L, MemoryRecords.withTransactionalRecords(100L, CompressionType.NONE,
pid1, producerEpoch, sequence, partitionLeaderEpoch, new SimpleRecord("a".getBytes), new SimpleRecord("b".getBytes)))
// append transactional records from pid2
- segment.append(largestOffset = 103L, largestTimestamp = RecordBatch.NO_TIMESTAMP,
- shallowOffsetOfMaxTimestamp = 102L, records = MemoryRecords.withTransactionalRecords(102L, CompressionType.NONE,
+ segment.append(103L, RecordBatch.NO_TIMESTAMP, 102L, MemoryRecords.withTransactionalRecords(102L, CompressionType.NONE,
pid2, producerEpoch, sequence, partitionLeaderEpoch, new SimpleRecord("a".getBytes), new SimpleRecord("b".getBytes)))
// append non-transactional records
- segment.append(largestOffset = 105L, largestTimestamp = RecordBatch.NO_TIMESTAMP,
- shallowOffsetOfMaxTimestamp = 104L, records = MemoryRecords.withRecords(104L, CompressionType.NONE,
+ segment.append(105L, RecordBatch.NO_TIMESTAMP, 104L, MemoryRecords.withRecords(104L, CompressionType.NONE,
partitionLeaderEpoch, new SimpleRecord("a".getBytes), new SimpleRecord("b".getBytes)))
// abort the transaction from pid2 (note LSO should be 100L since the txn from pid1 has not completed)
- segment.append(largestOffset = 106L, largestTimestamp = RecordBatch.NO_TIMESTAMP,
- shallowOffsetOfMaxTimestamp = 106L, records = endTxnRecords(ControlRecordType.ABORT, pid2, producerEpoch, offset = 106L))
+ segment.append(106L, RecordBatch.NO_TIMESTAMP, 106L,
+ endTxnRecords(ControlRecordType.ABORT, pid2, producerEpoch, offset = 106L))
// commit the transaction from pid1
- segment.append(largestOffset = 107L, largestTimestamp = RecordBatch.NO_TIMESTAMP,
- shallowOffsetOfMaxTimestamp = 107L, records = endTxnRecords(ControlRecordType.COMMIT, pid1, producerEpoch, offset = 107L))
+ segment.append(107L, RecordBatch.NO_TIMESTAMP, 107L,
+ endTxnRecords(ControlRecordType.COMMIT, pid1, producerEpoch, offset = 107L))
var stateManager = newProducerStateManager()
- segment.recover(stateManager)
+ segment.recover(stateManager, Optional.empty())
assertEquals(108L, stateManager.mapEndOffset)
@@ -384,7 +381,7 @@ class LogSegmentTest {
// recover again, but this time assuming the transaction from pid2 began on a previous segment
stateManager = newProducerStateManager()
stateManager.loadProducerEntry(new ProducerStateEntry(pid2, producerEpoch, 0, RecordBatch.NO_TIMESTAMP, OptionalLong.of(75L), java.util.Optional.of(new BatchMetadata(10, 10L, 5, RecordBatch.NO_TIMESTAMP))))
- segment.recover(stateManager)
+ segment.recover(stateManager, Optional.empty())
assertEquals(108L, stateManager.mapEndOffset)
abortedTxns = segment.txnIndex.allAbortedTxns
@@ -415,23 +412,19 @@ class LogSegmentTest {
}
val cache = new LeaderEpochFileCache(topicPartition, checkpoint)
- seg.append(largestOffset = 105L, largestTimestamp = RecordBatch.NO_TIMESTAMP,
- shallowOffsetOfMaxTimestamp = 104L, records = MemoryRecords.withRecords(104L, CompressionType.NONE, 0,
+ seg.append(105L, RecordBatch.NO_TIMESTAMP, 104L, MemoryRecords.withRecords(104L, CompressionType.NONE, 0,
new SimpleRecord("a".getBytes), new SimpleRecord("b".getBytes)))
- seg.append(largestOffset = 107L, largestTimestamp = RecordBatch.NO_TIMESTAMP,
- shallowOffsetOfMaxTimestamp = 106L, records = MemoryRecords.withRecords(106L, CompressionType.NONE, 1,
+ seg.append(107L, RecordBatch.NO_TIMESTAMP, 106L, MemoryRecords.withRecords(106L, CompressionType.NONE, 1,
new SimpleRecord("a".getBytes), new SimpleRecord("b".getBytes)))
- seg.append(largestOffset = 109L, largestTimestamp = RecordBatch.NO_TIMESTAMP,
- shallowOffsetOfMaxTimestamp = 108L, records = MemoryRecords.withRecords(108L, CompressionType.NONE, 1,
+ seg.append(109L, RecordBatch.NO_TIMESTAMP, 108L, MemoryRecords.withRecords(108L, CompressionType.NONE, 1,
new SimpleRecord("a".getBytes), new SimpleRecord("b".getBytes)))
- seg.append(largestOffset = 111L, largestTimestamp = RecordBatch.NO_TIMESTAMP,
- shallowOffsetOfMaxTimestamp = 110, records = MemoryRecords.withRecords(110L, CompressionType.NONE, 2,
+ seg.append(111L, RecordBatch.NO_TIMESTAMP, 110, MemoryRecords.withRecords(110L, CompressionType.NONE, 2,
new SimpleRecord("a".getBytes), new SimpleRecord("b".getBytes)))
- seg.recover(newProducerStateManager(), Some(cache))
+ seg.recover(newProducerStateManager(), Optional.of(cache))
assertEquals(java.util.Arrays.asList(new EpochEntry(0, 104L),
new EpochEntry(1, 106),
new EpochEntry(2, 110)),
@@ -458,13 +451,13 @@ class LogSegmentTest {
val seg = createSegment(0)
for(i <- 0 until 100)
seg.append(i, i * 10, i, records(i, i.toString))
- val timeIndexFile = seg.lazyTimeIndex.file
+ val timeIndexFile = seg.timeIndexFile
TestUtils.writeNonsenseToFile(timeIndexFile, 5, timeIndexFile.length.toInt)
- seg.recover(newProducerStateManager())
+ seg.recover(newProducerStateManager(), Optional.empty())
for(i <- 0 until 100) {
- assertEquals(i, seg.findOffsetByTimestamp(i * 10).get.offset)
+ assertEquals(i, seg.findOffsetByTimestamp(i * 10, 0L).get.offset)
if (i < 99)
- assertEquals(i + 1, seg.findOffsetByTimestamp(i * 10 + 1).get.offset)
+ assertEquals(i + 1, seg.findOffsetByTimestamp(i * 10 + 1, 0L).get.offset)
}
}
@@ -484,7 +477,7 @@ class LogSegmentTest {
val recordPosition = seg.log.searchForOffsetWithSize(offsetToBeginCorruption, 0)
val position = recordPosition.position + TestUtils.random.nextInt(15)
TestUtils.writeNonsenseToFile(seg.log.file, position, (seg.log.file.length - position).toInt)
- seg.recover(newProducerStateManager())
+ seg.recover(newProducerStateManager(), Optional.empty())
assertEquals((0 until offsetToBeginCorruption).toList, seg.log.batches.asScala.map(_.lastOffset).toList,
"Should have truncated off bad messages.")
seg.deleteIfExists()
@@ -498,8 +491,7 @@ class LogSegmentTest {
TopicConfig.SEGMENT_INDEX_BYTES_CONFIG -> 1000,
TopicConfig.SEGMENT_JITTER_MS_CONFIG -> 0
).asJava)
- val seg = LogSegment.open(tempDir, baseOffset, logConfig, Time.SYSTEM, fileAlreadyExists = fileAlreadyExists,
- initFileSize = initFileSize, preallocate = preallocate)
+ val seg = LogSegment.open(tempDir, baseOffset, logConfig, Time.SYSTEM, fileAlreadyExists, initFileSize, preallocate, "")
segments += seg
seg
}
@@ -512,7 +504,7 @@ class LogSegmentTest {
seg.append(51, RecordBatch.NO_TIMESTAMP, -1L, ms)
val ms2 = records(60, "alpha", "beta")
seg.append(61, RecordBatch.NO_TIMESTAMP, -1L, ms2)
- val read = seg.read(startOffset = 55, maxSize = 200)
+ val read = seg.read(55, 200)
checkEquals(ms2.records.iterator, read.records.records.iterator)
}
@@ -526,14 +518,14 @@ class LogSegmentTest {
TopicConfig.SEGMENT_JITTER_MS_CONFIG -> 0
).asJava)
- val seg = LogSegment.open(tempDir, baseOffset = 40, logConfig, Time.SYSTEM,
- initFileSize = 512 * 1024 * 1024, preallocate = true)
+ val seg = LogSegment.open(tempDir, 40, logConfig, Time.SYSTEM,
+ 512 * 1024 * 1024, true)
val ms = records(50, "hello", "there")
seg.append(51, RecordBatch.NO_TIMESTAMP, -1L, ms)
val ms2 = records(60, "alpha", "beta")
seg.append(61, RecordBatch.NO_TIMESTAMP, -1L, ms2)
- val read = seg.read(startOffset = 55, maxSize = 200)
+ val read = seg.read(55, 200)
checkEquals(ms2.records.iterator, read.records.records.iterator)
val oldSize = seg.log.sizeInBytes()
val oldPosition = seg.log.channel.position
@@ -543,11 +535,10 @@ class LogSegmentTest {
//After close, file should be trimmed
assertEquals(oldSize, seg.log.file.length)
- val segReopen = LogSegment.open(tempDir, baseOffset = 40, logConfig, Time.SYSTEM, fileAlreadyExists = true,
- initFileSize = 512 * 1024 * 1024, preallocate = true)
+ val segReopen = LogSegment.open(tempDir, 40, logConfig, Time.SYSTEM, true, 512 * 1024 * 1024, true, "")
segments += segReopen
- val readAgain = segReopen.read(startOffset = 55, maxSize = 200)
+ val readAgain = segReopen.read(55, 200)
checkEquals(ms2.records.iterator, readAgain.records.records.iterator)
val size = segReopen.log.sizeInBytes()
val position = segReopen.log.channel.position
@@ -589,7 +580,7 @@ class LogSegmentTest {
// create a log file in a separate directory to avoid conflicting with created segments
val tempDir = TestUtils.tempDir()
- val fileRecords = FileRecords.open(UnifiedLog.logFile(tempDir, 0))
+ val fileRecords = FileRecords.open(LogFileUtils.logFile(tempDir, 0))
// Simulate a scenario where we have a single log with an offset range exceeding Int.MaxValue
fileRecords.append(records(0, 1024))
diff --git a/core/src/test/scala/unit/kafka/log/LogSegmentsTest.scala b/core/src/test/scala/unit/kafka/log/LogSegmentsTest.scala
index dc6aaa1dfc4..2cf352cea46 100644
--- a/core/src/test/scala/unit/kafka/log/LogSegmentsTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogSegmentsTest.scala
@@ -17,14 +17,18 @@
package kafka.log
import java.io.File
-
import kafka.utils.TestUtils
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.utils.{Time, Utils}
+import org.apache.kafka.storage.internals.log.{LogSegment, LogSegments}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import org.mockito.Mockito.{mock, when}
+import java.util.Arrays.asList
+import java.util.Optional
+import scala.jdk.CollectionConverters._
+
class LogSegmentsTest {
val topicPartition = new TopicPartition("topic", 0)
@@ -47,7 +51,7 @@ class LogSegmentsTest {
Utils.delete(logDir)
}
- private def assertEntry(segment: LogSegment, tested: java.util.Map.Entry[Long, LogSegment]): Unit = {
+ private def assertEntry(segment: LogSegment, tested: java.util.Map.Entry[java.lang.Long, LogSegment]): Unit = {
assertEquals(segment.baseOffset, tested.getKey())
assertEquals(segment, tested.getValue())
}
@@ -70,7 +74,7 @@ class LogSegmentsTest {
assertTrue(segments.nonEmpty)
assertEquals(1, segments.numberOfSegments)
assertTrue(segments.contains(offset1))
- assertEquals(Some(seg1), segments.get(offset1))
+ assertEquals(Optional.of(seg1), segments.get(offset1))
// Add seg2
segments.add(seg2)
@@ -78,7 +82,7 @@ class LogSegmentsTest {
assertTrue(segments.nonEmpty)
assertEquals(2, segments.numberOfSegments)
assertTrue(segments.contains(offset2))
- assertEquals(Some(seg2), segments.get(offset2))
+ assertEquals(Optional.of(seg2), segments.get(offset2))
// Replace seg1 with seg3
segments.add(seg3)
@@ -86,7 +90,7 @@ class LogSegmentsTest {
assertTrue(segments.nonEmpty)
assertEquals(2, segments.numberOfSegments)
assertTrue(segments.contains(offset1))
- assertEquals(Some(seg3), segments.get(offset1))
+ assertEquals(Optional.of(seg3), segments.get(offset1))
// Remove seg2
segments.remove(offset2)
@@ -119,31 +123,30 @@ class LogSegmentsTest {
val seg4 = createSegment(offset4)
// Test firstEntry, lastEntry
- List(seg1, seg2, seg3, seg4).foreach {
- seg =>
- segments.add(seg)
- assertEntry(seg1, segments.firstEntry.get)
- assertEquals(Some(seg1), segments.firstSegment)
- assertEntry(seg, segments.lastEntry.get)
- assertEquals(Some(seg), segments.lastSegment)
+ List(seg1, seg2, seg3, seg4).foreach { seg =>
+ segments.add(seg)
+ assertEntry(seg1, segments.firstEntry.get)
+ assertEquals(Optional.of(seg1), segments.firstSegment)
+ assertEntry(seg, segments.lastEntry.get)
+ assertEquals(Optional.of(seg), segments.lastSegment)
}
// Test baseOffsets
- assertEquals(Seq(offset1, offset2, offset3, offset4), segments.baseOffsets)
+ assertEquals(Seq(offset1, offset2, offset3, offset4), segments.baseOffsets.asScala.toSeq)
// Test values
- assertEquals(Seq(seg1, seg2, seg3, seg4), segments.values.toSeq)
+ assertEquals(Seq(seg1, seg2, seg3, seg4), segments.values.asScala.toSeq)
// Test values(to, from)
assertThrows(classOf[IllegalArgumentException], () => segments.values(2, 1))
- assertEquals(Seq(), segments.values(1, 1).toSeq)
- assertEquals(Seq(seg1), segments.values(1, 2).toSeq)
- assertEquals(Seq(seg1, seg2), segments.values(1, 3).toSeq)
- assertEquals(Seq(seg1, seg2, seg3), segments.values(1, 4).toSeq)
- assertEquals(Seq(seg2, seg3), segments.values(2, 4).toSeq)
- assertEquals(Seq(seg3), segments.values(3, 4).toSeq)
- assertEquals(Seq(), segments.values(4, 4).toSeq)
- assertEquals(Seq(seg4), segments.values(4, 5).toSeq)
+ assertEquals(Seq(), segments.values(1, 1).asScala.toSeq)
+ assertEquals(Seq(seg1), segments.values(1, 2).asScala.toSeq)
+ assertEquals(Seq(seg1, seg2), segments.values(1, 3).asScala.toSeq)
+ assertEquals(Seq(seg1, seg2, seg3), segments.values(1, 4).asScala.toSeq)
+ assertEquals(Seq(seg2, seg3), segments.values(2, 4).asScala.toSeq)
+ assertEquals(Seq(seg3), segments.values(3, 4).asScala.toSeq)
+ assertEquals(Seq(), segments.values(4, 4).asScala.toSeq)
+ assertEquals(Seq(seg4), segments.values(4, 5).asScala.toSeq)
segments.close()
}
@@ -160,17 +163,17 @@ class LogSegmentsTest {
List(seg1, seg2, seg3, seg4).foreach(segments.add)
// Test floorSegment
- assertEquals(Some(seg1), segments.floorSegment(2))
- assertEquals(Some(seg2), segments.floorSegment(3))
+ assertEquals(Optional.of(seg1), segments.floorSegment(2))
+ assertEquals(Optional.of(seg2), segments.floorSegment(3))
// Test lowerSegment
- assertEquals(Some(seg1), segments.lowerSegment(3))
- assertEquals(Some(seg2), segments.lowerSegment(4))
+ assertEquals(Optional.of(seg1), segments.lowerSegment(3))
+ assertEquals(Optional.of(seg2), segments.lowerSegment(4))
// Test higherSegment, higherEntry
- assertEquals(Some(seg3), segments.higherSegment(4))
+ assertEquals(Optional.of(seg3), segments.higherSegment(4))
assertEntry(seg3, segments.higherEntry(4).get)
- assertEquals(Some(seg4), segments.higherSegment(5))
+ assertEquals(Optional.of(seg4), segments.higherSegment(5))
assertEntry(seg4, segments.higherEntry(5).get)
segments.close()
@@ -232,8 +235,8 @@ class LogSegmentsTest {
when(logSegment.size).thenReturn(Int.MaxValue)
- assertEquals(Int.MaxValue, LogSegments.sizeInBytes(Seq(logSegment)))
- assertEquals(largeSize, LogSegments.sizeInBytes(Seq(logSegment, logSegment)))
- assertTrue(UnifiedLog.sizeInBytes(Seq(logSegment, logSegment)) > Int.MaxValue)
+ assertEquals(Int.MaxValue, LogSegments.sizeInBytes(asList(logSegment)))
+ assertEquals(largeSize, LogSegments.sizeInBytes(asList(logSegment, logSegment)))
+ assertTrue(UnifiedLog.sizeInBytes(asList(logSegment, logSegment)) > Int.MaxValue)
}
}
diff --git a/core/src/test/scala/unit/kafka/log/LogTestUtils.scala b/core/src/test/scala/unit/kafka/log/LogTestUtils.scala
index d9ab92e8a4d..ad47da05a00 100644
--- a/core/src/test/scala/unit/kafka/log/LogTestUtils.scala
+++ b/core/src/test/scala/unit/kafka/log/LogTestUtils.scala
@@ -33,7 +33,7 @@ import java.util.concurrent.{ConcurrentHashMap, ConcurrentMap}
import org.apache.kafka.common.config.TopicConfig
import org.apache.kafka.server.util.Scheduler
import org.apache.kafka.storage.internals.checkpoint.LeaderEpochCheckpointFile
-import org.apache.kafka.storage.internals.log.{AbortedTxn, AppendOrigin, FetchDataInfo, FetchIsolation, LazyIndex, LogAppendInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetsListener, ProducerStateManager, ProducerStateManagerConfig, TransactionIndex}
+import org.apache.kafka.storage.internals.log.{AbortedTxn, AppendOrigin, FetchDataInfo, FetchIsolation, LazyIndex, LogAppendInfo, LogConfig, LogDirFailureChannel, LogFileUtils, LogOffsetsListener, LogSegment, ProducerStateManager, ProducerStateManagerConfig, TransactionIndex}
import scala.jdk.CollectionConverters._
@@ -45,9 +45,9 @@ object LogTestUtils {
logDir: File,
indexIntervalBytes: Int = 10,
time: Time = Time.SYSTEM): LogSegment = {
- val ms = FileRecords.open(UnifiedLog.logFile(logDir, offset))
- val idx = LazyIndex.forOffset(UnifiedLog.offsetIndexFile(logDir, offset), offset, 1000)
- val timeIdx = LazyIndex.forTime(UnifiedLog.timeIndexFile(logDir, offset), offset, 1500)
+ val ms = FileRecords.open(LogFileUtils.logFile(logDir, offset))
+ val idx = LazyIndex.forOffset(LogFileUtils.offsetIndexFile(logDir, offset), offset, 1000)
+ val timeIdx = LazyIndex.forTime(LogFileUtils.timeIndexFile(logDir, offset), offset, 1500)
val txnIndex = new TransactionIndex(offset, UnifiedLog.transactionIndexFile(logDir, offset))
new LogSegment(ms, idx, timeIdx, txnIndex, offset, indexIntervalBytes, 0, time)
@@ -128,7 +128,7 @@ object LogTestUtils {
def hasOverflow(baseOffset: Long, batch: RecordBatch): Boolean =
batch.lastOffset > baseOffset + Int.MaxValue || batch.baseOffset < baseOffset
- for (segment <- log.logSegments) {
+ for (segment <- log.logSegments.asScala) {
val overflowBatch = segment.log.batches.asScala.find(batch => hasOverflow(segment.baseOffset, batch))
if (overflowBatch.isDefined)
return Some(segment)
@@ -137,7 +137,7 @@ object LogTestUtils {
}
def rawSegment(logDir: File, baseOffset: Long): FileRecords =
- FileRecords.open(UnifiedLog.logFile(logDir, baseOffset))
+ FileRecords.open(LogFileUtils.logFile(logDir, baseOffset))
/**
* Initialize the given log directory with a set of segments, one of which will have an
@@ -158,8 +158,8 @@ object LogTestUtils {
segment.append(MemoryRecords.withRecords(baseOffset + Int.MaxValue - 1, CompressionType.NONE, 0,
record(baseOffset + Int.MaxValue - 1)))
// Need to create the offset files explicitly to avoid triggering segment recovery to truncate segment.
- Files.createFile(UnifiedLog.offsetIndexFile(logDir, baseOffset).toPath)
- Files.createFile(UnifiedLog.timeIndexFile(logDir, baseOffset).toPath)
+ Files.createFile(LogFileUtils.offsetIndexFile(logDir, baseOffset).toPath)
+ Files.createFile(LogFileUtils.timeIndexFile(logDir, baseOffset).toPath)
baseOffset + Int.MaxValue
}
@@ -186,7 +186,7 @@ object LogTestUtils {
/* extract all the keys from a log */
def keysInLog(log: UnifiedLog): Iterable[Long] = {
- for (logSegment <- log.logSegments;
+ for (logSegment <- log.logSegments.asScala;
batch <- logSegment.log.batches.asScala if !batch.isControlBatch;
record <- batch.asScala if record.hasValue && record.hasKey)
yield TestUtils.readString(record.key).toLong
@@ -238,7 +238,8 @@ object LogTestUtils {
log.read(startOffset, maxLength, isolation, minOneMessage)
}
- def allAbortedTransactions(log: UnifiedLog): Iterable[AbortedTxn] = log.logSegments.flatMap(_.txnIndex.allAbortedTxns.asScala)
+ def allAbortedTransactions(log: UnifiedLog): Iterable[AbortedTxn] =
+ log.logSegments.asScala.flatMap(_.txnIndex.allAbortedTxns.asScala)
def deleteProducerSnapshotFiles(logDir: File): Unit = {
val files = logDir.listFiles.filter(f => f.isFile && f.getName.endsWith(LogFileUtils.PRODUCER_SNAPSHOT_FILE_SUFFIX))
diff --git a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
index 16d8c1e7adb..8c96312f6ca 100755
--- a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
+++ b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
@@ -36,7 +36,7 @@ import org.apache.kafka.server.metrics.KafkaYammerMetrics
import org.apache.kafka.server.util.{KafkaScheduler, MockTime, Scheduler}
import org.apache.kafka.storage.internals.checkpoint.LeaderEpochCheckpointFile
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
-import org.apache.kafka.storage.internals.log.{AbortedTxn, AppendOrigin, EpochEntry, FetchIsolation, LogConfig, LogFileUtils, LogOffsetMetadata, LogOffsetSnapshot, LogOffsetsListener, LogStartOffsetIncrementReason, ProducerStateManager, ProducerStateManagerConfig, RecordValidationException}
+import org.apache.kafka.storage.internals.log.{AbortedTxn, AppendOrigin, EpochEntry, FetchIsolation, LogConfig, LogFileUtils, LogOffsetMetadata, LogOffsetSnapshot, LogOffsetsListener, LogSegment, LogStartOffsetIncrementReason, ProducerStateManager, ProducerStateManagerConfig, RecordValidationException}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import org.junit.jupiter.params.ParameterizedTest
@@ -80,8 +80,8 @@ class UnifiedLogTest {
def createEmptyLogs(dir: File, offsets: Int*): Unit = {
for(offset <- offsets) {
- Files.createFile(UnifiedLog.logFile(dir, offset).toPath)
- Files.createFile(UnifiedLog.offsetIndexFile(dir, offset).toPath)
+ Files.createFile(LogFileUtils.logFile(dir, offset).toPath)
+ Files.createFile(LogFileUtils.offsetIndexFile(dir, offset).toPath)
}
}
@@ -719,7 +719,7 @@ class UnifiedLogTest {
// Reload after clean shutdown
log = createLog(logDir, logConfig, recoveryPoint = logEndOffset)
- var expectedSnapshotOffsets = log.logSegments.map(_.baseOffset).takeRight(2).toVector :+ log.logEndOffset
+ var expectedSnapshotOffsets = log.logSegments.asScala.map(_.baseOffset).takeRight(2).toVector :+ log.logEndOffset
assertEquals(expectedSnapshotOffsets, LogTestUtils.listProducerSnapshotOffsets(logDir))
log.close()
@@ -735,7 +735,7 @@ class UnifiedLogTest {
// Reload after unclean shutdown with recoveryPoint set to 0
log = createLog(logDir, logConfig, recoveryPoint = 0L, lastShutdownClean = false)
// We progressively create a snapshot for each segment after the recovery point
- expectedSnapshotOffsets = log.logSegments.map(_.baseOffset).tail.toVector :+ log.logEndOffset
+ expectedSnapshotOffsets = log.logSegments.asScala.map(_.baseOffset).tail.toVector :+ log.logEndOffset
assertEquals(expectedSnapshotOffsets, LogTestUtils.listProducerSnapshotOffsets(logDir))
log.close()
}
@@ -1051,7 +1051,7 @@ class UnifiedLogTest {
log.appendAsLeader(TestUtils.records(List(new SimpleRecord("a".getBytes, "c".getBytes())), producerId = pid1,
producerEpoch = epoch, sequence = 2), leaderEpoch = 0)
log.updateHighWatermark(log.logEndOffset)
- assertEquals(log.logSegments.map(_.baseOffset).toSeq.sorted.drop(1), ProducerStateManager.listSnapshotFiles(logDir).asScala.map(_.offset).sorted,
+ assertEquals(log.logSegments.asScala.map(_.baseOffset).toSeq.sorted.drop(1), ProducerStateManager.listSnapshotFiles(logDir).asScala.map(_.offset).sorted,
"expected a snapshot file per segment base offset, except the first segment")
assertEquals(2, ProducerStateManager.listSnapshotFiles(logDir).size)
@@ -1061,7 +1061,7 @@ class UnifiedLogTest {
log.deleteOldSegments()
// Sleep to breach the file delete delay and run scheduled file deletion tasks
mockTime.sleep(1)
- assertEquals(log.logSegments.map(_.baseOffset).toSeq.sorted.drop(1), ProducerStateManager.listSnapshotFiles(logDir).asScala.map(_.offset).sorted,
+ assertEquals(log.logSegments.asScala.map(_.baseOffset).toSeq.sorted.drop(1), ProducerStateManager.listSnapshotFiles(logDir).asScala.map(_.offset).sorted,
"expected a snapshot file per segment base offset, excluding the first")
}
@@ -1596,7 +1596,7 @@ class UnifiedLogTest {
log.appendAsLeader(TestUtils.singletonRecords(value = "42".getBytes), leaderEpoch = 0)
// now manually truncate off all but one message from the first segment to create a gap in the messages
- log.logSegments.head.truncateTo(1)
+ log.logSegments.asScala.head.truncateTo(1)
assertEquals(log.logEndOffset - 1, LogTestUtils.readLog(log, 1, 200).records.batches.iterator.next().lastOffset,
"A read should now return the last message in the log")
@@ -1996,7 +1996,7 @@ class UnifiedLogTest {
MemoryRecords.withRecords(100 + i, CompressionType.NONE, 0, new SimpleRecord(mockTime.milliseconds + i, i.toString.getBytes()))
}
messages.foreach(log.appendAsFollower)
- val timeIndexEntries = log.logSegments.foldLeft(0) { (entries, segment) => entries + segment.timeIndex.entries }
+ val timeIndexEntries = log.logSegments.asScala.foldLeft(0) { (entries, segment) => entries + segment.timeIndex.entries }
assertEquals(numMessages - 1, timeIndexEntries, s"There should be ${numMessages - 1} time index entries")
assertEquals(mockTime.milliseconds + numMessages - 1, log.activeSegment.timeIndex.lastEntry.timestamp,
s"The last time index entry should have timestamp ${mockTime.milliseconds + numMessages - 1}")
@@ -2203,16 +2203,16 @@ class UnifiedLogTest {
assertEquals(2, log.numberOfSegments, "There should be exactly 2 segment.")
val expectedEntries = msgPerSeg - 1
- assertEquals(expectedEntries, log.logSegments.toList.head.offsetIndex.maxEntries,
+ assertEquals(expectedEntries, log.logSegments.asScala.toList.head.offsetIndex.maxEntries,
s"The index of the first segment should have $expectedEntries entries")
- assertEquals(expectedEntries, log.logSegments.toList.head.timeIndex.maxEntries,
+ assertEquals(expectedEntries, log.logSegments.asScala.toList.head.timeIndex.maxEntries,
s"The time index of the first segment should have $expectedEntries entries")
log.truncateTo(0)
assertEquals(1, log.numberOfSegments, "There should be exactly 1 segment.")
- assertEquals(log.config.maxIndexSize/8, log.logSegments.toList.head.offsetIndex.maxEntries,
+ assertEquals(log.config.maxIndexSize/8, log.logSegments.asScala.toList.head.offsetIndex.maxEntries,
"The index of segment 1 should be resized to maxIndexSize")
- assertEquals(log.config.maxIndexSize/12, log.logSegments.toList.head.timeIndex.maxEntries,
+ assertEquals(log.config.maxIndexSize/12, log.logSegments.asScala.toList.head.timeIndex.maxEntries,
"The time index of segment 1 should be resized to maxIndexSize")
mockTime.sleep(msgPerSeg)
@@ -2238,22 +2238,22 @@ class UnifiedLogTest {
log.appendAsLeader(createRecords, leaderEpoch = 0)
// files should be renamed
- val segments = log.logSegments.toArray
- val oldFiles = segments.map(_.log.file) ++ segments.map(_.lazyOffsetIndex.file)
+ val segments = log.logSegments.asScala.toArray
+ val oldFiles = segments.map(_.log.file) ++ segments.map(_.offsetIndexFile)
log.updateHighWatermark(log.logEndOffset)
log.deleteOldSegments()
assertEquals(1, log.numberOfSegments, "Only one segment should remain.")
assertTrue(segments.forall(_.log.file.getName.endsWith(LogFileUtils.DELETED_FILE_SUFFIX)) &&
- segments.forall(_.lazyOffsetIndex.file.getName.endsWith(LogFileUtils.DELETED_FILE_SUFFIX)),
+ segments.forall(_.offsetIndexFile.getName.endsWith(LogFileUtils.DELETED_FILE_SUFFIX)),
"All log and index files should end in .deleted")
- assertTrue(segments.forall(_.log.file.exists) && segments.forall(_.lazyOffsetIndex.file.exists),
+ assertTrue(segments.forall(_.log.file.exists) && segments.forall(_.offsetIndexFile.exists),
"The .deleted files should still be there.")
assertTrue(oldFiles.forall(!_.exists), "The original file should be gone.")
// when enough time passes the files should be deleted
- val deletedFiles = segments.map(_.log.file) ++ segments.map(_.lazyOffsetIndex.file)
+ val deletedFiles = segments.map(_.log.file) ++ segments.map(_.offsetIndexFile)
mockTime.sleep(asyncDeleteMs + 1)
assertTrue(deletedFiles.forall(!_.exists), "Files should all be gone.")
}
@@ -2520,8 +2520,8 @@ class UnifiedLogTest {
private def testDegenerateSplitSegmentWithOverflow(segmentBaseOffset: Long, records: List[MemoryRecords]): Unit = {
val segment = LogTestUtils.rawSegment(logDir, segmentBaseOffset)
// Need to create the offset files explicitly to avoid triggering segment recovery to truncate segment.
- Files.createFile(UnifiedLog.offsetIndexFile(logDir, segmentBaseOffset).toPath)
- Files.createFile(UnifiedLog.timeIndexFile(logDir, segmentBaseOffset).toPath)
+ Files.createFile(LogFileUtils.offsetIndexFile(logDir, segmentBaseOffset).toPath)
+ Files.createFile(LogFileUtils.timeIndexFile(logDir, segmentBaseOffset).toPath)
records.foreach(segment.append _)
segment.close()
@@ -2568,8 +2568,8 @@ class UnifiedLogTest {
log.updateHighWatermark(hw)
log.deleteOldSegments()
assertTrue(log.logStartOffset <= hw)
- log.logSegments.foreach { segment =>
- val segmentFetchInfo = segment.read(startOffset = segment.baseOffset, maxSize = Int.MaxValue)
+ log.logSegments.forEach { segment =>
+ val segmentFetchInfo = segment.read(segment.baseOffset, Int.MaxValue)
val segmentLastOffsetOpt = segmentFetchInfo.records.records.asScala.lastOption.map(_.offset)
segmentLastOffsetOpt.foreach { lastOffset =>
assertTrue(lastOffset >= hw)
@@ -2715,7 +2715,7 @@ class UnifiedLogTest {
log.appendAsLeader(createRecords, leaderEpoch = 0)
// mark oldest segment as older the retention.ms
- log.logSegments.head.lastModified = mockTime.milliseconds - 20000
+ log.logSegments.asScala.head.setLastModified(mockTime.milliseconds - 20000)
val segments = log.numberOfSegments
log.updateHighWatermark(log.logEndOffset)
@@ -2750,7 +2750,7 @@ class UnifiedLogTest {
log.appendAsLeader(createRecords, leaderEpoch = 0)
// Three segments should be created
- assertEquals(3, log.logSegments.count(_ => true))
+ assertEquals(3, log.logSegments.asScala.count(_ => true))
log.updateHighWatermark(log.logEndOffset)
log.maybeIncrementLogStartOffset(recordsPerSegment, LogStartOffsetIncrementReason.ClientRecordDeletion)
@@ -2760,8 +2760,8 @@ class UnifiedLogTest {
log.updateHighWatermark(log.logEndOffset)
log.deleteOldSegments()
assertEquals(2, log.numberOfSegments, "There should be 2 segments remaining")
- assertTrue(log.logSegments.head.baseOffset <= log.logStartOffset)
- assertTrue(log.logSegments.tail.forall(s => s.baseOffset > log.logStartOffset))
+ assertTrue(log.logSegments.asScala.head.baseOffset <= log.logStartOffset)
+ assertTrue(log.logSegments.asScala.tail.forall(s => s.baseOffset > log.logStartOffset))
}
@Test
@@ -3149,9 +3149,9 @@ class UnifiedLogTest {
assertTrue(offsetMetadata.relativePositionInSegment <= segment.size)
val readInfo = segment.read(offsetMetadata.messageOffset,
- maxSize = 2048,
- maxPosition = segment.size,
- minOneMessage = false)
+ 2048,
+ segment.size,
+ false)
if (offsetMetadata.relativePositionInSegment < segment.size)
assertEquals(offsetMetadata, readInfo.fetchOffsetMetadata)
@@ -3483,11 +3483,11 @@ class UnifiedLogTest {
assertTrue(log.logStartOffset <= hw)
// verify that all segments up to the high watermark have been deleted
- log.logSegments.headOption.foreach { segment =>
+ log.logSegments.asScala.headOption.foreach { segment =>
assertTrue(segment.baseOffset <= hw)
assertTrue(segment.baseOffset >= log.logStartOffset)
}
- log.logSegments.tail.foreach { segment =>
+ log.logSegments.asScala.tail.foreach { segment =>
assertTrue(segment.baseOffset > hw)
assertTrue(segment.baseOffset >= log.logStartOffset)
}
@@ -3989,7 +3989,7 @@ class UnifiedLogTest {
object UnifiedLogTest {
def allRecords(log: UnifiedLog): List[Record] = {
val recordsFound = ListBuffer[Record]()
- for (logSegment <- log.logSegments) {
+ for (logSegment <- log.logSegments.asScala) {
for (batch <- logSegment.log.batches.asScala) {
recordsFound ++= batch.iterator().asScala
}
diff --git a/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala b/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
index 2751f0c71af..64481b85b78 100644
--- a/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicConfigChangeTest.scala
@@ -143,7 +143,7 @@ class DynamicConfigChangeTest extends KafkaServerTestHarness {
(1 to 50).foreach(i => TestUtils.produceMessage(brokers, tp.topic, i.toString))
// Verify that the new config is used for all segments
- assertTrue(log.logSegments.forall(_.size > 1000), "Log segment size change not applied")
+ assertTrue(log.logSegments.stream.allMatch(_.size > 1000), "Log segment size change not applied")
}
@nowarn("cat=deprecation")
diff --git a/core/src/test/scala/unit/kafka/server/LogOffsetTest.scala b/core/src/test/scala/unit/kafka/server/LogOffsetTest.scala
index c2ca4fb869a..6783e83ca44 100755
--- a/core/src/test/scala/unit/kafka/server/LogOffsetTest.scala
+++ b/core/src/test/scala/unit/kafka/server/LogOffsetTest.scala
@@ -17,7 +17,7 @@
package kafka.server
-import kafka.log.{LogSegment, UnifiedLog}
+import kafka.log.UnifiedLog
import kafka.utils.TestUtils
import org.apache.kafka.common.message.ListOffsetsRequestData.{ListOffsetsPartition, ListOffsetsTopic}
import org.apache.kafka.common.message.ListOffsetsResponseData.{ListOffsetsPartitionResponse, ListOffsetsTopicResponse}
@@ -25,7 +25,7 @@ import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.requests.{FetchRequest, FetchResponse, ListOffsetsRequest, ListOffsetsResponse}
import org.apache.kafka.common.utils.Time
import org.apache.kafka.common.{IsolationLevel, TopicPartition}
-import org.apache.kafka.storage.internals.log.LogStartOffsetIncrementReason
+import org.apache.kafka.storage.internals.log.{LogSegment, LogStartOffsetIncrementReason}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Timeout
import org.junit.jupiter.params.ParameterizedTest
@@ -35,6 +35,8 @@ import org.mockito.invocation.InvocationOnMock
import org.mockito.stubbing.Answer
import java.io.File
+import java.util
+import java.util.Arrays.asList
import java.util.concurrent.atomic.AtomicInteger
import java.util.{Optional, Properties, Random}
import scala.collection.mutable.Buffer
@@ -276,7 +278,7 @@ class LogOffsetTest extends BaseRequestTest {
private[this] val value = new AtomicInteger(0)
override def answer(invocation: InvocationOnMock): Int = value.getAndIncrement()
})
- val logSegments = Seq(logSegment)
+ val logSegments = Seq(logSegment).asJava
when(log.logSegments).thenReturn(logSegments)
log.legacyFetchOffsetsBefore(System.currentTimeMillis, 100)
}
@@ -289,9 +291,9 @@ class LogOffsetTest extends BaseRequestTest {
val log: UnifiedLog = mock(classOf[UnifiedLog])
val logSegment: LogSegment = mock(classOf[LogSegment])
when(log.logSegments).thenReturn(
- new Iterable[LogSegment] {
+ new util.AbstractCollection[LogSegment] {
override def size = 2
- override def iterator = Seq(logSegment).iterator
+ override def iterator = asList(logSegment).iterator
}
)
log.legacyFetchOffsetsBefore(System.currentTimeMillis, 100)
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index 0921a747bae..3054cc4c364 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -60,7 +60,7 @@ import org.apache.kafka.server.common.OffsetAndEpoch
import org.apache.kafka.server.common.MetadataVersion.IBP_2_6_IV0
import org.apache.kafka.server.metrics.{KafkaMetricsGroup, KafkaYammerMetrics}
import org.apache.kafka.server.util.{MockScheduler, MockTime}
-import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchDataInfo, FetchIsolation, FetchParams, FetchPartitionData, LogConfig, LogDirFailureChannel, LogOffsetMetadata, LogStartOffsetIncrementReason, ProducerStateManager, ProducerStateManagerConfig, RemoteStorageFetchInfo}
+import org.apache.kafka.storage.internals.log.{AppendOrigin, FetchDataInfo, FetchIsolation, FetchParams, FetchPartitionData, LogConfig, LogDirFailureChannel, LogOffsetMetadata, LogSegments, LogStartOffsetIncrementReason, ProducerStateManager, ProducerStateManagerConfig, RemoteStorageFetchInfo}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import org.junit.jupiter.params.ParameterizedTest
@@ -2617,7 +2617,7 @@ class ReplicaManagerTest {
segments,
0L,
0L,
- leaderEpochCache,
+ leaderEpochCache.asJava,
producerStateManager
).load()
val localLog = new LocalLog(logDir, logConfig, segments, offsets.recoveryPoint,
diff --git a/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala b/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala
index b939db2a8c1..133f37420df 100644
--- a/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala
+++ b/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala
@@ -246,12 +246,11 @@ class LeaderEpochIntegrationTest extends QuorumTestHarness with Logging {
val tp = new TopicPartition(topic, 0)
val leo = broker.getLogManager.getLog(tp).get.logEndOffset
result = result && leo > 0 && brokers.forall { broker =>
- broker.getLogManager.getLog(tp).get.logSegments.iterator.forall { segment =>
+ broker.getLogManager.getLog(tp).get.logSegments.stream.allMatch { segment =>
if (segment.read(minOffset, Integer.MAX_VALUE) == null) {
false
} else {
- segment.read(minOffset, Integer.MAX_VALUE)
- .records.batches().iterator().asScala.forall(
+ segment.read(minOffset, Integer.MAX_VALUE).records.batches().iterator().asScala.forall(
expectedLeaderEpoch == _.partitionLeaderEpoch()
)
}
diff --git a/core/src/test/scala/unit/kafka/utils/CoreUtilsTest.scala b/core/src/test/scala/unit/kafka/utils/CoreUtilsTest.scala
index 832810d82f6..ee3401062c8 100755
--- a/core/src/test/scala/unit/kafka/utils/CoreUtilsTest.scala
+++ b/core/src/test/scala/unit/kafka/utils/CoreUtilsTest.scala
@@ -31,7 +31,6 @@ import org.apache.kafka.common.utils.Utils
import org.slf4j.event.Level
import scala.jdk.CollectionConverters._
-import scala.collection.mutable
import scala.concurrent.duration.Duration
import scala.concurrent.{Await, ExecutionContext, ExecutionContextExecutorService, Future}
@@ -44,56 +43,6 @@ class CoreUtilsTest extends Logging {
CoreUtils.swallow(throw new KafkaException("test"), this, Level.INFO)
}
- @Test
- def testTryAll(): Unit = {
- case class TestException(key: String) extends Exception
-
- val recorded = mutable.Map.empty[String, Either[TestException, String]]
- def recordingFunction(v: Either[TestException, String]): Unit = {
- val key = v match {
- case Right(key) => key
- case Left(e) => e.key
- }
- recorded(key) = v
- }
-
- CoreUtils.tryAll(Seq(
- () => recordingFunction(Right("valid-0")),
- () => recordingFunction(Left(TestException("exception-1"))),
- () => recordingFunction(Right("valid-2")),
- () => recordingFunction(Left(TestException("exception-3")))
- ))
- var expected = Map(
- "valid-0" -> Right("valid-0"),
- "exception-1" -> Left(TestException("exception-1")),
- "valid-2" -> Right("valid-2"),
- "exception-3" -> Left(TestException("exception-3"))
- )
- assertEquals(expected, recorded)
-
- recorded.clear()
- CoreUtils.tryAll(Seq(
- () => recordingFunction(Right("valid-0")),
- () => recordingFunction(Right("valid-1"))
- ))
- expected = Map(
- "valid-0" -> Right("valid-0"),
- "valid-1" -> Right("valid-1")
- )
- assertEquals(expected, recorded)
-
- recorded.clear()
- CoreUtils.tryAll(Seq(
- () => recordingFunction(Left(TestException("exception-0"))),
- () => recordingFunction(Left(TestException("exception-1")))
- ))
- expected = Map(
- "exception-0" -> Left(TestException("exception-0")),
- "exception-1" -> Left(TestException("exception-1"))
- )
- assertEquals(expected, recorded)
- }
-
@Test
def testReadBytes(): Unit = {
for(testCase <- List("", "a", "abcd")) {
diff --git a/core/src/test/scala/unit/kafka/utils/SchedulerTest.scala b/core/src/test/scala/unit/kafka/utils/SchedulerTest.scala
index 1aa04d6c846..eb9c03343e5 100644
--- a/core/src/test/scala/unit/kafka/utils/SchedulerTest.scala
+++ b/core/src/test/scala/unit/kafka/utils/SchedulerTest.scala
@@ -19,14 +19,16 @@ package kafka.utils
import java.util.Properties
import java.util.concurrent.atomic._
import java.util.concurrent.{CountDownLatch, Executors, TimeUnit}
-import kafka.log.{LocalLog, LogLoader, LogSegments, UnifiedLog}
+import kafka.log.{LocalLog, LogLoader, UnifiedLog}
import kafka.server.BrokerTopicStats
import kafka.utils.TestUtils.retry
import org.apache.kafka.server.util.{KafkaScheduler, MockTime}
-import org.apache.kafka.storage.internals.log.{LogConfig, LogDirFailureChannel, ProducerStateManager, ProducerStateManagerConfig}
+import org.apache.kafka.storage.internals.log.{LogConfig, LogDirFailureChannel, LogSegments, ProducerStateManager, ProducerStateManagerConfig}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test, Timeout}
+import scala.compat.java8.OptionConverters._
+
class SchedulerTest {
val scheduler = new KafkaScheduler(1)
@@ -150,7 +152,7 @@ class SchedulerTest {
segments,
0L,
0L,
- leaderEpochCache,
+ leaderEpochCache.asJava,
producerStateManager
).load()
val localLog = new LocalLog(logDir, logConfig, segments, offsets.recoveryPoint,
diff --git a/gradle/spotbugs-exclude.xml b/gradle/spotbugs-exclude.xml
index 9bcb95e1b26..9b7f4b7669c 100644
--- a/gradle/spotbugs-exclude.xml
+++ b/gradle/spotbugs-exclude.xml
@@ -161,7 +161,10 @@ For a detailed description of spotbugs bug categories, see https://spotbugs.read
<!-- Scala doesn't have checked exceptions so one cannot catch RuntimeException and rely
on the compiler to fail if the code is changed to call a method that throws Exception.
Given that, this bug pattern doesn't make sense for Scala code. -->
- <Class name="kafka.log.Log"/>
+ <Or>
+ <Class name="kafka.log.Log"/>
+ <Class name="kafka.log.LocalLog$"/>
+ </Or>
<Bug pattern="REC_CATCH_EXCEPTION"/>
</Match>
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LazyIndex.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LazyIndex.java
index 9d0725e0e33..feb6dc94952 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LazyIndex.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LazyIndex.java
@@ -41,7 +41,7 @@ import org.apache.kafka.common.utils.Utils;
* Methods of this class are thread safe. Make sure to check `AbstractIndex` subclasses
* documentation to establish their thread safety.
*/
-public class LazyIndex<T extends AbstractIndex> {
+public class LazyIndex<T extends AbstractIndex> implements Closeable {
private enum IndexType {
OFFSET, TIME
@@ -214,6 +214,7 @@ public class LazyIndex<T extends AbstractIndex> {
}
}
+ @Override
public void close() throws IOException {
lock.lock();
try {
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java
new file mode 100644
index 00000000000..583a7604328
--- /dev/null
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java
@@ -0,0 +1,889 @@
+/*
+ * 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.storage.internals.log;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.attribute.FileTime;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.Timer;
+import org.apache.kafka.common.InvalidRecordException;
+import org.apache.kafka.common.errors.CorruptRecordException;
+import org.apache.kafka.common.record.FileLogInputStream.FileChannelRecordBatch;
+import org.apache.kafka.common.record.FileRecords;
+import org.apache.kafka.common.record.FileRecords.LogOffsetPosition;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.utils.BufferSupplier;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.metrics.KafkaMetricsGroup;
+import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.event.Level;
+
+import static java.util.Arrays.asList;
+
+/**
+ * A segment of the log. Each segment has two components: a log and an index. The log is a FileRecords containing
+ * the actual messages. The index is an OffsetIndex that maps from logical offsets to physical file positions. Each
+ * segment has a base offset which is an offset <= the least offset of any message in this segment and > any offset in
+ * any previous segment.
+ *
+ * A segment with a base offset of [base_offset] would be stored in two files, a [base_offset].index and a [base_offset].log file.
+ *
+ * This class is not thread-safe.
+ */
+public class LogSegment implements Closeable {
+ private static final Logger LOGGER = LoggerFactory.getLogger(LogSegment.class);
+ private static final Timer LOG_FLUSH_TIMER;
+
+ static {
+ KafkaMetricsGroup logFlushStatsMetricsGroup = new KafkaMetricsGroup(LogSegment.class) {
+ @Override
+ public MetricName metricName(String name, Map<String, String> tags) {
+ // Override the group and type names for compatibility - this metrics group was previously defined within
+ // a Scala object named `kafka.log.LogFlushStats`
+ return KafkaMetricsGroup.explicitMetricName("kafka.log", "LogFlushStats", name, tags);
+ }
+ };
+ LOG_FLUSH_TIMER = logFlushStatsMetricsGroup.newTimer("LogFlushRateAndTimeMs", TimeUnit.MILLISECONDS, TimeUnit.SECONDS);
+ }
+
+ private final FileRecords log;
+ private final LazyIndex<OffsetIndex> lazyOffsetIndex;
+ private final LazyIndex<TimeIndex> lazyTimeIndex;
+ private final TransactionIndex txnIndex;
+ private final long baseOffset;
+ private final int indexIntervalBytes;
+ private final long rollJitterMs;
+ private final Time time;
+
+ // The timestamp we used for time based log rolling and for ensuring max compaction delay
+ // volatile for LogCleaner to see the update
+ private volatile OptionalLong rollingBasedTimestamp = OptionalLong.empty();
+
+ /* The maximum timestamp and offset we see so far */
+ private volatile TimestampOffset maxTimestampAndOffsetSoFar = TimestampOffset.UNKNOWN;
+
+ private long created;
+
+ /* the number of bytes since we last added an entry in the offset index */
+ private int bytesSinceLastIndexEntry = 0;
+
+ /**
+ * Create a LogSegment with the provided parameters.
+ *
+ * @param log The file records containing log entries
+ * @param lazyOffsetIndex The offset index
+ * @param lazyTimeIndex The timestamp index
+ * @param txnIndex The transaction index
+ * @param baseOffset A lower bound on the offsets in this segment
+ * @param indexIntervalBytes The approximate number of bytes between entries in the index
+ * @param rollJitterMs The maximum random jitter subtracted from the scheduled segment roll time
+ * @param time The time instance
+ */
+ public LogSegment(FileRecords log,
+ LazyIndex<OffsetIndex> lazyOffsetIndex,
+ LazyIndex<TimeIndex> lazyTimeIndex,
+ TransactionIndex txnIndex,
+ long baseOffset,
+ int indexIntervalBytes,
+ long rollJitterMs,
+ Time time) {
+ this.log = log;
+ this.lazyOffsetIndex = lazyOffsetIndex;
+ this.lazyTimeIndex = lazyTimeIndex;
+ this.txnIndex = txnIndex;
+ this.baseOffset = baseOffset;
+ this.indexIntervalBytes = indexIntervalBytes;
+ this.rollJitterMs = rollJitterMs;
+ this.time = time;
+ this.created = time.milliseconds();
+ }
+
+ // Visible for testing
+ public LogSegment(LogSegment segment) {
+ this(segment.log, segment.lazyOffsetIndex, segment.lazyTimeIndex, segment.txnIndex, segment.baseOffset,
+ segment.indexIntervalBytes, segment.rollJitterMs, segment.time);
+ }
+
+ public OffsetIndex offsetIndex() throws IOException {
+ return lazyOffsetIndex.get();
+ }
+
+ public File offsetIndexFile() {
+ return lazyOffsetIndex.file();
+ }
+
+ public TimeIndex timeIndex() throws IOException {
+ return lazyTimeIndex.get();
+ }
+
+ public File timeIndexFile() {
+ return lazyTimeIndex.file();
+ }
+
+ public long baseOffset() {
+ return baseOffset;
+ }
+
+ public FileRecords log() {
+ return log;
+ }
+
+ public long rollJitterMs() {
+ return rollJitterMs;
+ }
+
+ public TransactionIndex txnIndex() {
+ return txnIndex;
+ }
+
+ public boolean shouldRoll(RollParams rollParams) throws IOException {
+ boolean reachedRollMs = timeWaitedForRoll(rollParams.now, rollParams.maxTimestampInMessages) > rollParams.maxSegmentMs - rollJitterMs;
+ int size = size();
+ return size > rollParams.maxSegmentBytes - rollParams.messagesSize ||
+ (size > 0 && reachedRollMs) ||
+ offsetIndex().isFull() || timeIndex().isFull() || !canConvertToRelativeOffset(rollParams.maxOffsetInMessages);
+ }
+
+ public void resizeIndexes(int size) throws IOException {
+ offsetIndex().resize(size);
+ timeIndex().resize(size);
+ }
+
+ public void sanityCheck(boolean timeIndexFileNewlyCreated) throws IOException {
+ if (offsetIndexFile().exists()) {
+ // Resize the time index file to 0 if it is newly created.
+ if (timeIndexFileNewlyCreated)
+ timeIndex().resize(0);
+ // Sanity checks for time index and offset index are skipped because
+ // we will recover the segments above the recovery point in recoverLog()
+ // in any case so sanity checking them here is redundant.
+ txnIndex.sanityCheck();
+ } else
+ throw new NoSuchFileException("Offset index file " + offsetIndexFile().getAbsolutePath() + " does not exist");
+ }
+
+ /**
+ * The first time this is invoked, it will result in a time index lookup (including potential materialization of
+ * the time index).
+ */
+ public TimestampOffset readMaxTimestampAndOffsetSoFar() throws IOException {
+ if (maxTimestampAndOffsetSoFar == TimestampOffset.UNKNOWN)
+ maxTimestampAndOffsetSoFar = timeIndex().lastEntry();
+ return maxTimestampAndOffsetSoFar;
+ }
+
+ /**
+ * The maximum timestamp we see so far.
+ *
+ * Note that this may result in time index materialization.
+ */
+ public long maxTimestampSoFar() throws IOException {
+ return readMaxTimestampAndOffsetSoFar().timestamp;
+ }
+
+ /**
+ * Note that this may result in time index materialization.
+ */
+ private long offsetOfMaxTimestampSoFar() throws IOException {
+ return readMaxTimestampAndOffsetSoFar().offset;
+ }
+
+ /* Return the size in bytes of this log segment */
+ public int size() {
+ return log.sizeInBytes();
+ }
+
+ /**
+ * checks that the argument offset can be represented as an integer offset relative to the baseOffset.
+ */
+ private boolean canConvertToRelativeOffset(long offset) throws IOException {
+ return offsetIndex().canAppendOffset(offset);
+ }
+
+ /**
+ * Append the given messages starting with the given offset. Add
+ * an entry to the index if needed.
+ *
+ * It is assumed this method is being called from within a lock, it is not thread-safe otherwise.
+ *
+ * @param largestOffset The last offset in the message set
+ * @param largestTimestampMs The largest timestamp in the message set.
+ * @param shallowOffsetOfMaxTimestamp The offset of the message that has the largest timestamp in the messages to append.
+ * @param records The log entries to append.
+ * @throws LogSegmentOffsetOverflowException if the largest offset causes index offset overflow
+ */
+ public void append(long largestOffset,
+ long largestTimestampMs,
+ long shallowOffsetOfMaxTimestamp,
+ MemoryRecords records) throws IOException {
+ if (records.sizeInBytes() > 0) {
+ LOGGER.trace("Inserting {} bytes at end offset {} at position {} with largest timestamp {} at shallow offset {}",
+ records.sizeInBytes(), largestOffset, log.sizeInBytes(), largestTimestampMs, shallowOffsetOfMaxTimestamp);
+ int physicalPosition = log.sizeInBytes();
+ if (physicalPosition == 0)
+ rollingBasedTimestamp = OptionalLong.of(largestTimestampMs);
+
+ ensureOffsetInRange(largestOffset);
+
+ // append the messages
+ long appendedBytes = log.append(records);
+ LOGGER.trace("Appended {} to {} at end offset {}", appendedBytes, log.file(), largestOffset);
+ // Update the in memory max timestamp and corresponding offset.
+ if (largestTimestampMs > maxTimestampSoFar()) {
+ maxTimestampAndOffsetSoFar = new TimestampOffset(largestTimestampMs, shallowOffsetOfMaxTimestamp);
+ }
+ // append an entry to the index (if needed)
+ if (bytesSinceLastIndexEntry > indexIntervalBytes) {
+ offsetIndex().append(largestOffset, physicalPosition);
+ timeIndex().maybeAppend(maxTimestampSoFar(), offsetOfMaxTimestampSoFar());
+ bytesSinceLastIndexEntry = 0;
+ }
+ bytesSinceLastIndexEntry += records.sizeInBytes();
+ }
+ }
+
+ private void ensureOffsetInRange(long offset) throws IOException {
+ if (!canConvertToRelativeOffset(offset))
+ throw new LogSegmentOffsetOverflowException(this, offset);
+ }
+
+ private int appendChunkFromFile(FileRecords records, int position, BufferSupplier bufferSupplier) throws IOException {
+ int bytesToAppend = 0;
+ long maxTimestamp = Long.MIN_VALUE;
+ long offsetOfMaxTimestamp = Long.MIN_VALUE;
+ long maxOffset = Long.MIN_VALUE;
+ ByteBuffer readBuffer = bufferSupplier.get(1024 * 1024);
+
+ // find all batches that are valid to be appended to the current log segment and
+ // determine the maximum offset and timestamp
+ Iterator<FileChannelRecordBatch> nextBatches = records.batchesFrom(position).iterator();
+ FileChannelRecordBatch batch;
+ while ((batch = nextAppendableBatch(nextBatches, readBuffer, bytesToAppend)) != null) {
+ if (batch.maxTimestamp() > maxTimestamp) {
+ maxTimestamp = batch.maxTimestamp();
+ offsetOfMaxTimestamp = batch.lastOffset();
+ }
+ maxOffset = batch.lastOffset();
+ bytesToAppend += batch.sizeInBytes();
+ }
+
+ if (bytesToAppend > 0) {
+ // Grow buffer if needed to ensure we copy at least one batch
+ if (readBuffer.capacity() < bytesToAppend)
+ readBuffer = bufferSupplier.get(bytesToAppend);
+
+ readBuffer.limit(bytesToAppend);
+ records.readInto(readBuffer, position);
+
+ append(maxOffset, maxTimestamp, offsetOfMaxTimestamp, MemoryRecords.readableRecords(readBuffer));
+ }
+
+ bufferSupplier.release(readBuffer);
+ return bytesToAppend;
+ }
+
+ private FileChannelRecordBatch nextAppendableBatch(Iterator<FileChannelRecordBatch> recordBatches,
+ ByteBuffer readBuffer,
+ int bytesToAppend) throws IOException {
+ if (recordBatches.hasNext()) {
+ FileChannelRecordBatch batch = recordBatches.next();
+ if (canConvertToRelativeOffset(batch.lastOffset()) &&
+ (bytesToAppend == 0 || bytesToAppend + batch.sizeInBytes() < readBuffer.capacity()))
+ return batch;
+ }
+ return null;
+ }
+
+ /**
+ * Append records from a file beginning at the given position until either the end of the file
+ * is reached or an offset is found which is too large to convert to a relative offset for the indexes.
+ *
+ * @return the number of bytes appended to the log (may be less than the size of the input if an
+ * offset is encountered which would overflow this segment)
+ */
+ public int appendFromFile(FileRecords records, int start) throws IOException {
+ int position = start;
+ BufferSupplier bufferSupplier = new BufferSupplier.GrowableBufferSupplier();
+ while (position < start + records.sizeInBytes()) {
+ int bytesAppended = appendChunkFromFile(records, position, bufferSupplier);
+ if (bytesAppended == 0)
+ return position - start;
+ position += bytesAppended;
+ }
+ return position - start;
+ }
+
+ /* not thread safe */
+ public void updateTxnIndex(CompletedTxn completedTxn, long lastStableOffset) throws IOException {
+ if (completedTxn.isAborted) {
+ LOGGER.trace("Writing aborted transaction {} to transaction index, last stable offset is {}", completedTxn, lastStableOffset);
+ txnIndex.append(new AbortedTxn(completedTxn, lastStableOffset));
+ }
+ }
+
+ private void updateProducerState(ProducerStateManager producerStateManager, RecordBatch batch) throws IOException {
+ if (batch.hasProducerId()) {
+ long producerId = batch.producerId();
+ ProducerAppendInfo appendInfo = producerStateManager.prepareUpdate(producerId, AppendOrigin.REPLICATION);
+ Optional<CompletedTxn> maybeCompletedTxn = appendInfo.append(batch, Optional.empty());
+ producerStateManager.update(appendInfo);
+ if (maybeCompletedTxn.isPresent()) {
+ CompletedTxn completedTxn = maybeCompletedTxn.get();
+ long lastStableOffset = producerStateManager.lastStableOffset(completedTxn);
+ updateTxnIndex(completedTxn, lastStableOffset);
+ producerStateManager.completeTxn(completedTxn);
+ }
+ }
+ producerStateManager.updateMapEndOffset(batch.lastOffset() + 1);
+ }
+
+ /**
+ * Equivalent to {@code translateOffset(offset, 0)}.
+ *
+ * See {@link #translateOffset(long, int)} for details.
+ */
+ public LogOffsetPosition translateOffset(long offset) throws IOException {
+ return translateOffset(offset, 0);
+ }
+
+ /**
+ * Find the physical file position for the first message with offset >= the requested offset.
+ *
+ * The startingFilePosition argument is an optimization that can be used if we already know a valid starting position
+ * in the file higher than the greatest-lower-bound from the index.
+ *
+ * This method is thread-safe.
+ *
+ * @param offset The offset we want to translate
+ * @param startingFilePosition A lower bound on the file position from which to begin the search. This is purely an optimization and
+ * when omitted, the search will begin at the position in the offset index.
+ * @return The position in the log storing the message with the least offset >= the requested offset and the size of the
+ * message or null if no message meets this criteria.
+ */
+ LogOffsetPosition translateOffset(long offset, int startingFilePosition) throws IOException {
+ OffsetPosition mapping = offsetIndex().lookup(offset);
+ return log.searchForOffsetWithSize(offset, Math.max(mapping.position, startingFilePosition));
+ }
+
+ /**
+ * Equivalent to {@code read(startOffset, maxSize, size())}.
+ *
+ * See {@link #read(long, int, long, boolean)} for details.
+ */
+ public FetchDataInfo read(long startOffset, int maxSize) throws IOException {
+ return read(startOffset, maxSize, size());
+ }
+
+ /**
+ * Equivalent to {@code read(startOffset, maxSize, maxPosition, false)}.
+ *
+ * See {@link #read(long, int, long, boolean)} for details.
+ */
+ public FetchDataInfo read(long startOffset, int maxSize, long maxPosition) throws IOException {
+ return read(startOffset, maxSize, maxPosition, false);
+ }
+
+ /**
+ * Read a message set from this segment beginning with the first offset >= startOffset. The message set will include
+ * no more than maxSize bytes and will end before maxOffset if a maxOffset is specified.
+ *
+ * This method is thread-safe.
+ *
+ * @param startOffset A lower bound on the first offset to include in the message set we read
+ * @param maxSize The maximum number of bytes to include in the message set we read
+ * @param maxPosition The maximum position in the log segment that should be exposed for read
+ * @param minOneMessage If this is true, the first message will be returned even if it exceeds `maxSize` (if one exists)
+ *
+ * @return The fetched data and the offset metadata of the first message whose offset is >= startOffset,
+ * or null if the startOffset is larger than the largest offset in this log
+ */
+ public FetchDataInfo read(long startOffset, int maxSize, long maxPosition, boolean minOneMessage) throws IOException {
+ if (maxSize < 0)
+ throw new IllegalArgumentException("Invalid max size " + maxSize + " for log read from segment " + log);
+
+ LogOffsetPosition startOffsetAndSize = translateOffset(startOffset);
+
+ // if the start position is already off the end of the log, return null
+ if (startOffsetAndSize == null)
+ return null;
+
+ int startPosition = startOffsetAndSize.position;
+ LogOffsetMetadata offsetMetadata = new LogOffsetMetadata(startOffset, this.baseOffset, startPosition);
+
+ int adjustedMaxSize = maxSize;
+ if (minOneMessage)
+ adjustedMaxSize = Math.max(maxSize, startOffsetAndSize.size);
+
+ // return a log segment but with zero size in the case below
+ if (adjustedMaxSize == 0)
+ return new FetchDataInfo(offsetMetadata, MemoryRecords.EMPTY);
+
+ // calculate the length of the message set to read based on whether or not they gave us a maxOffset
+ int fetchSize = Math.min((int) (maxPosition - startPosition), adjustedMaxSize);
+
+ return new FetchDataInfo(offsetMetadata, log.slice(startPosition, fetchSize),
+ adjustedMaxSize < startOffsetAndSize.size, Optional.empty());
+ }
+
+ public OptionalLong fetchUpperBoundOffset(OffsetPosition startOffsetPosition, int fetchSize) throws IOException {
+ Optional<OffsetPosition> position = offsetIndex().fetchUpperBoundOffset(startOffsetPosition, fetchSize);
+ if (position.isPresent())
+ return OptionalLong.of(position.get().offset);
+ return OptionalLong.empty();
+ }
+
+ /**
+ * Run recovery on the given segment. This will rebuild the index from the log file and lop off any invalid bytes
+ * from the end of the log and index.
+ *
+ * This method is not thread-safe.
+ *
+ * @param producerStateManager Producer state corresponding to the segment's base offset. This is needed to recover
+ * the transaction index.
+ * @param leaderEpochCache Optionally a cache for updating the leader epoch during recovery.
+ * @return The number of bytes truncated from the log
+ * @throws LogSegmentOffsetOverflowException if the log segment contains an offset that causes the index offset to overflow
+ */
+ public int recover(ProducerStateManager producerStateManager, Optional<LeaderEpochFileCache> leaderEpochCache) throws IOException {
+ offsetIndex().reset();
+ timeIndex().reset();
+ txnIndex.reset();
+ int validBytes = 0;
+ int lastIndexEntry = 0;
+ maxTimestampAndOffsetSoFar = TimestampOffset.UNKNOWN;
+ try {
+ for (RecordBatch batch : log.batches()) {
+ batch.ensureValid();
+ ensureOffsetInRange(batch.lastOffset());
+
+ // The max timestamp is exposed at the batch level, so no need to iterate the records
+ if (batch.maxTimestamp() > maxTimestampSoFar()) {
+ maxTimestampAndOffsetSoFar = new TimestampOffset(batch.maxTimestamp(), batch.lastOffset());
+ }
+
+ // Build offset index
+ if (validBytes - lastIndexEntry > indexIntervalBytes) {
+ offsetIndex().append(batch.lastOffset(), validBytes);
+ timeIndex().maybeAppend(maxTimestampSoFar(), offsetOfMaxTimestampSoFar());
+ lastIndexEntry = validBytes;
+ }
+ validBytes += batch.sizeInBytes();
+
+ if (batch.magic() >= RecordBatch.MAGIC_VALUE_V2) {
+ leaderEpochCache.ifPresent(cache -> {
+ if (batch.partitionLeaderEpoch() >= 0 &&
+ (!cache.latestEpoch().isPresent() || batch.partitionLeaderEpoch() > cache.latestEpoch().getAsInt()))
+ cache.assign(batch.partitionLeaderEpoch(), batch.baseOffset());
+ });
+ updateProducerState(producerStateManager, batch);
+ }
+ }
+ } catch (CorruptRecordException | InvalidRecordException e) {
+ LOGGER.warn("Found invalid messages in log segment {} at byte offset {}: {}. {}", log.file().getAbsolutePath(),
+ validBytes, e.getMessage(), e.getCause());
+ }
+ int truncated = log.sizeInBytes() - validBytes;
+ if (truncated > 0)
+ LOGGER.debug("Truncated {} invalid bytes at the end of segment {} during recovery", truncated, log.file().getAbsolutePath());
+
+ log.truncateTo(validBytes);
+ offsetIndex().trimToValidSize();
+ // A normally closed segment always appends the biggest timestamp ever seen into log segment, we do this as well.
+ timeIndex().maybeAppend(maxTimestampSoFar(), offsetOfMaxTimestampSoFar(), true);
+ timeIndex().trimToValidSize();
+ return truncated;
+ }
+
+ /**
+ * Check whether the last offset of the last batch in this segment overflows the indexes.
+ */
+ public boolean hasOverflow() throws IOException {
+ long nextOffset = readNextOffset();
+ return nextOffset > baseOffset && !canConvertToRelativeOffset(nextOffset - 1);
+ }
+
+ public TxnIndexSearchResult collectAbortedTxns(long fetchOffset, long upperBoundOffset) {
+ return txnIndex.collectAbortedTxns(fetchOffset, upperBoundOffset);
+ }
+
+ @Override
+ public String toString() {
+ // We don't call `largestRecordTimestamp` below to avoid materializing the time index when `toString` is invoked
+ return "LogSegment(baseOffset=" + baseOffset +
+ ", size=" + size() +
+ ", lastModifiedTime=" + lastModified() +
+ ", largestRecordTimestamp=" + maxTimestampAndOffsetSoFar.timestamp +
+ ")";
+ }
+
+ /**
+ * Truncate off all index and log entries with offsets >= the given offset.
+ * If the given offset is larger than the largest message in this segment, do nothing.
+ *
+ * This method is not thread-safe.
+ *
+ * @param offset The offset to truncate to
+ * @return The number of log bytes truncated
+ */
+ public int truncateTo(long offset) throws IOException {
+ // Do offset translation before truncating the index to avoid needless scanning
+ // in case we truncate the full index
+ LogOffsetPosition mapping = translateOffset(offset);
+ OffsetIndex offsetIndex = offsetIndex();
+ TimeIndex timeIndex = timeIndex();
+
+ offsetIndex.truncateTo(offset);
+ timeIndex.truncateTo(offset);
+ txnIndex.truncateTo(offset);
+
+ // After truncation, reset and allocate more space for the (new currently active) index
+ offsetIndex.resize(offsetIndex.maxIndexSize());
+ timeIndex.resize(timeIndex.maxIndexSize());
+
+ int bytesTruncated;
+ if (mapping == null)
+ bytesTruncated = 0;
+ else
+ bytesTruncated = log.truncateTo(mapping.position);
+
+ if (log.sizeInBytes() == 0) {
+ created = time.milliseconds();
+ rollingBasedTimestamp = OptionalLong.empty();
+ }
+
+ bytesSinceLastIndexEntry = 0;
+ if (maxTimestampSoFar() >= 0)
+ maxTimestampAndOffsetSoFar = readLargestTimestamp();
+
+ return bytesTruncated;
+ }
+
+ private TimestampOffset readLargestTimestamp() throws IOException {
+ // Get the last time index entry. If the time index is empty, it will return (-1, baseOffset)
+ TimestampOffset lastTimeIndexEntry = timeIndex().lastEntry();
+ OffsetPosition offsetPosition = offsetIndex().lookup(lastTimeIndexEntry.offset);
+
+ // Scan the rest of the messages to see if there is a larger timestamp after the last time index entry.
+ FileRecords.TimestampAndOffset maxTimestampOffsetAfterLastEntry = log.largestTimestampAfter(offsetPosition.position);
+ if (maxTimestampOffsetAfterLastEntry.timestamp > lastTimeIndexEntry.timestamp)
+ return new TimestampOffset(maxTimestampOffsetAfterLastEntry.timestamp, maxTimestampOffsetAfterLastEntry.offset);
+
+ return lastTimeIndexEntry;
+ }
+
+ /**
+ * Calculate the offset that would be used for the next message to be append to this segment.
+ * Note that this is expensive.
+ *
+ * This method is thread-safe.
+ */
+ public long readNextOffset() throws IOException {
+ FetchDataInfo fetchData = read(offsetIndex().lastOffset(), log.sizeInBytes());
+ if (fetchData == null)
+ return baseOffset;
+ else
+ return fetchData.records.lastBatch()
+ .map(batch -> batch.nextOffset())
+ .orElse(baseOffset);
+ }
+
+ /**
+ * Flush this log segment to disk.
+ *
+ * This method is thread-safe.
+ */
+ public void flush() throws IOException {
+ try {
+ LOG_FLUSH_TIMER.time(new Callable<Void>() {
+ // lambdas cannot declare a more specific exception type, so we use an anonymous inner class
+ @Override
+ public Void call() throws IOException {
+ log.flush();
+ offsetIndex().flush();
+ timeIndex().flush();
+ txnIndex.flush();
+ return null;
+ }
+ });
+ } catch (Exception e) {
+ if (e instanceof IOException)
+ throw (IOException) e;
+ else if (e instanceof RuntimeException)
+ throw (RuntimeException) e;
+ else
+ throw new IllegalStateException("Unexpected exception thrown: " + e, e);
+ }
+ }
+
+ /**
+ * Update the directory reference for the log and indices in this segment. This would typically be called after a
+ * directory is renamed.
+ */
+ void updateParentDir(File dir) {
+ log.updateParentDir(dir);
+ lazyOffsetIndex.updateParentDir(dir);
+ lazyTimeIndex.updateParentDir(dir);
+ txnIndex.updateParentDir(dir);
+ }
+
+ /**
+ * Change the suffix for the index and log files for this log segment
+ * IOException from this method should be handled by the caller
+ */
+ public void changeFileSuffixes(String oldSuffix, String newSuffix) throws IOException {
+ log.renameTo(new File(Utils.replaceSuffix(log.file().getPath(), oldSuffix, newSuffix)));
+ lazyOffsetIndex.renameTo(new File(Utils.replaceSuffix(offsetIndexFile().getPath(), oldSuffix, newSuffix)));
+ lazyTimeIndex.renameTo(new File(Utils.replaceSuffix(timeIndexFile().getPath(), oldSuffix, newSuffix)));
+ txnIndex.renameTo(new File(Utils.replaceSuffix(txnIndex.file().getPath(), oldSuffix, newSuffix)));
+ }
+
+ public boolean hasSuffix(String suffix) {
+ return log.file().getName().endsWith(suffix) &&
+ offsetIndexFile().getName().endsWith(suffix) &&
+ timeIndexFile().getName().endsWith(suffix) &&
+ txnIndex.file().getName().endsWith(suffix);
+ }
+
+ /**
+ * Append the largest time index entry to the time index and trim the log and indexes.
+ *
+ * The time index entry appended will be used to decide when to delete the segment.
+ */
+ public void onBecomeInactiveSegment() throws IOException {
+ timeIndex().maybeAppend(maxTimestampSoFar(), offsetOfMaxTimestampSoFar(), true);
+ offsetIndex().trimToValidSize();
+ timeIndex().trimToValidSize();
+ log.trim();
+ }
+
+ /**
+ * If not previously loaded,
+ * load the timestamp of the first message into memory.
+ */
+ private void loadFirstBatchTimestamp() {
+ if (!rollingBasedTimestamp.isPresent()) {
+ Iterator<FileChannelRecordBatch> iter = log.batches().iterator();
+ if (iter.hasNext())
+ rollingBasedTimestamp = OptionalLong.of(iter.next().maxTimestamp());
+ }
+ }
+
+ /**
+ * The time this segment has waited to be rolled.
+ * If the first message batch has a timestamp we use its timestamp to determine when to roll a segment. A segment
+ * is rolled if the difference between the new batch's timestamp and the first batch's timestamp exceeds the
+ * segment rolling time.
+ * If the first batch does not have a timestamp, we use the wall clock time to determine when to roll a segment. A
+ * segment is rolled if the difference between the current wall clock time and the segment create time exceeds the
+ * segment rolling time.
+ */
+ public long timeWaitedForRoll(long now, long messageTimestamp) {
+ // Load the timestamp of the first message into memory
+ loadFirstBatchTimestamp();
+ long ts = rollingBasedTimestamp.orElse(-1L);
+ if (ts >= 0)
+ return messageTimestamp - ts;
+ return now - created;
+ }
+
+ /**
+ * @return the first batch timestamp if the timestamp is available. Otherwise return Long.MaxValue
+ */
+ long getFirstBatchTimestamp() {
+ loadFirstBatchTimestamp();
+ OptionalLong timestamp = rollingBasedTimestamp;
+ if (timestamp.isPresent() && timestamp.getAsLong() >= 0)
+ return timestamp.getAsLong();
+ return Long.MAX_VALUE;
+ }
+
+ /**
+ * Search the message offset based on timestamp and offset.
+ *
+ * This method returns an option of TimestampOffset. The returned value is determined using the following ordered list of rules:
+ *
+ * - If all the messages in the segment have smaller offsets, return None
+ * - If all the messages in the segment have smaller timestamps, return None
+ * - If all the messages in the segment have larger timestamps, or no message in the segment has a timestamp
+ * the returned the offset will be max(the base offset of the segment, startingOffset) and the timestamp will be Message.NoTimestamp.
+ * - Otherwise, return an option of TimestampOffset. The offset is the offset of the first message whose timestamp
+ * is greater than or equals to the target timestamp and whose offset is greater than or equals to the startingOffset.
+ *
+ * This method only returns None when 1) all messages' offset < startOffing or 2) the log is not empty but we did not
+ * see any message when scanning the log from the indexed position. The latter could happen if the log is truncated
+ * after we get the indexed position but before we scan the log from there. In this case we simply return None and the
+ * caller will need to check on the truncated log and maybe retry or even do the search on another log segment.
+ *
+ * @param timestampMs The timestamp to search for.
+ * @param startingOffset The starting offset to search.
+ * @return the timestamp and offset of the first message that meets the requirements. None will be returned if there is no such message.
+ */
+ public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long timestampMs, long startingOffset) throws IOException {
+ // Get the index entry with a timestamp less than or equal to the target timestamp
+ TimestampOffset timestampOffset = timeIndex().lookup(timestampMs);
+ int position = offsetIndex().lookup(Math.max(timestampOffset.offset, startingOffset)).position;
+
+ // Search the timestamp
+ return Optional.ofNullable(log.searchForTimestamp(timestampMs, position, startingOffset));
+ }
+
+ /**
+ * Close this log segment
+ */
+ @Override
+ public void close() throws IOException {
+ if (maxTimestampAndOffsetSoFar != TimestampOffset.UNKNOWN)
+ Utils.swallow(LOGGER, Level.WARN, "maybeAppend", () -> timeIndex().maybeAppend(maxTimestampSoFar(), offsetOfMaxTimestampSoFar(), true));
+ Utils.closeQuietly(lazyOffsetIndex, "offsetIndex", LOGGER);
+ Utils.closeQuietly(lazyTimeIndex, "timeIndex", LOGGER);
+ Utils.closeQuietly(log, "log", LOGGER);
+ Utils.closeQuietly(txnIndex, "txnIndex", LOGGER);
+ }
+
+ /**
+ * Close file handlers used by the log segment but don't write to disk. This is used when the disk may have failed
+ */
+ void closeHandlers() {
+ Utils.swallow(LOGGER, Level.WARN, "offsetIndex", () -> lazyOffsetIndex.closeHandler());
+ Utils.swallow(LOGGER, Level.WARN, "timeIndex", () -> lazyTimeIndex.closeHandler());
+ Utils.swallow(LOGGER, Level.WARN, "log", () -> log.closeHandlers());
+ Utils.closeQuietly(txnIndex, "txnIndex", LOGGER);
+ }
+
+ /**
+ * Delete this log segment from the filesystem.
+ */
+ public void deleteIfExists() throws IOException {
+ try {
+ Utils.tryAll(asList(
+ () -> deleteTypeIfExists(() -> log.deleteIfExists(), "log", log.file(), true),
+ () -> deleteTypeIfExists(() -> lazyOffsetIndex.deleteIfExists(), "offset index", offsetIndexFile(), true),
+ () -> deleteTypeIfExists(() -> lazyTimeIndex.deleteIfExists(), "time index", timeIndexFile(), true),
+ () -> deleteTypeIfExists(() -> txnIndex.deleteIfExists(), "transaction index", txnIndex.file(), false)));
+ } catch (Throwable t) {
+ if (t instanceof IOException)
+ throw (IOException) t;
+ if (t instanceof Error)
+ throw (Error) t;
+ if (t instanceof RuntimeException)
+ throw (RuntimeException) t;
+ throw new IllegalStateException("Unexpected exception: " + t.getMessage(), t);
+ }
+ }
+
+ // Helper method for `deleteIfExists()`
+ private Void deleteTypeIfExists(StorageAction<Boolean, IOException> delete, String fileType, File file, boolean logIfMissing) throws IOException {
+ try {
+ if (delete.execute())
+ LOGGER.info("Deleted {} {}.", fileType, file.getAbsolutePath());
+ else if (logIfMissing)
+ LOGGER.info("Failed to delete {} {} because it does not exist.", fileType, file.getAbsolutePath());
+ return null;
+ } catch (IOException e) {
+ throw new IOException("Delete of " + fileType + " " + file.getAbsolutePath() + " failed.", e);
+ }
+ }
+
+ // Visible for testing
+ public boolean deleted() {
+ return !log.file().exists() && !offsetIndexFile().exists() && !timeIndexFile().exists() && !txnIndex.file().exists();
+ }
+
+
+ /**
+ * The last modified time of this log segment as a unix time stamp
+ */
+ public long lastModified() {
+ return log.file().lastModified();
+ }
+
+ /**
+ * The largest timestamp this segment contains, if maxTimestampSoFar >= 0, otherwise None.
+ */
+ public OptionalLong largestRecordTimestamp() throws IOException {
+ long maxTimestampSoFar = maxTimestampSoFar();
+ if (maxTimestampSoFar >= 0)
+ return OptionalLong.of(maxTimestampSoFar);
+ return OptionalLong.empty();
+ }
+
+ /**
+ * The largest timestamp this segment contains.
+ */
+ public long largestTimestamp() throws IOException {
+ long maxTimestampSoFar = maxTimestampSoFar();
+ if (maxTimestampSoFar >= 0)
+ return maxTimestampSoFar;
+ return lastModified();
+ }
+
+ /**
+ * Change the last modified time for this log segment
+ */
+ public void setLastModified(long ms) throws IOException {
+ FileTime fileTime = FileTime.fromMillis(ms);
+ Files.setLastModifiedTime(log.file().toPath(), fileTime);
+ Files.setLastModifiedTime(offsetIndexFile().toPath(), fileTime);
+ Files.setLastModifiedTime(timeIndexFile().toPath(), fileTime);
+ }
+
+ public static LogSegment open(File dir, long baseOffset, LogConfig config, Time time, int initFileSize, boolean preallocate) throws IOException {
+ return open(dir, baseOffset, config, time, false, initFileSize, preallocate, "");
+ }
+
+ public static LogSegment open(File dir, long baseOffset, LogConfig config, Time time, boolean fileAlreadyExists,
+ int initFileSize, boolean preallocate, String fileSuffix) throws IOException {
+ int maxIndexSize = config.maxIndexSize;
+ return new LogSegment(
+ FileRecords.open(LogFileUtils.logFile(dir, baseOffset, fileSuffix), fileAlreadyExists, initFileSize, preallocate),
+ LazyIndex.forOffset(LogFileUtils.offsetIndexFile(dir, baseOffset, fileSuffix), baseOffset, maxIndexSize),
+ LazyIndex.forTime(LogFileUtils.timeIndexFile(dir, baseOffset, fileSuffix), baseOffset, maxIndexSize),
+ new TransactionIndex(baseOffset, LogFileUtils.transactionIndexFile(dir, baseOffset, fileSuffix)),
+ baseOffset,
+ config.indexInterval,
+ config.randomSegmentJitter(),
+ time);
+ }
+
+ public static void deleteIfExists(File dir, long baseOffset, String fileSuffix) throws IOException {
+ deleteFileIfExists(LogFileUtils.offsetIndexFile(dir, baseOffset, fileSuffix));
+ deleteFileIfExists(LogFileUtils.timeIndexFile(dir, baseOffset, fileSuffix));
+ deleteFileIfExists(LogFileUtils.transactionIndexFile(dir, baseOffset, fileSuffix));
+ deleteFileIfExists(LogFileUtils.logFile(dir, baseOffset, fileSuffix));
+ }
+
+ private static boolean deleteFileIfExists(File file) throws IOException {
+ return Files.deleteIfExists(file.toPath());
+ }
+
+}
diff --git a/core/src/main/scala/kafka/common/LogSegmentOffsetOverflowException.scala b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegmentOffsetOverflowException.java
similarity index 64%
rename from core/src/main/scala/kafka/common/LogSegmentOffsetOverflowException.scala
rename to storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegmentOffsetOverflowException.java
index 2de5906109a..2312fa046de 100644
--- a/core/src/main/scala/kafka/common/LogSegmentOffsetOverflowException.scala
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegmentOffsetOverflowException.java
@@ -1,10 +1,10 @@
-/**
+/*
* Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
+ * 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
+ * the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
@@ -14,10 +14,9 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.kafka.storage.internals.log;
-package kafka.common
-
-import kafka.log.LogSegment
+import org.apache.kafka.common.KafkaException;
/**
* Indicates that the log segment contains one or more messages that overflow the offset (and / or time) index. This is
@@ -25,6 +24,13 @@ import kafka.log.LogSegment
* KAFKA-5413. With KAFKA-6264, we have the ability to split such log segments into multiple log segments such that we
* do not have any segments with offset overflow.
*/
-class LogSegmentOffsetOverflowException(val segment: LogSegment, val offset: Long)
- extends org.apache.kafka.common.KafkaException(s"Detected offset overflow at offset $offset in segment $segment") {
+public class LogSegmentOffsetOverflowException extends KafkaException {
+ public final LogSegment segment;
+ public final long offset;
+
+ public LogSegmentOffsetOverflowException(LogSegment segment, long offset) {
+ super("Detected offset overflow at offset " + offset + " in segment " + segment);
+ this.segment = segment;
+ this.offset = offset;
+ }
}
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegments.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegments.java
new file mode 100644
index 00000000000..c9af4b2227a
--- /dev/null
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegments.java
@@ -0,0 +1,355 @@
+/*
+ * 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.storage.internals.log;
+
+import org.apache.kafka.common.TopicPartition;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.concurrent.ConcurrentNavigableMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+/**
+ * This class encapsulates a thread-safe navigable map of LogSegment instances and provides the
+ * required read and write behavior on the map.
+ */
+public class LogSegments {
+
+ private final TopicPartition topicPartition;
+ /* the segments of the log with key being LogSegment base offset and value being a LogSegment */
+ private final ConcurrentNavigableMap<Long, LogSegment> segments = new ConcurrentSkipListMap<>();
+
+ /**
+ * Create new instance.
+ *
+ * @param topicPartition the TopicPartition associated with the segments
+ * (useful for logging purposes)
+ */
+ public LogSegments(TopicPartition topicPartition) {
+ this.topicPartition = topicPartition;
+ }
+
+ /**
+ * Return true if the segments are empty, false otherwise.
+ *
+ * This method is thread-safe.
+ */
+ public boolean isEmpty() {
+ return segments.isEmpty();
+ }
+
+ /**
+ * Return true if the segments are non-empty, false otherwise.
+ *
+ * This method is thread-safe.
+ */
+ public boolean nonEmpty() {
+ return !isEmpty();
+ }
+
+ /**
+ * Add the given segment, or replace an existing entry.
+ *
+ * This method is thread-safe.
+ *
+ * @param segment the segment to add
+ */
+ public LogSegment add(LogSegment segment) {
+ return this.segments.put(segment.baseOffset(), segment);
+ }
+
+ /**
+ * Remove the segment at the provided offset.
+ *
+ * This method is thread-safe.
+ *
+ * @param offset the offset to be removed
+ */
+ public void remove(long offset) {
+ segments.remove(offset);
+ }
+
+ /**
+ * Clears all entries.
+ *
+ * This method is thread-safe.
+ */
+ public void clear() {
+ segments.clear();
+ }
+
+ /**
+ * Close all segments.
+ */
+ public void close() throws IOException {
+ for (LogSegment s : values())
+ s.close();
+ }
+
+ /**
+ * Close the handlers for all segments.
+ */
+ public void closeHandlers() {
+ for (LogSegment s : values())
+ s.closeHandlers();
+ }
+
+ /**
+ * Update the directory reference for the log and indices of all segments.
+ *
+ * @param dir the renamed directory
+ */
+ public void updateParentDir(File dir) {
+ for (LogSegment s : values())
+ s.updateParentDir(dir);
+ }
+
+ /**
+ * Take care! this is an O(n) operation, where n is the number of segments.
+ *
+ * This method is thread-safe.
+ *
+ * @return The number of segments.
+ *
+ */
+ public int numberOfSegments() {
+ return segments.size();
+ }
+
+ /**
+ * @return the base offsets of all segments
+ */
+ public Collection<Long> baseOffsets() {
+ return values().stream().map(s -> s.baseOffset()).collect(Collectors.toList());
+ }
+
+ /**
+ * Return true if a segment exists at the provided offset, false otherwise.
+ *
+ * This method is thread-safe.
+ *
+ * @param offset the segment to be checked
+ */
+ public boolean contains(long offset) {
+ return segments.containsKey(offset);
+ }
+
+ /**
+ * Retrieves a segment at the specified offset.
+ *
+ * This method is thread-safe.
+ *
+ * @param offset the segment to be retrieved
+ *
+ * @return the segment if it exists, otherwise None.
+ */
+ public Optional<LogSegment> get(long offset) {
+ return Optional.ofNullable(segments.get(offset));
+ }
+
+ /**
+ * @return an iterator to the log segments ordered from oldest to newest.
+ */
+ public Collection<LogSegment> values() {
+ return segments.values();
+ }
+
+ /**
+ * @return An iterator to all segments beginning with the segment that includes "from" and ending
+ * with the segment that includes up to "to-1" or the end of the log (if to > end of log).
+ */
+ public Collection<LogSegment> values(long from, long to) {
+ if (from == to) {
+ // Handle non-segment-aligned empty sets
+ return Collections.emptyList();
+ } else if (to < from) {
+ throw new IllegalArgumentException("Invalid log segment range: requested segments in " + topicPartition +
+ " from offset " + from + " which is greater than limit offset " + to);
+ } else {
+ Long floor = segments.floorKey(from);
+ if (floor != null)
+ return segments.subMap(floor, to).values();
+ return segments.headMap(to).values();
+ }
+ }
+
+ public Collection<LogSegment> nonActiveLogSegmentsFrom(long from) {
+ LogSegment activeSegment = lastSegment().get();
+ if (from > activeSegment.baseOffset())
+ return Collections.emptyList();
+ else
+ return values(from, activeSegment.baseOffset());
+ }
+
+ /**
+ * Return the entry associated with the greatest offset less than or equal to the given offset,
+ * if it exists.
+ *
+ * This method is thread-safe.
+ */
+ private Optional<Map.Entry<Long, LogSegment>> floorEntry(long offset) {
+ return Optional.ofNullable(segments.floorEntry(offset));
+ }
+
+ /**
+ * Return the log segment with the greatest offset less than or equal to the given offset,
+ * if it exists.
+ *
+ * This method is thread-safe.
+ */
+ public Optional<LogSegment> floorSegment(long offset) {
+ return floorEntry(offset).map(e -> e.getValue());
+ }
+
+ /**
+ * Return the entry associated with the greatest offset strictly less than the given offset,
+ * if it exists.
+ *
+ * This method is thread-safe.
+ */
+ private Optional<Map.Entry<Long, LogSegment>> lowerEntry(long offset) {
+ return Optional.ofNullable(segments.lowerEntry(offset));
+ }
+
+ /**
+ * Return the log segment with the greatest offset strictly less than the given offset,
+ * if it exists.
+ *
+ * This method is thread-safe.
+ */
+ public Optional<LogSegment> lowerSegment(long offset) {
+ return lowerEntry(offset).map(e -> e.getValue());
+ }
+
+ /**
+ * Return the entry associated with the smallest offset strictly greater than the given offset,
+ * if it exists.
+ *
+ * This method is thread-safe.
+ */
+ public Optional<Map.Entry<Long, LogSegment>> higherEntry(long offset) {
+ return Optional.ofNullable(segments.higherEntry(offset));
+ }
+
+ /**
+ * Return the log segment with the smallest offset strictly greater than the given offset,
+ * if it exists.
+ *
+ * This method is thread-safe.
+ */
+ public Optional<LogSegment> higherSegment(long offset) {
+ return higherEntry(offset).map(e -> e.getValue());
+ }
+
+ /**
+ * Return the entry associated with the smallest offset, if it exists.
+ *
+ * This method is thread-safe.
+ */
+ public Optional<Map.Entry<Long, LogSegment>> firstEntry() {
+ return Optional.ofNullable(segments.firstEntry());
+ }
+
+ /**
+ * Return the log segment associated with the smallest offset, if it exists.
+ *
+ * This method is thread-safe.
+ */
+ public Optional<LogSegment> firstSegment() {
+ return firstEntry().map(s -> s.getValue());
+ }
+
+ /**
+ * @return the base offset of the log segment associated with the smallest offset, if it exists
+ */
+ public OptionalLong firstSegmentBaseOffset() {
+ Optional<LogSegment> first = firstSegment();
+ if (first.isPresent())
+ return OptionalLong.of(first.get().baseOffset());
+ return OptionalLong.empty();
+ }
+
+ /**
+ * Return the entry associated with the greatest offset, if it exists.
+ *
+ * This method is thread-safe.
+ */
+ public Optional<Map.Entry<Long, LogSegment>> lastEntry() {
+ return Optional.ofNullable(segments.lastEntry());
+ }
+
+ /**
+ * Return the log segment with the greatest offset, if it exists.
+ *
+ * This method is thread-safe.
+ */
+ public Optional<LogSegment> lastSegment() {
+ return lastEntry().map(e -> e.getValue());
+ }
+
+ /**
+ * @return an iterable with log segments ordered from lowest base offset to highest,
+ * each segment returned has a base offset strictly greater than the provided baseOffset.
+ */
+ public Collection<LogSegment> higherSegments(long baseOffset) {
+ Long higherOffset = segments.higherKey(baseOffset);
+ if (higherOffset != null)
+ return segments.tailMap(higherOffset, true).values();
+ return Collections.emptyList();
+ }
+
+ /**
+ * The active segment that is currently taking appends
+ */
+ public LogSegment activeSegment() {
+ return lastSegment().get();
+ }
+
+ public long sizeInBytes() {
+ return LogSegments.sizeInBytes(values());
+ }
+
+ /**
+ * Returns an Iterable containing segments matching the provided predicate.
+ *
+ * @param predicate the predicate to be used for filtering segments.
+ */
+ public Collection<LogSegment> filter(Predicate<LogSegment> predicate) {
+ return values().stream().filter(predicate).collect(Collectors.toList());
+ }
+
+ /**
+ * Calculate a log's size (in bytes) from the provided log segments.
+ *
+ * @param segments The log segments to calculate the size of
+ * @return Sum of the log segments' sizes (in bytes)
+ */
+ public static long sizeInBytes(Collection<LogSegment> segments) {
+ return segments.stream().mapToLong(s -> s.size()).sum();
+ }
+
+ public static Collection<Long> getFirstBatchTimestampForSegments(Collection<LogSegment> segments) {
+ return segments.stream().map(s -> s.getFirstBatchTimestamp()).collect(Collectors.toList());
+ }
+}