You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2025/12/24 10:26:57 UTC
(kafka) branch trunk updated: KAFKA-19752 Move UnifiedLogTest to storage module (1/N) (#20635)
This is an automated email from the ASF dual-hosted git repository.
chia7712 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 3a950f45ea5 KAFKA-19752 Move UnifiedLogTest to storage module (1/N) (#20635)
3a950f45ea5 is described below
commit 3a950f45ea5ecfb81e4aa7de84e388ea6c68b38a
Author: TaiJuWu <tj...@gmail.com>
AuthorDate: Wed Dec 24 18:26:51 2025 +0800
KAFKA-19752 Move UnifiedLogTest to storage module (1/N) (#20635)
This PR migrates several unit tests from UnifiedLogTest.scala to
UnifiedLogTest.java as part of the ongoing effort to convert core broker
logic and tests to Java.
The following test methods have been rewritten and moved:
- testDeleteOldSegments
- testLogDeletionAfterClose
- testLogDeletionAfterDeleteRecords
- shouldDeleteSizeBasedSegments
- shouldNotDeleteSizeBasedSegmentsWhenUnderRetentionSize
- shouldDeleteTimeBasedSegmentsReadyToBeDeleted
- shouldNotDeleteTimeBasedSegmentsWhenNoneReadyToBeDeleted
- shouldNotDeleteSegmentsWhenPolicyDoesNotIncludeDelete
- shouldDeleteSegmentsReadyToBeDeletedWhenCleanupPolicyIsCompactAndDelete
- shouldDeleteLocalLogSegmentsWhenPolicyIsEmptyWithSizeRetention
- shouldDeleteLocalLogSegmentsWhenPolicyIsEmptyWithMsRetention
- shouldDeleteStartOffsetBreachedSegmentsWhenPolicyDoesNotIncludeDelete
- shouldApplyEpochToMessageOnAppendIfLeader
- followerShouldSaveEpochInformationFromReplicatedMessagesToTheEpochCache
- shouldTruncateLeaderEpochsWhenDeletingSegments
- shouldUpdateOffsetForLeaderEpochsWhenDeletingSegments
- shouldTruncateLeaderEpochCheckpointFileWhenTruncatingLogt
- testFirstUnstableOffsetNoTransactionalData
- testFirstUnstableOffsetWithTransactionalData
New Utility Methods: To support the Java tests, new utility methods,
including LogConfigBuilder and transactional append helpers, were added
to LogTestUtils.java.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../test/scala/unit/kafka/log/UnifiedLogTest.scala | 519 +--------------
.../kafka/storage/internals/log/LogTestUtils.java | 122 ++++
.../storage/internals/log/UnifiedLogTest.java | 713 +++++++++++++++++++++
.../org/apache/kafka/common/test/TestUtils.java | 1 +
4 files changed, 862 insertions(+), 493 deletions(-)
diff --git a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
index d38e65e29f1..eb54319eb94 100755
--- a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
+++ b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
@@ -52,7 +52,6 @@ import org.junit.jupiter.params.provider.{EnumSource, ValueSource}
import org.mockito.ArgumentMatchers
import org.mockito.ArgumentMatchers.{any, anyLong}
import org.mockito.Mockito.{doAnswer, doThrow, spy}
-
import net.jqwik.api.AfterFailureMode
import net.jqwik.api.ForAll
import net.jqwik.api.Property
@@ -1016,7 +1015,7 @@ class UnifiedLogTest {
assertEquals(numProducerSnapshots, ProducerStateManager.listSnapshotFiles(logDir).size)
// Sleep to breach the retention period
mockTime.sleep(1000 * 60 + 1)
- log.deleteOldSegments()
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")
// Sleep to breach the file delete delay and run scheduled file deletion tasks
mockTime.sleep(1)
assertEquals(1, ProducerStateManager.listSnapshotFiles(logDir).size,
@@ -1065,7 +1064,7 @@ class UnifiedLogTest {
// Increment the log start offset to exclude the first two segments.
log.maybeIncrementLogStartOffset(log.logEndOffset - 1, LogStartOffsetIncrementReason.ClientRecordDeletion)
- log.deleteOldSegments()
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")
// Sleep to breach the file delete delay and run scheduled file deletion tasks
mockTime.sleep(1)
assertEquals(1, ProducerStateManager.listSnapshotFiles(logDir).size,
@@ -1103,7 +1102,8 @@ class UnifiedLogTest {
// Clean segments, this should delete everything except the active segment since there only
// exists the key "a".
cleaner.clean(new LogToClean(log, 0, log.logEndOffset, false))
- log.deleteOldSegments()
+ // There is no other key so we don't delete anything
+ assertEquals(0, log.deleteOldSegments())
// Sleep to breach the file delete delay and run scheduled file deletion tasks
mockTime.sleep(1)
assertEquals(log.logSegments.asScala.map(_.baseOffset).toSeq.sorted.drop(1), ProducerStateManager.listSnapshotFiles(logDir).asScala.map(_.offset).sorted,
@@ -1166,7 +1166,7 @@ class UnifiedLogTest {
assertEquals(util.Set.of(pid1, pid2), log.activeProducersWithLastSequence.keySet)
log.updateHighWatermark(log.logEndOffset)
- log.deleteOldSegments()
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")
// Producer state should not be removed when deleting log segment
assertEquals(2, log.logSegments.size)
@@ -1547,7 +1547,7 @@ class UnifiedLogTest {
log.updateHighWatermark(log.logEndOffset)
log.maybeIncrementLogStartOffset(2L, LogStartOffsetIncrementReason.ClientRecordDeletion)
- log.deleteOldSegments() // force retention to kick in so that the snapshot files are cleaned up.
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")// force retention to kick in so that the snapshot files are cleaned up.
mockTime.sleep(logConfig.fileDeleteDelayMs + 1000) // advance the clock so file deletion takes place
// Deleting records should not remove producer state but should delete snapshots after the file deletion delay.
@@ -2723,7 +2723,7 @@ class UnifiedLogTest {
val oldFiles = segments.map(_.log.file) ++ segments.map(_.offsetIndexFile)
log.updateHighWatermark(log.logEndOffset)
- log.deleteOldSegments()
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")
assertEquals(1, log.numberOfSegments, "Only one segment should remain.")
assertTrue(segments.forall(_.log.file.getName.endsWith(LogFileUtils.DELETED_FILE_SUFFIX)) &&
@@ -3015,476 +3015,8 @@ class UnifiedLogTest {
assertFalse(LogTestUtils.hasOffsetOverflow(log))
}
- @Test
- def testDeleteOldSegments(): Unit = {
- def createRecords = TestUtils.singletonRecords(value = "test".getBytes, timestamp = mockTime.milliseconds - 1000)
- val logConfig = LogTestUtils.createLogConfig(segmentBytes = createRecords.sizeInBytes * 5, segmentIndexBytes = 1000, retentionMs = 999)
- val log = createLog(logDir, logConfig)
-
- // append some messages to create some segments
- for (_ <- 0 until 100)
- log.appendAsLeader(createRecords, 0)
-
- log.assignEpochStartOffset(0, 40)
- log.assignEpochStartOffset(1, 90)
-
- // segments are not eligible for deletion if no high watermark has been set
- val numSegments = log.numberOfSegments
- log.deleteOldSegments()
- assertEquals(numSegments, log.numberOfSegments)
- assertEquals(0L, log.logStartOffset)
-
- // only segments with offset before the current high watermark are eligible for deletion
- for (hw <- 25 to 30) {
- log.updateHighWatermark(hw)
- log.deleteOldSegments()
- assertTrue(log.logStartOffset <= hw)
- 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)
- }
- }
- }
-
- // expire all segments
- log.updateHighWatermark(log.logEndOffset)
- log.deleteOldSegments()
- assertEquals(1, log.numberOfSegments, "The deleted segments should be gone.")
- assertEquals(1, epochCache(log).epochEntries.size, "Epoch entries should have gone.")
- assertEquals(new EpochEntry(1, 100), epochCache(log).epochEntries.get(0), "Epoch entry should be the latest epoch and the leo.")
-
- // append some messages to create some segments
- for (_ <- 0 until 100)
- log.appendAsLeader(createRecords, 0)
-
- log.delete()
- assertEquals(0, log.numberOfSegments, "The number of segments should be 0")
- assertEquals(0, log.deleteOldSegments(), "The number of deleted segments should be zero.")
- assertEquals(0, epochCache(log).epochEntries.size, "Epoch entries should have gone.")
- }
-
- @Test
- def testLogDeletionAfterClose(): Unit = {
- def createRecords = TestUtils.singletonRecords(value = "test".getBytes, timestamp = mockTime.milliseconds - 1000)
- val logConfig = LogTestUtils.createLogConfig(segmentBytes = createRecords.sizeInBytes * 5, segmentIndexBytes = 1000, retentionMs = 999)
- val log = createLog(logDir, logConfig)
-
- // append some messages to create some segments
- log.appendAsLeader(createRecords, 0)
-
- assertEquals(1, log.numberOfSegments, "The deleted segments should be gone.")
- assertEquals(1, epochCache(log).epochEntries.size, "Epoch entries should have gone.")
-
- log.close()
- log.delete()
- assertEquals(0, log.numberOfSegments, "The number of segments should be 0")
- assertEquals(0, epochCache(log).epochEntries.size, "Epoch entries should have gone.")
- }
-
- @Test
- def testLogDeletionAfterDeleteRecords(): Unit = {
- def createRecords = TestUtils.singletonRecords("test".getBytes)
- val logConfig = LogTestUtils.createLogConfig(segmentBytes = createRecords.sizeInBytes * 5)
- val log = createLog(logDir, logConfig)
-
- for (_ <- 0 until 15)
- log.appendAsLeader(createRecords, 0)
- assertEquals(3, log.numberOfSegments, "should have 3 segments")
- assertEquals(log.logStartOffset, 0)
- log.updateHighWatermark(log.logEndOffset)
-
- log.maybeIncrementLogStartOffset(1, LogStartOffsetIncrementReason.ClientRecordDeletion)
- log.deleteOldSegments()
- assertEquals(3, log.numberOfSegments, "should have 3 segments")
- assertEquals(log.logStartOffset, 1)
-
- log.maybeIncrementLogStartOffset(6, LogStartOffsetIncrementReason.ClientRecordDeletion)
- log.deleteOldSegments()
- assertEquals(2, log.numberOfSegments, "should have 2 segments")
- assertEquals(log.logStartOffset, 6)
-
- log.maybeIncrementLogStartOffset(15, LogStartOffsetIncrementReason.ClientRecordDeletion)
- log.deleteOldSegments()
- assertEquals(1, log.numberOfSegments, "should have 1 segments")
- assertEquals(log.logStartOffset, 15)
- }
-
def epochCache(log: UnifiedLog): LeaderEpochFileCache = log.leaderEpochCache
- @Test
- def shouldDeleteSizeBasedSegments(): Unit = {
- def createRecords = TestUtils.singletonRecords("test".getBytes)
- val logConfig = LogTestUtils.createLogConfig(segmentBytes = createRecords.sizeInBytes * 5, retentionBytes = createRecords.sizeInBytes * 10)
- val log = createLog(logDir, logConfig)
-
- // append some messages to create some segments
- for (_ <- 0 until 15)
- log.appendAsLeader(createRecords, 0)
-
- log.updateHighWatermark(log.logEndOffset)
- log.deleteOldSegments()
- assertEquals(2,log.numberOfSegments, "should have 2 segments")
- }
-
- @Test
- def shouldNotDeleteSizeBasedSegmentsWhenUnderRetentionSize(): Unit = {
- def createRecords = TestUtils.singletonRecords("test".getBytes)
- val logConfig = LogTestUtils.createLogConfig(segmentBytes = createRecords.sizeInBytes * 5, retentionBytes = createRecords.sizeInBytes * 15)
- val log = createLog(logDir, logConfig)
-
- // append some messages to create some segments
- for (_ <- 0 until 15)
- log.appendAsLeader(createRecords, 0)
-
- log.updateHighWatermark(log.logEndOffset)
- log.deleteOldSegments()
- assertEquals(3,log.numberOfSegments, "should have 3 segments")
- }
-
- @Test
- def shouldDeleteTimeBasedSegmentsReadyToBeDeleted(): Unit = {
- def createRecords = TestUtils.singletonRecords("test".getBytes, timestamp = 10)
- val logConfig = LogTestUtils.createLogConfig(segmentBytes = createRecords.sizeInBytes * 5, retentionMs = 10000)
- val log = createLog(logDir, logConfig)
-
- // append some messages to create some segments
- for (_ <- 0 until 15)
- log.appendAsLeader(createRecords, 0)
-
- log.updateHighWatermark(log.logEndOffset)
- log.deleteOldSegments()
- assertEquals(1, log.numberOfSegments, "There should be 1 segment remaining")
- }
-
- @Test
- def shouldNotDeleteTimeBasedSegmentsWhenNoneReadyToBeDeleted(): Unit = {
- def createRecords = TestUtils.singletonRecords("test".getBytes, timestamp = mockTime.milliseconds)
- val logConfig = LogTestUtils.createLogConfig(segmentBytes = createRecords.sizeInBytes * 5, retentionMs = 10000000)
- val log = createLog(logDir, logConfig)
-
- // append some messages to create some segments
- for (_ <- 0 until 15)
- log.appendAsLeader(createRecords, 0)
-
- log.updateHighWatermark(log.logEndOffset)
- log.deleteOldSegments()
- assertEquals(3, log.numberOfSegments, "There should be 3 segments remaining")
- }
-
- @Test
- def shouldNotDeleteSegmentsWhenPolicyDoesNotIncludeDelete(): Unit = {
- def createRecords = TestUtils.singletonRecords("test".getBytes, key = "test".getBytes(), timestamp = 10L)
- val logConfig = LogTestUtils.createLogConfig(segmentBytes = createRecords.sizeInBytes * 5, retentionMs = 10000, cleanupPolicy = "compact")
- val log = createLog(logDir, logConfig)
-
- // append some messages to create some segments
- for (_ <- 0 until 15)
- log.appendAsLeader(createRecords, 0)
-
- // mark the oldest segment as older the retention.ms
- log.logSegments.asScala.head.setLastModified(mockTime.milliseconds - 20000)
-
- val segments = log.numberOfSegments
- log.updateHighWatermark(log.logEndOffset)
- log.deleteOldSegments()
- assertEquals(segments, log.numberOfSegments, "There should be 3 segments remaining")
- }
-
- @Test
- def shouldDeleteLocalLogSegmentsWhenPolicyIsEmptyWithSizeRetention(): Unit = {
- def createRecords = TestUtils.singletonRecords("test".getBytes, key = "test".getBytes(), timestamp = 10L)
- val recordSize = createRecords.sizeInBytes
- val logConfig = LogTestUtils.createLogConfig(
- segmentBytes = recordSize * 2,
- localRetentionBytes = recordSize / 2,
- cleanupPolicy = "",
- remoteLogStorageEnable = true
- )
- val log = createLog(logDir, logConfig, remoteStorageSystemEnable = true)
-
- for (_ <- 0 until 10)
- log.appendAsLeader(createRecords, 0)
-
- val segmentsBefore = log.numberOfSegments
- log.updateHighWatermark(log.logEndOffset)
- log.updateHighestOffsetInRemoteStorage(log.logEndOffset - 1)
- val deleteOldSegments = log.deleteOldSegments()
-
- assertTrue(log.numberOfSegments < segmentsBefore, "Some segments should be deleted due to size retention")
- assertTrue(deleteOldSegments > 0, "At least one segment should be deleted")
- }
-
- @Test
- def shouldDeleteLocalLogSegmentsWhenPolicyIsEmptyWithMsRetention(): Unit = {
- val oldTimestamp = mockTime.milliseconds - 20000
- def oldRecords = TestUtils.singletonRecords("test".getBytes, key = "test".getBytes(), timestamp = oldTimestamp)
- val recordSize = oldRecords.sizeInBytes
- val logConfig = LogTestUtils.createLogConfig(
- segmentBytes = recordSize * 2,
- localRetentionMs = 5000,
- cleanupPolicy = "",
- remoteLogStorageEnable = true
- )
- val log = createLog(logDir, logConfig, remoteStorageSystemEnable = true)
-
- for (_ <- 0 until 10)
- log.appendAsLeader(oldRecords, 0)
-
- def newRecords = TestUtils.singletonRecords("test".getBytes, key = "test".getBytes(), timestamp = mockTime.milliseconds)
- for (_ <- 0 until 5)
- log.appendAsLeader(newRecords, 0)
-
- val segmentsBefore = log.numberOfSegments
-
- log.updateHighWatermark(log.logEndOffset)
- log.updateHighestOffsetInRemoteStorage(log.logEndOffset - 1)
- val deleteOldSegments = log.deleteOldSegments()
-
- assertTrue(log.numberOfSegments < segmentsBefore, "Some segments should be deleted due to time retention")
- assertTrue(deleteOldSegments > 0, "At least one segment should be deleted")
- }
-
- @Test
- def shouldDeleteSegmentsReadyToBeDeletedWhenCleanupPolicyIsCompactAndDelete(): Unit = {
- def createRecords = TestUtils.singletonRecords("test".getBytes, key = "test".getBytes, timestamp = 10L)
- val logConfig = LogTestUtils.createLogConfig(segmentBytes = createRecords.sizeInBytes * 5, retentionMs = 10000, cleanupPolicy = "compact,delete")
- val log = createLog(logDir, logConfig)
-
- // append some messages to create some segments
- for (_ <- 0 until 15)
- log.appendAsLeader(createRecords, 0)
-
- log.updateHighWatermark(log.logEndOffset)
- log.deleteOldSegments()
- assertEquals(1, log.numberOfSegments, "There should be 1 segment remaining")
- }
-
- @Test
- def shouldDeleteStartOffsetBreachedSegmentsWhenPolicyDoesNotIncludeDelete(): Unit = {
- def createRecords = TestUtils.singletonRecords("test".getBytes, key = "test".getBytes, timestamp = 10L)
- val recordsPerSegment = 5
- val logConfig = LogTestUtils.createLogConfig(segmentBytes = createRecords.sizeInBytes * recordsPerSegment, retentionMs = 10000, cleanupPolicy = "compact")
- val log = createLog(logDir, logConfig, brokerTopicStats)
-
- // append some messages to create some segments
- for (_ <- 0 until 15)
- log.appendAsLeader(createRecords, 0)
-
- // Three segments should be created
- assertEquals(3, log.logSegments.asScala.count(_ => true))
- log.updateHighWatermark(log.logEndOffset)
- log.maybeIncrementLogStartOffset(recordsPerSegment, LogStartOffsetIncrementReason.ClientRecordDeletion)
-
- // The first segment, which is entirely before the log start offset, should be deleted
- // Of the remaining the segments, the first can overlap the log start offset and the rest must have a base offset
- // greater than the start offset
- log.updateHighWatermark(log.logEndOffset)
- log.deleteOldSegments()
- assertEquals(2, log.numberOfSegments, "There should be 2 segments remaining")
- assertTrue(log.logSegments.asScala.head.baseOffset <= log.logStartOffset)
- assertTrue(log.logSegments.asScala.tail.forall(s => s.baseOffset > log.logStartOffset))
- }
-
- @Test
- def shouldApplyEpochToMessageOnAppendIfLeader(): Unit = {
- val records = (0 until 50).toArray.map(id => new SimpleRecord(id.toString.getBytes))
-
- //Given this partition is on leader epoch 72
- val epoch = 72
- val log = createLog(logDir, new LogConfig(new Properties))
- log.assignEpochStartOffset(epoch, records.length)
-
- //When appending messages as a leader (i.e. assignOffsets = true)
- for (record <- records)
- log.appendAsLeader(
- MemoryRecords.withRecords(Compression.NONE, record),
- epoch
- )
-
- //Then leader epoch should be set on messages
- for (i <- records.indices) {
- val read = LogTestUtils.readLog(log, i, 1).records.batches.iterator.next()
- assertEquals(72, read.partitionLeaderEpoch, "Should have set leader epoch")
- }
- }
-
- @Test
- def followerShouldSaveEpochInformationFromReplicatedMessagesToTheEpochCache(): Unit = {
- val messageIds = (0 until 50).toArray
- val records = messageIds.map(id => new SimpleRecord(id.toString.getBytes))
-
- //Given each message has an offset & epoch, as msgs from leader would
- def recordsForEpoch(i: Int): MemoryRecords = {
- val recs = MemoryRecords.withRecords(messageIds(i), Compression.NONE, records(i))
- recs.batches.forEach{record =>
- record.setPartitionLeaderEpoch(42)
- record.setLastOffset(i)
- }
- recs
- }
-
- val log = createLog(logDir, new LogConfig(new Properties))
-
- //When appending as follower (assignOffsets = false)
- for (i <- records.indices)
- log.appendAsFollower(recordsForEpoch(i), i)
-
- assertEquals(Optional.of(42), log.latestEpoch)
- }
-
- @Test
- def shouldTruncateLeaderEpochsWhenDeletingSegments(): Unit = {
- def createRecords = TestUtils.singletonRecords("test".getBytes)
- val logConfig = LogTestUtils.createLogConfig(segmentBytes = createRecords.sizeInBytes * 5, retentionBytes = createRecords.sizeInBytes * 10)
- val log = createLog(logDir, logConfig)
- val cache = epochCache(log)
-
- // Given three segments of 5 messages each
- for (_ <- 0 until 15) {
- log.appendAsLeader(createRecords, 0)
- }
-
- //Given epochs
- cache.assign(0, 0)
- cache.assign(1, 5)
- cache.assign(2, 10)
-
- //When first segment is removed
- log.updateHighWatermark(log.logEndOffset)
- log.deleteOldSegments()
-
- //The oldest epoch entry should have been removed
- assertEquals(util.List.of(new EpochEntry(1, 5), new EpochEntry(2, 10)), cache.epochEntries)
- }
-
- @Test
- def shouldUpdateOffsetForLeaderEpochsWhenDeletingSegments(): Unit = {
- def createRecords = TestUtils.singletonRecords("test".getBytes)
- val logConfig = LogTestUtils.createLogConfig(segmentBytes = createRecords.sizeInBytes * 5, retentionBytes = createRecords.sizeInBytes * 10)
- val log = createLog(logDir, logConfig)
- val cache = epochCache(log)
-
- // Given three segments of 5 messages each
- for (_ <- 0 until 15) {
- log.appendAsLeader(createRecords, 0)
- }
-
- //Given epochs
- cache.assign(0, 0)
- cache.assign(1, 7)
- cache.assign(2, 10)
-
- //When first segment removed (up to offset 5)
- log.updateHighWatermark(log.logEndOffset)
- log.deleteOldSegments()
-
- //The first entry should have gone from (0,0) => (0,5)
- assertEquals(util.List.of(new EpochEntry(0, 5), new EpochEntry(1, 7), new EpochEntry(2, 10)), cache.epochEntries)
- }
-
- @Test
- def shouldTruncateLeaderEpochCheckpointFileWhenTruncatingLog(): Unit = {
- def createRecords(startOffset: Long, epoch: Int): MemoryRecords = {
- TestUtils.records(Seq(new SimpleRecord("value".getBytes)),
- baseOffset = startOffset, partitionLeaderEpoch = epoch)
- }
-
- val logConfig = LogTestUtils.createLogConfig(segmentBytes = 10 * createRecords(0, 0).sizeInBytes)
- val log = createLog(logDir, logConfig)
- val cache = epochCache(log)
-
- def append(epoch: Int, startOffset: Long, count: Int): Unit = {
- for (i <- 0 until count)
- log.appendAsFollower(createRecords(startOffset + i, epoch), epoch)
- }
-
- //Given 2 segments, 10 messages per segment
- append(epoch = 0, startOffset = 0, count = 10)
- append(epoch = 1, startOffset = 10, count = 6)
- append(epoch = 2, startOffset = 16, count = 4)
-
- assertEquals(2, log.numberOfSegments)
- assertEquals(20, log.logEndOffset)
-
- //When truncate to LEO (no op)
- log.truncateTo(log.logEndOffset)
-
- //Then no change
- assertEquals(3, cache.epochEntries.size)
-
- //When truncate
- log.truncateTo(11)
-
- //Then no change
- assertEquals(2, cache.epochEntries.size)
-
- //When truncate
- log.truncateTo(10)
-
- //Then
- assertEquals(1, cache.epochEntries.size)
-
- //When truncate all
- log.truncateTo(0)
-
- //Then
- assertEquals(0, cache.epochEntries.size)
- }
-
- @Test
- def testFirstUnstableOffsetNoTransactionalData(): Unit = {
- val logConfig = LogTestUtils.createLogConfig(segmentBytes = 1024 * 1024 * 5)
- val log = createLog(logDir, logConfig)
-
- val records = MemoryRecords.withRecords(Compression.NONE,
- new SimpleRecord("foo".getBytes),
- new SimpleRecord("bar".getBytes),
- new SimpleRecord("baz".getBytes))
-
- log.appendAsLeader(records, 0)
- assertEquals(Optional.empty, log.firstUnstableOffset)
- }
-
- @Test
- def testFirstUnstableOffsetWithTransactionalData(): Unit = {
- val logConfig = LogTestUtils.createLogConfig(segmentBytes = 1024 * 1024 * 5)
- val log = createLog(logDir, logConfig)
-
- val pid = 137L
- val epoch = 5.toShort
- var seq = 0
-
- // add some transactional records
- val records = MemoryRecords.withTransactionalRecords(Compression.NONE, pid, epoch, seq,
- new SimpleRecord("foo".getBytes),
- new SimpleRecord("bar".getBytes),
- new SimpleRecord("baz".getBytes))
-
- val firstAppendInfo = log.appendAsLeader(records, 0)
- assertEquals(Optional.of(firstAppendInfo.firstOffset), log.firstUnstableOffset)
-
- // add more transactional records
- seq += 3
- log.appendAsLeader(MemoryRecords.withTransactionalRecords(Compression.NONE, pid, epoch, seq,
- new SimpleRecord("blah".getBytes)), 0)
-
- // LSO should not have changed
- assertEquals(Optional.of(firstAppendInfo.firstOffset), log.firstUnstableOffset)
-
- // now transaction is committed
- val commitAppendInfo = LogTestUtils.appendEndTxnMarkerAsLeader(log, pid, epoch, ControlRecordType.COMMIT,
- mockTime.milliseconds(), transactionVersion = TransactionVersion.TV_0.featureLevel())
-
- // first unstable offset is not updated until the high watermark is advanced
- assertEquals(Optional.of(firstAppendInfo.firstOffset), log.firstUnstableOffset)
- log.updateHighWatermark(commitAppendInfo.lastOffset + 1)
-
- // now there should be no first unstable offset
- assertEquals(Optional.empty, log.firstUnstableOffset)
- }
-
@Test
def testReadCommittedWithConcurrentHighWatermarkUpdates(): Unit = {
val logConfig = LogTestUtils.createLogConfig(segmentBytes = 1024 * 1024 * 5)
@@ -3535,6 +3067,7 @@ class UnifiedLogTest {
}
}
+
@Test
def testTransactionIndexUpdated(): Unit = {
val logConfig = LogTestUtils.createLogConfig(segmentBytes = 1024 * 1024 * 5)
@@ -3960,7 +3493,7 @@ class UnifiedLogTest {
log.updateHighWatermark(log.logEndOffset)
log.maybeIncrementLogStartOffset(8L, LogStartOffsetIncrementReason.ClientRecordDeletion)
log.updateHighWatermark(log.logEndOffset)
- log.deleteOldSegments()
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")
assertEquals(1, log.logSegments.size)
// the first unstable offset should be lower bounded by the log start offset
@@ -4160,7 +3693,7 @@ class UnifiedLogTest {
assertEquals(25L, initialHighWatermark)
val initialNumSegments = log.numberOfSegments
- log.deleteOldSegments()
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")
assertTrue(log.numberOfSegments < initialNumSegments)
assertTrue(log.logStartOffset <= initialHighWatermark)
}
@@ -4870,7 +4403,7 @@ class UnifiedLogTest {
mockTime.sleep(2)
// It should have rolled the active segment as they are eligible for deletion
- log.deleteOldSegments()
+ assertEquals(0, log.deleteOldSegments())
assertEquals(2, log.logSegments.size)
log.logSegments.asScala.zipWithIndex.foreach {
case (segment, idx) => assertEquals(idx, segment.baseOffset)
@@ -4878,7 +4411,7 @@ class UnifiedLogTest {
// Once rolled, the segment should be uploaded to remote storage and eligible for deletion
log.updateHighestOffsetInRemoteStorage(1)
- log.deleteOldSegments()
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")
assertEquals(1, log.logSegments.size)
assertEquals(1, log.logSegments.asScala.head.baseOffset())
assertEquals(1, log.localLogStartOffset())
@@ -4910,14 +4443,14 @@ class UnifiedLogTest {
// No segments are uploaded to remote storage, none of the local log segments should be eligible for deletion
log.updateHighestOffsetInRemoteStorage(-1L)
- log.deleteOldSegments()
+ assertEquals(0, log.deleteOldSegments())
mockTime.sleep(1)
assertEquals(2, log.logSegments.size)
assertFalse(log.isEmpty)
// Update the log-start-offset from 0 to 3, then the base segment should not be eligible for deletion
log.updateLogStartOffsetFromRemoteTier(3L)
- log.deleteOldSegments()
+ assertEquals(0, log.deleteOldSegments())
mockTime.sleep(1)
assertEquals(2, log.logSegments.size)
assertFalse(log.isEmpty)
@@ -4925,13 +4458,13 @@ class UnifiedLogTest {
// Update the log-start-offset from 3 to 4, then the base segment should be eligible for deletion now even
// if it is not uploaded to remote storage
log.updateLogStartOffsetFromRemoteTier(4L)
- log.deleteOldSegments()
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")
mockTime.sleep(1)
assertEquals(1, log.logSegments.size)
assertFalse(log.isEmpty)
log.updateLogStartOffsetFromRemoteTier(5L)
- log.deleteOldSegments()
+ assertEquals(0, log.deleteOldSegments())
mockTime.sleep(1)
assertEquals(1, log.logSegments.size)
assertTrue(log.isEmpty)
@@ -4954,7 +4487,7 @@ class UnifiedLogTest {
log.updateHighWatermark(log.logEndOffset)
// simulate calls to upload 2 segments to remote storage
log.updateHighestOffsetInRemoteStorage(1)
- log.deleteOldSegments()
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")
assertEquals(4, log.logSegments.size())
assertEquals(0, log.logStartOffset)
assertEquals(2, log.localLogStartOffset())
@@ -4979,7 +4512,7 @@ class UnifiedLogTest {
log.updateHighestOffsetInRemoteStorage(1)
mockTime.sleep(1001)
- log.deleteOldSegments()
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")
assertEquals(4, log.logSegments.size())
assertEquals(0, log.logStartOffset)
assertEquals(2, log.localLogStartOffset())
@@ -5002,7 +4535,7 @@ class UnifiedLogTest {
log.updateHighWatermark(log.logEndOffset)
// Should not delete local log because highest remote storage offset is -1 (default value)
- log.deleteOldSegments()
+ assertEquals(0, log.deleteOldSegments())
assertEquals(6, log.logSegments.size())
assertEquals(0, log.logStartOffset)
assertEquals(0, log.localLogStartOffset())
@@ -5010,7 +4543,7 @@ class UnifiedLogTest {
// simulate calls to upload 2 segments to remote storage
log.updateHighestOffsetInRemoteStorage(1)
- log.deleteOldSegments()
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")
assertEquals(4, log.logSegments.size())
assertEquals(0, log.logStartOffset)
assertEquals(2, log.localLogStartOffset())
@@ -5022,7 +4555,7 @@ class UnifiedLogTest {
// No local logs will be deleted even though local retention bytes is 1 because we'll adopt retention.ms/bytes
// when remote.log.copy.disable = true
- log.deleteOldSegments()
+ assertEquals(0, log.deleteOldSegments())
assertEquals(4, log.logSegments.size())
assertEquals(0, log.logStartOffset)
assertEquals(2, log.localLogStartOffset())
@@ -5042,7 +4575,7 @@ class UnifiedLogTest {
// try to delete local logs again, 2 segments will be deleted this time because we'll adopt retention.ms/bytes (retention.bytes = 5)
// when remote.log.copy.disable = true
- log.deleteOldSegments()
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")
assertEquals(5, log.logSegments.size())
assertEquals(4, log.logStartOffset)
assertEquals(4, log.localLogStartOffset())
@@ -5054,14 +4587,14 @@ class UnifiedLogTest {
// Should not delete any logs because no local logs expired using retention.ms = 1000
mockTime.sleep(10)
- log.deleteOldSegments()
+ assertEquals(0, log.deleteOldSegments())
assertEquals(5, log.logSegments.size())
assertEquals(4, log.logStartOffset)
assertEquals(4, log.localLogStartOffset())
// Should delete all logs because all of them are expired based on retentionMs = 1000
mockTime.sleep(1000)
- log.deleteOldSegments()
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")
assertEquals(1, log.logSegments.size())
assertEquals(9, log.logStartOffset)
assertEquals(9, log.localLogStartOffset())
@@ -5085,7 +4618,7 @@ class UnifiedLogTest {
// simulate calls to upload 3 segments to remote storage
log.updateHighestOffsetInRemoteStorage(30)
- log.deleteOldSegments()
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")
assertEquals(2, log.logSegments.size())
assertEquals(0, log.logStartOffset)
assertEquals(31, log.localLogStartOffset())
@@ -5109,7 +4642,7 @@ class UnifiedLogTest {
// simulate calls to upload 3 segments to remote storage
log.updateHighestOffsetInRemoteStorage(30)
- log.deleteOldSegments()
+ assertTrue(log.deleteOldSegments > 0, "At least one segment should be deleted")
assertEquals(2, log.logSegments.size())
assertEquals(0, log.logStartOffset)
assertEquals(31, log.localLogStartOffset())
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/LogTestUtils.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogTestUtils.java
index 50b666a0fb1..6a89b82043d 100644
--- a/storage/src/test/java/org/apache/kafka/storage/internals/log/LogTestUtils.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogTestUtils.java
@@ -16,11 +16,18 @@
*/
package org.apache.kafka.storage.internals.log;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.record.ControlRecordType;
+import org.apache.kafka.common.record.EndTransactionMarker;
import org.apache.kafka.common.record.FileRecords;
+import org.apache.kafka.common.record.MemoryRecords;
import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.server.common.RequestLocal;
import java.io.File;
import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
public class LogTestUtils {
public static LogSegment createSegment(long offset, File logDir, int indexIntervalBytes, Time time) throws IOException {
@@ -33,4 +40,119 @@ public class LogTestUtils {
// Create and return the LogSegment instance
return new LogSegment(ms, idx, timeIdx, txnIndex, offset, indexIntervalBytes, 0, time);
}
+
+
+ /**
+ * Append an end transaction marker (commit or abort) to the log as a leader.
+ *
+ * @param transactionVersion the transaction version (1 = TV1, 2 = TV2) etc. Must be explicitly specified.
+ * TV2 markers require strict epoch validation (markerEpoch > currentEpoch),
+ * while legacy markers use relaxed validation (markerEpoch >= currentEpoch).
+ */
+ public static LogAppendInfo appendEndTxnMarkerAsLeader(UnifiedLog log,
+ long producerId,
+ short producerEpoch,
+ ControlRecordType controlType,
+ long timestamp,
+ int coordinatorEpoch,
+ int leaderEpoch,
+ short transactionVersion) {
+ MemoryRecords records = endTxnRecords(controlType, producerId, producerEpoch, 0L, coordinatorEpoch, leaderEpoch, timestamp);
+
+ return log.appendAsLeader(records, leaderEpoch, AppendOrigin.COORDINATOR, RequestLocal.noCaching(), VerificationGuard.SENTINEL, transactionVersion);
+ }
+
+ public static MemoryRecords endTxnRecords(ControlRecordType controlRecordType,
+ long producerId,
+ short epoch,
+ long offset,
+ int coordinatorEpoch,
+ int partitionLeaderEpoch,
+ long timestamp) {
+ EndTransactionMarker marker = new EndTransactionMarker(controlRecordType, coordinatorEpoch);
+ return MemoryRecords.withEndTransactionMarker(offset, timestamp, partitionLeaderEpoch, producerId, epoch, marker);
+ }
+
+ public static class LogConfigBuilder {
+ private final Map<String, Object> configs = new HashMap<>();
+
+ public LogConfigBuilder segmentMs(long segmentMs) {
+ configs.put(TopicConfig.SEGMENT_MS_CONFIG, segmentMs);
+ return this;
+ }
+
+ public LogConfigBuilder segmentBytes(int segmentBytes) {
+ configs.put(LogConfig.INTERNAL_SEGMENT_BYTES_CONFIG, segmentBytes);
+ return this;
+ }
+
+ public LogConfigBuilder retentionMs(long retentionMs) {
+ configs.put(TopicConfig.RETENTION_MS_CONFIG, retentionMs);
+ return this;
+ }
+
+ public LogConfigBuilder localRetentionMs(long localRetentionMs) {
+ configs.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, localRetentionMs);
+ return this;
+ }
+
+ public LogConfigBuilder retentionBytes(long retentionBytes) {
+ configs.put(TopicConfig.RETENTION_BYTES_CONFIG, retentionBytes);
+ return this;
+ }
+
+ public LogConfigBuilder localRetentionBytes(long localRetentionBytes) {
+ configs.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, localRetentionBytes);
+ return this;
+ }
+
+ public LogConfigBuilder segmentJitterMs(long segmentJitterMs) {
+ configs.put(TopicConfig.SEGMENT_JITTER_MS_CONFIG, segmentJitterMs);
+ return this;
+ }
+
+ public LogConfigBuilder cleanupPolicy(String cleanupPolicy) {
+ configs.put(TopicConfig.CLEANUP_POLICY_CONFIG, cleanupPolicy);
+ return this;
+ }
+
+ public LogConfigBuilder maxMessageBytes(int maxMessageBytes) {
+ configs.put(TopicConfig.MAX_MESSAGE_BYTES_CONFIG, maxMessageBytes);
+ return this;
+ }
+
+ public LogConfigBuilder indexIntervalBytes(int indexIntervalBytes) {
+ configs.put(TopicConfig.INDEX_INTERVAL_BYTES_CONFIG, indexIntervalBytes);
+ return this;
+ }
+
+ public LogConfigBuilder segmentIndexBytes(int segmentIndexBytes) {
+ configs.put(TopicConfig.SEGMENT_INDEX_BYTES_CONFIG, segmentIndexBytes);
+ return this;
+ }
+
+ public LogConfigBuilder fileDeleteDelayMs(long fileDeleteDelayMs) {
+ configs.put(TopicConfig.FILE_DELETE_DELAY_MS_CONFIG, fileDeleteDelayMs);
+ return this;
+ }
+
+ public LogConfigBuilder remoteLogStorageEnable(boolean remoteLogStorageEnable) {
+ configs.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, remoteLogStorageEnable);
+ return this;
+ }
+
+ public LogConfigBuilder remoteLogCopyDisable(boolean remoteLogCopyDisable) {
+ configs.put(TopicConfig.REMOTE_LOG_COPY_DISABLE_CONFIG, remoteLogCopyDisable);
+ return this;
+ }
+
+ public LogConfigBuilder remoteLogDeleteOnDisable(boolean remoteLogDeleteOnDisable) {
+ configs.put(TopicConfig.REMOTE_LOG_DELETE_ON_DISABLE_CONFIG, remoteLogDeleteOnDisable);
+ return this;
+ }
+
+ public LogConfig build() {
+ return new LogConfig(configs);
+ }
+ }
}
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java
index ea14932ff20..0c7b4271d00 100644
--- a/storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java
@@ -16,17 +16,63 @@
*/
package org.apache.kafka.storage.internals.log;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.compress.Compression;
+import org.apache.kafka.common.record.ControlRecordType;
+import org.apache.kafka.common.record.DefaultRecordBatch;
+import org.apache.kafka.common.record.MemoryRecords;
+import org.apache.kafka.common.record.MemoryRecordsBuilder;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.record.SimpleRecord;
+import org.apache.kafka.common.record.TimestampType;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.transaction.TransactionLogConfig;
+import org.apache.kafka.server.common.TransactionVersion;
+import org.apache.kafka.server.storage.log.FetchIsolation;
+import org.apache.kafka.server.util.MockTime;
+import org.apache.kafka.server.util.Scheduler;
+import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache;
+import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+import java.util.function.Supplier;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class UnifiedLogTest {
private final File tmpDir = TestUtils.tempDirectory();
+ private final File logDir = TestUtils.randomPartitionLogDir(tmpDir);
+ private final BrokerTopicStats brokerTopicStats = new BrokerTopicStats(false);
+ private final MockTime mockTime = new MockTime();
+ private final int maxTransactionTimeoutMs = 60 * 60 * 1000;
+ private final ProducerStateManagerConfig producerStateManagerConfig = new ProducerStateManagerConfig(maxTransactionTimeoutMs, false);
+ private final List<UnifiedLog> logsToClose = new ArrayList<>();
+
+ private UnifiedLog log;
+
+ @AfterEach
+ public void tearDown() throws IOException {
+ brokerTopicStats.close();
+ for (UnifiedLog log : logsToClose) {
+ log.close();
+ }
+ Utils.delete(tmpDir);
+ }
@Test
public void testOffsetFromProducerSnapshotFile() {
@@ -34,4 +80,671 @@ public class UnifiedLogTest {
File snapshotFile = LogFileUtils.producerSnapshotFile(tmpDir, offset);
assertEquals(offset, UnifiedLog.offsetFromFile(snapshotFile));
}
+
+ @Test
+ public void shouldApplyEpochToMessageOnAppendIfLeader() throws IOException {
+ SimpleRecord[] records = java.util.stream.IntStream.range(0, 50)
+ .mapToObj(id -> new SimpleRecord(String.valueOf(id).getBytes()))
+ .toArray(SimpleRecord[]::new);
+
+ // Given this partition is on leader epoch 72
+ int epoch = 72;
+ try (UnifiedLog log = createLog(logDir, new LogConfig(new Properties()))) {
+ log.assignEpochStartOffset(epoch, records.length);
+
+ // When appending messages as a leader (i.e. assignOffsets = true)
+ for (SimpleRecord record : records) {
+ log.appendAsLeader(MemoryRecords.withRecords(Compression.NONE, record), epoch);
+ }
+
+ // Then leader epoch should be set on messages
+ for (int i = 0; i < records.length; i++) {
+ FetchDataInfo read = log.read(i, 1, FetchIsolation.LOG_END, true);
+ RecordBatch batch = read.records.batches().iterator().next();
+ assertEquals(epoch, batch.partitionLeaderEpoch(), "Should have set leader epoch");
+ }
+ }
+ }
+
+ @Test
+ public void followerShouldSaveEpochInformationFromReplicatedMessagesToTheEpochCache() throws IOException {
+ int[] messageIds = java.util.stream.IntStream.range(0, 50).toArray();
+ SimpleRecord[] records = Arrays.stream(messageIds)
+ .mapToObj(id -> new SimpleRecord(String.valueOf(id).getBytes()))
+ .toArray(SimpleRecord[]::new);
+
+ //Given each message has an offset & epoch, as msgs from leader would
+ Function<Integer, MemoryRecords> recordsForEpoch = i -> {
+ MemoryRecords recs = MemoryRecords.withRecords(messageIds[i], Compression.NONE, records[i]);
+ recs.batches().forEach(record -> {
+ record.setPartitionLeaderEpoch(42);
+ record.setLastOffset(i);
+ });
+ return recs;
+ };
+
+ try (UnifiedLog log = createLog(logDir, new LogConfig(new Properties()))) {
+ // Given each message has an offset & epoch, as msgs from leader would
+ for (int i = 0; i < records.length; i++) {
+ log.appendAsFollower(recordsForEpoch.apply(i), i);
+ }
+
+ assertEquals(Optional.of(42), log.latestEpoch());
+ }
+ }
+
+ @Test
+ public void shouldTruncateLeaderEpochsWhenDeletingSegments() throws IOException {
+ Supplier<MemoryRecords> records = () -> singletonRecords("test".getBytes());
+ LogConfig config = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(records.get().sizeInBytes() * 5)
+ .retentionBytes(records.get().sizeInBytes() * 10L)
+ .build();
+
+ log = createLog(logDir, config);
+ LeaderEpochFileCache cache = epochCache(log);
+
+ // Given three segments of 5 messages each
+ for (int i = 0; i < 15; i++) {
+ log.appendAsLeader(records.get(), 0);
+ }
+
+ // Given epochs
+ cache.assign(0, 0);
+ cache.assign(1, 5);
+ cache.assign(2, 10);
+
+ // When first segment is removed
+ log.updateHighWatermark(log.logEndOffset());
+ assertTrue(log.deleteOldSegments() > 0, "At least one segment should be deleted");
+
+ //The oldest epoch entry should have been removed
+ assertEquals(List.of(new EpochEntry(1, 5), new EpochEntry(2, 10)), cache.epochEntries());
+ }
+
+ @Test
+ public void shouldUpdateOffsetForLeaderEpochsWhenDeletingSegments() throws IOException {
+ Supplier<MemoryRecords> records = () -> singletonRecords("test".getBytes());
+ LogConfig config = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(records.get().sizeInBytes() * 5)
+ .retentionBytes(records.get().sizeInBytes() * 10L)
+ .build();
+
+ log = createLog(logDir, config);
+ LeaderEpochFileCache cache = epochCache(log);
+
+ // Given three segments of 5 messages each
+ for (int i = 0; i < 15; i++) {
+ log.appendAsLeader(records.get(), 0);
+ }
+
+ // Given epochs
+ cache.assign(0, 0);
+ cache.assign(1, 7);
+ cache.assign(2, 10);
+
+ // When first segment removed (up to offset 5)
+ log.updateHighWatermark(log.logEndOffset());
+ assertTrue(log.deleteOldSegments() > 0, "At least one segment should be deleted");
+
+ //The first entry should have gone from (0,0) => (0,5)
+ assertEquals(List.of(new EpochEntry(0, 5), new EpochEntry(1, 7), new EpochEntry(2, 10)), cache.epochEntries());
+ }
+
+ @Test
+ public void shouldTruncateLeaderEpochCheckpointFileWhenTruncatingLog() throws IOException {
+ Supplier<MemoryRecords> records = () -> records(List.of(new SimpleRecord("value".getBytes())), 0, 0);
+ LogConfig config = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(10 * records.get().sizeInBytes())
+ .build();
+ log = createLog(logDir, config);
+ LeaderEpochFileCache cache = epochCache(log);
+
+ //Given 2 segments, 10 messages per segment
+ append(0, 0, 10);
+ append(1, 10, 6);
+ append(2, 16, 4);
+
+ assertEquals(2, log.numberOfSegments());
+ assertEquals(20, log.logEndOffset());
+
+ // When truncate to LEO (no op)
+ log.truncateTo(log.logEndOffset());
+ // Then no change
+ assertEquals(3, cache.epochEntries().size());
+
+ // When truncate
+ log.truncateTo(11);
+ // Then no change
+ assertEquals(2, cache.epochEntries().size());
+
+ // When truncate
+ log.truncateTo(10);
+ assertEquals(1, cache.epochEntries().size());
+
+ // When truncate all
+ log.truncateTo(0);
+ assertEquals(0, cache.epochEntries().size());
+ }
+
+ @Test
+ public void shouldDeleteSizeBasedSegments() throws IOException {
+ Supplier<MemoryRecords> records = () -> singletonRecords("test".getBytes());
+ LogConfig config = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(records.get().sizeInBytes() * 5)
+ .retentionBytes(records.get().sizeInBytes() * 10L)
+ .build();
+ log = createLog(logDir, config);
+
+ // append some messages to create some segments
+ for (int i = 0; i < 15; i++) {
+ log.appendAsLeader(records.get(), 0);
+ }
+
+ log.updateHighWatermark(log.logEndOffset());
+ assertTrue(log.deleteOldSegments() > 0, "At least one segment should be deleted");
+ assertEquals(2, log.numberOfSegments(), "should have 2 segments");
+ }
+
+ @Test
+ public void shouldNotDeleteSizeBasedSegmentsWhenUnderRetentionSize() throws IOException {
+ Supplier<MemoryRecords> records = () -> singletonRecords("test".getBytes());
+ LogConfig config = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(records.get().sizeInBytes() * 5)
+ .retentionBytes(records.get().sizeInBytes() * 15L)
+ .build();
+
+ log = createLog(logDir, config);
+
+ // append some messages to create some segments
+ for (int i = 0; i < 15; i++) {
+ log.appendAsLeader(records.get(), 0);
+ }
+
+ log.updateHighWatermark(log.logEndOffset());
+ assertEquals(0, log.deleteOldSegments());
+ assertEquals(3, log.numberOfSegments(), "should have 3 segments");
+ }
+
+ @Test
+ public void shouldDeleteTimeBasedSegmentsReadyToBeDeleted() throws IOException {
+ Supplier<MemoryRecords> records = () -> singletonRecords("test".getBytes(), 10L);
+ LogConfig config = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(records.get().sizeInBytes() * 15)
+ .retentionMs(10000L)
+ .build();
+ log = createLog(logDir, config);
+
+ // append some messages to create some segments
+ for (int i = 0; i < 15; i++) {
+ log.appendAsLeader(records.get(), 0);
+ }
+
+ log.updateHighWatermark(log.logEndOffset());
+ assertTrue(log.deleteOldSegments() > 0, "At least one segment should be deleted");
+ assertEquals(1, log.numberOfSegments(), "There should be 1 segment remaining");
+ }
+
+ @Test
+ public void shouldNotDeleteTimeBasedSegmentsWhenNoneReadyToBeDeleted() throws IOException {
+ Supplier<MemoryRecords> records = () -> singletonRecords("test".getBytes(), mockTime.milliseconds());
+ LogConfig logConfig = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(records.get().sizeInBytes() * 5)
+ .retentionMs(10000000)
+ .build();
+ log = createLog(logDir, logConfig);
+
+ // append some messages to create some segments
+ for (int i = 0; i < 15; i++) {
+ log.appendAsLeader(records.get(), 0);
+ }
+
+ log.updateHighWatermark(log.logEndOffset());
+ assertEquals(0, log.deleteOldSegments());
+ assertEquals(3, log.numberOfSegments(), "There should be 3 segments remaining");
+ }
+
+ @Test
+ public void shouldNotDeleteSegmentsWhenPolicyDoesNotIncludeDelete() throws IOException {
+ Supplier<MemoryRecords> records = () -> singletonRecords("test".getBytes(), "test".getBytes(), 10L);
+ LogConfig config = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(records.get().sizeInBytes() * 5)
+ .retentionMs(10000)
+ .cleanupPolicy("compact")
+ .build();
+ log = createLog(logDir, config);
+
+ // append some messages to create some segments
+ for (int i = 0; i < 15; i++) {
+ log.appendAsLeader(records.get(), 0);
+ }
+
+ // mark the oldest segment as older the retention.ms
+ log.logSegments().iterator().next().setLastModified(mockTime.milliseconds() - 20000);
+
+ int segments = log.numberOfSegments();
+ log.updateHighWatermark(log.logEndOffset());
+ assertEquals(0, log.deleteOldSegments());
+ assertEquals(segments, log.numberOfSegments(), "There should be 3 segments remaining");
+ }
+
+ @Test
+ public void shouldDeleteSegmentsReadyToBeDeletedWhenCleanupPolicyIsCompactAndDelete() throws IOException {
+ Supplier<MemoryRecords> records = () -> singletonRecords("test".getBytes(), "test".getBytes(), 10L);
+ LogConfig config = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(records.get().sizeInBytes() * 5)
+ .retentionBytes(records.get().sizeInBytes() * 10L)
+ .cleanupPolicy("compact, delete")
+ .build();
+
+ log = createLog(logDir, config);
+
+ // append some messages to create some segments
+ for (int i = 0; i < 15; i++) {
+ log.appendAsLeader(records.get(), 0);
+ }
+
+ log.updateHighWatermark(log.logEndOffset());
+ assertTrue(log.deleteOldSegments() > 0, "At least one segment should be deleted");
+ assertEquals(1, log.numberOfSegments(), "There should be 1 segment remaining");
+ }
+
+ @Test
+ public void shouldDeleteLocalLogSegmentsWhenPolicyIsEmptyWithSizeRetention() throws IOException {
+ Supplier<MemoryRecords> records = () -> singletonRecords("test".getBytes(), "test".getBytes(), 10L);
+ int recordSize = records.get().sizeInBytes();
+ LogConfig config = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(recordSize * 2)
+ .retentionBytes(recordSize / 2)
+ .cleanupPolicy("")
+ .remoteLogStorageEnable(true)
+ .build();
+ log = createLog(logDir, config, true);
+
+ for (int i = 0; i < 10; i++) {
+ log.appendAsLeader(records.get(), 0);
+ }
+
+ int segmentsBefore = log.numberOfSegments();
+ log.updateHighWatermark(log.logEndOffset());
+ log.updateHighestOffsetInRemoteStorage(log.logEndOffset() - 1);
+ int deletedSegments = log.deleteOldSegments();
+
+ assertTrue(log.numberOfSegments() < segmentsBefore, "Some segments should be deleted due to size retention");
+ assertTrue(deletedSegments > 0, "At least one segment should be deleted");
+ }
+
+ @Test
+ public void shouldDeleteLocalLogSegmentsWhenPolicyIsEmptyWithMsRetention() throws IOException {
+ long oldTimestamp = mockTime.milliseconds() - 20000;
+ Supplier<MemoryRecords> oldRecords = () -> singletonRecords("test".getBytes(), "test".getBytes(), oldTimestamp);
+ int recordSize = oldRecords.get().sizeInBytes();
+ LogConfig logConfig = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(recordSize * 2)
+ .localRetentionMs(5000)
+ .cleanupPolicy("")
+ .remoteLogStorageEnable(true)
+ .build();
+ log = createLog(logDir, logConfig, true);
+
+ for (int i = 0; i < 10; i++) {
+ log.appendAsLeader(oldRecords.get(), 0);
+ }
+
+ Supplier<MemoryRecords> newRecords = () -> singletonRecords("test".getBytes(), "test".getBytes(), mockTime.milliseconds());
+ for (int i = 0; i < 5; i++) {
+ log.appendAsLeader(newRecords.get(), 0);
+ }
+
+ int segmentsBefore = log.numberOfSegments();
+
+ log.updateHighWatermark(log.logEndOffset());
+ log.updateHighestOffsetInRemoteStorage(log.logEndOffset() - 1);
+ int deletedSegments = log.deleteOldSegments();
+
+ assertTrue(log.numberOfSegments() < segmentsBefore, "Some segments should be deleted due to time retention");
+ assertTrue(deletedSegments > 0, "At least one segment should be deleted");
+ }
+
+ @Test
+ public void testLogDeletionAfterDeleteRecords() throws IOException {
+ Supplier<MemoryRecords> records = () -> singletonRecords("test".getBytes());
+ LogConfig logConfig = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(records.get().sizeInBytes() * 5)
+ .build();
+ log = createLog(logDir, logConfig);
+
+ for (int i = 0; i < 15; i++) {
+ log.appendAsLeader(records.get(), 0);
+ }
+ assertEquals(3, log.numberOfSegments());
+ assertEquals(0, log.logStartOffset());
+ log.updateHighWatermark(log.logEndOffset());
+
+ // The logStartOffset at the first segment so we did not delete it.
+ log.maybeIncrementLogStartOffset(1, LogStartOffsetIncrementReason.ClientRecordDeletion);
+ assertEquals(0, log.deleteOldSegments());
+ assertEquals(3, log.numberOfSegments());
+ assertEquals(1, log.logStartOffset());
+
+ log.maybeIncrementLogStartOffset(6, LogStartOffsetIncrementReason.ClientRecordDeletion);
+ assertTrue(log.deleteOldSegments() > 0, "At least one segment should be deleted");
+ assertEquals(2, log.numberOfSegments());
+ assertEquals(6, log.logStartOffset());
+
+ log.maybeIncrementLogStartOffset(15, LogStartOffsetIncrementReason.ClientRecordDeletion);
+ assertTrue(log.deleteOldSegments() > 0, "At least one segment should be deleted");
+ assertEquals(1, log.numberOfSegments());
+ assertEquals(15, log.logStartOffset());
+ }
+
+ @Test
+ public void testLogDeletionAfterClose() throws IOException {
+ Supplier<MemoryRecords> records = () -> singletonRecords("test".getBytes(), mockTime.milliseconds() - 1000);
+ LogConfig logConfig = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(records.get().sizeInBytes() * 5)
+ .segmentIndexBytes(1000)
+ .retentionMs(999)
+ .build();
+ log = createLog(logDir, logConfig);
+ // avoid close after test because it is closed in this test
+ logsToClose.remove(log);
+
+ // append some messages to create some segments
+ log.appendAsLeader(records.get(), 0);
+
+ assertEquals(1, log.numberOfSegments(), "The deleted segments should be gone.");
+ assertEquals(1, epochCache(log).epochEntries().size(), "Epoch entries should have gone.");
+
+ log.close();
+ log.delete();
+ assertEquals(0, log.numberOfSegments());
+ assertEquals(0, epochCache(log).epochEntries().size(), "Epoch entries should have gone.");
+ }
+
+ @Test
+ public void testDeleteOldSegments() throws IOException {
+ Supplier<MemoryRecords> records = () -> singletonRecords("test".getBytes(), mockTime.milliseconds() - 1000);
+ LogConfig logConfig = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(records.get().sizeInBytes() * 5)
+ .segmentIndexBytes(1000)
+ .retentionMs(999)
+ .build();
+ log = createLog(logDir, logConfig);
+ // avoid close after test because it is closed in this test
+ logsToClose.remove(log);
+
+ // append some messages to create some segments
+ for (int i = 0; i < 100; i++) {
+ log.appendAsLeader(records.get(), 0);
+ }
+
+ log.assignEpochStartOffset(0, 40);
+ log.assignEpochStartOffset(1, 90);
+
+ // segments are not eligible for deletion if no high watermark has been set
+ int numSegments = log.numberOfSegments();
+ assertEquals(0, log.deleteOldSegments());
+ assertEquals(numSegments, log.numberOfSegments());
+ assertEquals(0L, log.logStartOffset());
+
+ // only segments with offset before the current high watermark are eligible for deletion
+ for (long hw = 25; hw <= 30; hw++) {
+ log.updateHighWatermark(hw);
+ log.deleteOldSegments();
+ assertTrue(log.logStartOffset() <= hw);
+ long finalHw = hw;
+ log.logSegments().forEach(segment -> {
+ FetchDataInfo segmentFetchInfo;
+ try {
+ segmentFetchInfo = segment.read(segment.baseOffset(), Integer.MAX_VALUE);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ Optional<RecordBatch> lastBatch = Optional.empty();
+ for (RecordBatch batch : segmentFetchInfo.records.batches()) {
+ lastBatch = Optional.of(batch);
+ }
+ lastBatch.ifPresent(batch -> assertTrue(batch.lastOffset() >= finalHw));
+ });
+ }
+
+ log.updateHighWatermark(log.logEndOffset());
+ assertTrue(log.deleteOldSegments() > 0, "At least one segment should be deleted");
+ assertEquals(1, log.numberOfSegments(), "The deleted segments should be gone.");
+ assertEquals(1, epochCache(log).epochEntries().size(), "Epoch entries should have gone.");
+ assertEquals(new EpochEntry(1, 100), epochCache(log).epochEntries().get(0), "Epoch entry should be the latest epoch and the leo.");
+
+ for (int i = 0; i < 100; i++) {
+ log.appendAsLeader(records.get(), 0);
+ }
+
+ log.delete();
+ assertEquals(0, log.numberOfSegments(), "The number of segments should be 0");
+ assertEquals(0, log.deleteOldSegments(), "The number of deleted segments should be zero.");
+ assertEquals(0, epochCache(log).epochEntries().size(), "Epoch entries should have gone.");
+ }
+
+ @Test
+ public void shouldDeleteStartOffsetBreachedSegmentsWhenPolicyDoesNotIncludeDelete() throws IOException {
+ Supplier<MemoryRecords> records = () -> singletonRecords("test".getBytes(), "test".getBytes(), 10L);
+ int recordsPerSegment = 5;
+ LogConfig logConfig = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(records.get().sizeInBytes() * recordsPerSegment)
+ .segmentIndexBytes(1000)
+ .cleanupPolicy("compact")
+ .build();
+ log = createLog(logDir, logConfig);
+
+ // append some messages to create some segments
+ for (int i = 0; i < 15; i++) {
+ log.appendAsLeader(records.get(), 0);
+ }
+
+ assertEquals(3, log.numberOfSegments());
+ log.updateHighWatermark(log.logEndOffset());
+ log.maybeIncrementLogStartOffset(recordsPerSegment, LogStartOffsetIncrementReason.ClientRecordDeletion);
+
+ // The first segment, which is entirely before the log start offset, should be deleted
+ // Of the remaining the segments, the first can overlap the log start offset and the rest must have a base offset
+ // greater than the start offset.
+ log.updateHighWatermark(log.logEndOffset());
+ assertTrue(log.deleteOldSegments() > 0, "At least one segment should be deleted");
+ assertEquals(2, log.numberOfSegments(), "There should be 2 segments remaining");
+ assertTrue(log.logSegments().iterator().next().baseOffset() <= log.logStartOffset());
+ log.logSegments().forEach(segment -> {
+ if (log.logSegments().iterator().next() != segment) {
+ assertTrue(segment.baseOffset() > log.logStartOffset());
+ }
+ });
+ }
+
+ @Test
+ public void testFirstUnstableOffsetNoTransactionalData() throws IOException {
+ LogConfig logConfig = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(1024 * 1024 * 5)
+ .build();
+ log = createLog(logDir, logConfig);
+
+ MemoryRecords records = MemoryRecords.withRecords(Compression.NONE,
+ new SimpleRecord("foo".getBytes()),
+ new SimpleRecord("bar".getBytes()),
+ new SimpleRecord("baz".getBytes()));
+
+ log.appendAsLeader(records, 0);
+ assertEquals(Optional.empty(), log.firstUnstableOffset());
+ }
+
+ @Test
+ public void testFirstUnstableOffsetWithTransactionalData() throws IOException {
+ LogConfig logConfig = new LogTestUtils.LogConfigBuilder()
+ .segmentBytes(1024 * 1024 * 5)
+ .build();
+ log = createLog(logDir, logConfig);
+
+ long pid = 137L;
+ short epoch = 5;
+ int seq = 0;
+
+ // add some transactional records
+ MemoryRecords records = MemoryRecords.withTransactionalRecords(
+ Compression.NONE, pid, epoch, seq,
+ new SimpleRecord("foo".getBytes()),
+ new SimpleRecord("bar".getBytes()),
+ new SimpleRecord("baz".getBytes()));
+
+ LogAppendInfo firstAppendInfo = log.appendAsLeader(records, 0);
+ assertEquals(Optional.of(firstAppendInfo.firstOffset()), log.firstUnstableOffset());
+
+ // add more transactional records
+ seq += 3;
+ log.appendAsLeader(MemoryRecords.withTransactionalRecords(Compression.NONE, pid, epoch, seq,
+ new SimpleRecord("blah".getBytes())), 0);
+ assertEquals(Optional.of(firstAppendInfo.firstOffset()), log.firstUnstableOffset());
+
+ // now transaction is committed
+ LogAppendInfo commitAppendInfo = LogTestUtils.appendEndTxnMarkerAsLeader(log, pid, epoch,
+ ControlRecordType.COMMIT, mockTime.milliseconds(), 0, 0, TransactionVersion.TV_0.featureLevel());
+
+ // first unstable offset is not updated until the high watermark is advanced
+ assertEquals(Optional.of(firstAppendInfo.firstOffset()), log.firstUnstableOffset());
+ log.updateHighWatermark(commitAppendInfo.lastOffset() + 1);
+
+ // now there should be no first unstable offset
+ assertEquals(Optional.empty(), log.firstUnstableOffset());
+ }
+
+ private void append(int epoch, long startOffset, int count) {
+ Function<Integer, MemoryRecords> records = i ->
+ records(List.of(new SimpleRecord("value".getBytes())), startOffset + i, epoch);
+ for (int i = 0; i < count; i++) {
+ log.appendAsFollower(records.apply(i), epoch);
+ }
+ }
+
+ private LeaderEpochFileCache epochCache(UnifiedLog log) {
+ return log.leaderEpochCache();
+ }
+
+ private UnifiedLog createLog(File dir, LogConfig config) throws IOException {
+ return createLog(dir, config, false);
+ }
+
+ private UnifiedLog createLog(File dir, LogConfig config, boolean remoteStorageSystemEnable) throws IOException {
+ return createLog(dir, config, this.brokerTopicStats, mockTime.scheduler, this.mockTime,
+ this.producerStateManagerConfig, Optional.empty(), remoteStorageSystemEnable);
+ }
+
+ private UnifiedLog createLog(
+ File dir,
+ LogConfig config,
+ BrokerTopicStats brokerTopicStats,
+ Scheduler scheduler,
+ MockTime time,
+ ProducerStateManagerConfig producerStateManagerConfig,
+ Optional<Uuid> topicId,
+ boolean remoteStorageSystemEnable) throws IOException {
+
+ UnifiedLog log = UnifiedLog.create(
+ dir,
+ config,
+ 0L,
+ 0L,
+ scheduler,
+ brokerTopicStats,
+ time,
+ 3600000,
+ producerStateManagerConfig,
+ TransactionLogConfig.PRODUCER_ID_EXPIRATION_CHECK_INTERVAL_MS_DEFAULT,
+ new LogDirFailureChannel(10),
+ true,
+ topicId,
+ new ConcurrentHashMap<>(),
+ remoteStorageSystemEnable,
+ LogOffsetsListener.NO_OP_OFFSETS_LISTENER
+ );
+
+ this.logsToClose.add(log);
+ return log;
+ }
+
+ public static MemoryRecords singletonRecords(byte[] value, byte[] key) {
+ return singletonRecords(value, key, Compression.NONE, RecordBatch.NO_TIMESTAMP, RecordBatch.CURRENT_MAGIC_VALUE);
+ }
+
+ public static MemoryRecords singletonRecords(byte[] value, long timestamp) {
+ return singletonRecords(value, null, Compression.NONE, timestamp, RecordBatch.CURRENT_MAGIC_VALUE);
+ }
+
+ public static MemoryRecords singletonRecords(
+ byte[] value
+ ) {
+ return records(List.of(new SimpleRecord(RecordBatch.NO_TIMESTAMP, null, value)),
+ RecordBatch.CURRENT_MAGIC_VALUE,
+ Compression.NONE,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ RecordBatch.NO_SEQUENCE,
+ 0,
+ RecordBatch.NO_PARTITION_LEADER_EPOCH
+ );
+ }
+
+ public static MemoryRecords singletonRecords(
+ byte[] value,
+ byte[] key,
+ Compression codec,
+ long timestamp,
+ byte magicValue
+ ) {
+ return records(List.of(new SimpleRecord(timestamp, key, value)),
+ magicValue, codec,
+ RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_EPOCH,
+ RecordBatch.NO_SEQUENCE,
+ 0,
+ RecordBatch.NO_PARTITION_LEADER_EPOCH
+ );
+ }
+
+ public static MemoryRecords singletonRecords(byte[] value, byte[] key, long timestamp) {
+ return singletonRecords(value, key, Compression.NONE, timestamp, RecordBatch.CURRENT_MAGIC_VALUE);
+ }
+
+ public static MemoryRecords records(List<SimpleRecord> records) {
+ return records(records, RecordBatch.CURRENT_MAGIC_VALUE, Compression.NONE, RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, 0L, RecordBatch.NO_PARTITION_LEADER_EPOCH);
+ }
+
+ public static MemoryRecords records(List<SimpleRecord> records, long baseOffset) {
+ return records(records, RecordBatch.CURRENT_MAGIC_VALUE, Compression.NONE, RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, baseOffset, RecordBatch.NO_PARTITION_LEADER_EPOCH);
+ }
+
+ public static MemoryRecords records(List<SimpleRecord> records, long baseOffset, int partitionLeaderEpoch) {
+ return records(records, RecordBatch.CURRENT_MAGIC_VALUE, Compression.NONE, RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, baseOffset, partitionLeaderEpoch);
+ }
+
+ public static MemoryRecords records(List<SimpleRecord> records, byte magicValue, Compression compression) {
+ return records(records, magicValue, compression, RecordBatch.NO_PRODUCER_ID,
+ RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, 0L, RecordBatch.NO_PARTITION_LEADER_EPOCH);
+ }
+
+ public static MemoryRecords records(List<SimpleRecord> records,
+ byte magicValue,
+ Compression compression,
+ long producerId,
+ short producerEpoch,
+ int sequence,
+ long baseOffset,
+ int partitionLeaderEpoch) {
+ ByteBuffer buf = ByteBuffer.allocate(DefaultRecordBatch.sizeInBytes(records));
+ MemoryRecordsBuilder builder = MemoryRecords.builder(buf, magicValue, compression, TimestampType.CREATE_TIME, baseOffset,
+ System.currentTimeMillis(), producerId, producerEpoch, sequence, false, partitionLeaderEpoch);
+ for (SimpleRecord record : records) {
+ builder.append(record);
+ }
+ return builder.build();
+ }
}
diff --git a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestUtils.java b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestUtils.java
index 4c75272edd4..a9e49dd0492 100644
--- a/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestUtils.java
+++ b/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestUtils.java
@@ -35,6 +35,7 @@ import static java.lang.String.format;
* Helper functions for writing unit tests.
* <p>
* <b>Package-private:</b> Not intended for use outside {@code org.apache.kafka.common.test}.
+ * Use {@code org/apache/kafka/test/TestUtils} instead.
*/
class TestUtils {
private static final Logger log = LoggerFactory.getLogger(TestUtils.class);