You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sh...@apache.org on 2024/05/27 09:44:31 UTC
(kafka) branch trunk updated: KAFKA-16452: Don't throw OOORE when converting the offset to metadata (#15825)
This is an automated email from the ASF dual-hosted git repository.
showuon 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 524ad1e14b1 KAFKA-16452: Don't throw OOORE when converting the offset to metadata (#15825)
524ad1e14b1 is described below
commit 524ad1e14b1f3ff2c5eff703fdf3534092ca58fd
Author: Kamal Chandraprakash <kc...@uber.com>
AuthorDate: Mon May 27 15:14:23 2024 +0530
KAFKA-16452: Don't throw OOORE when converting the offset to metadata (#15825)
Don't throw OFFSET_OUT_OF_RANGE error when converting the offset to metadata, and next time the leader should increment the high watermark by itself after receiving fetch requests from followers. This can happen when checkpoint files are missing and being elected as a leader.
Reviewers: Luke Chen <sh...@gmail.com>, Jun Rao <ju...@apache.org>
---
core/src/main/scala/kafka/log/LocalLog.scala | 19 +++--
core/src/main/scala/kafka/log/UnifiedLog.scala | 21 ++++--
.../main/scala/kafka/raft/KafkaMetadataLog.scala | 2 +-
.../src/main/scala/kafka/server/DelayedFetch.scala | 14 ++--
.../kafka/server/DelayedFetchTest.scala | 66 +++++++++++++++--
.../scala/kafka/raft/KafkaMetadataLogTest.scala | 18 +++++
.../test/scala/unit/kafka/log/LocalLogTest.scala | 37 ++++++++++
.../test/scala/unit/kafka/log/LogLoaderTest.scala | 2 +-
.../test/scala/unit/kafka/log/LogSegmentTest.scala | 32 ++++++++-
.../test/scala/unit/kafka/log/UnifiedLogTest.scala | 42 ++++++++++-
.../storage/internals/log/LogOffsetMetadata.java | 16 ++---
.../kafka/storage/internals/log/LogSegment.java | 18 ++---
.../internals/log/LogOffsetMetadataTest.java | 82 +++++++++++++++++++++
.../tiered/storage/TieredStorageTestBuilder.java | 8 +++
.../tiered/storage/TieredStorageTestContext.java | 17 ++++-
.../storage/actions/EraseBrokerStorageAction.java | 16 ++++-
...FetchFromLeaderWithCorruptedCheckpointTest.java | 84 ++++++++++++++++++++++
.../tiered/storage/utils/BrokerLocalStorage.java | 5 +-
18 files changed, 448 insertions(+), 51 deletions(-)
diff --git a/core/src/main/scala/kafka/log/LocalLog.scala b/core/src/main/scala/kafka/log/LocalLog.scala
index b2121f5312d..09fa464af04 100644
--- a/core/src/main/scala/kafka/log/LocalLog.scala
+++ b/core/src/main/scala/kafka/log/LocalLog.scala
@@ -383,12 +383,19 @@ class LocalLog(@volatile private var _dir: File,
val segment = segmentOpt.get
val baseOffset = segment.baseOffset
- val maxPosition =
- // Use the max offset position if it is on this segment; otherwise, the segment size is the limit.
- if (maxOffsetMetadata.segmentBaseOffset == segment.baseOffset) maxOffsetMetadata.relativePositionInSegment
- else segment.size
-
- fetchDataInfo = segment.read(startOffset, maxLength, maxPosition, minOneMessage)
+ // 1. If `maxOffsetMetadata#segmentBaseOffset < segment#baseOffset`, then return maxPosition as empty.
+ // 2. Use the max-offset position if it is on this segment; otherwise, the segment size is the limit.
+ // 3. When maxOffsetMetadata is message-offset-only, then we don't know the relativePositionInSegment so
+ // return maxPosition as empty to avoid reading beyond the max-offset
+ val maxPositionOpt: Optional[java.lang.Long] =
+ if (segment.baseOffset < maxOffsetMetadata.segmentBaseOffset)
+ Optional.of(segment.size)
+ else if (segment.baseOffset == maxOffsetMetadata.segmentBaseOffset && !maxOffsetMetadata.messageOffsetOnly())
+ Optional.of(maxOffsetMetadata.relativePositionInSegment)
+ else
+ Optional.empty()
+
+ fetchDataInfo = segment.read(startOffset, maxLength, maxPositionOpt, minOneMessage)
if (fetchDataInfo != null) {
if (includeAbortedTxns)
fetchDataInfo = addAbortedTransactions(startOffset, segment, fetchDataInfo)
diff --git a/core/src/main/scala/kafka/log/UnifiedLog.scala b/core/src/main/scala/kafka/log/UnifiedLog.scala
index 5508660a41a..b68a9f35a0c 100644
--- a/core/src/main/scala/kafka/log/UnifiedLog.scala
+++ b/core/src/main/scala/kafka/log/UnifiedLog.scala
@@ -361,7 +361,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
val offsetMetadata = highWatermarkMetadata
if (offsetMetadata.messageOffsetOnly) {
lock.synchronized {
- val fullOffset = convertToOffsetMetadataOrThrow(highWatermark)
+ val fullOffset = maybeConvertToOffsetMetadata(highWatermark)
updateHighWatermarkMetadata(fullOffset)
fullOffset
}
@@ -405,7 +405,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
case Some(offsetMetadata) if offsetMetadata.messageOffset < highWatermarkMetadata.messageOffset =>
if (offsetMetadata.messageOffsetOnly) {
lock synchronized {
- val fullOffset = convertToOffsetMetadataOrThrow(offsetMetadata.messageOffset)
+ val fullOffset = maybeConvertToOffsetMetadata(offsetMetadata.messageOffset)
if (firstUnstableOffsetMetadata.contains(offsetMetadata))
firstUnstableOffsetMetadata = Some(fullOffset)
fullOffset
@@ -964,7 +964,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
val updatedFirstUnstableOffset = producerStateManager.firstUnstableOffset.asScala match {
case Some(logOffsetMetadata) if logOffsetMetadata.messageOffsetOnly || logOffsetMetadata.messageOffset < logStartOffset =>
val offset = math.max(logOffsetMetadata.messageOffset, logStartOffset)
- Some(convertToOffsetMetadataOrThrow(offset))
+ Some(maybeConvertToOffsetMetadata(offset))
case other => other
}
@@ -1425,11 +1425,18 @@ class UnifiedLog(@volatile var logStartOffset: Long,
/**
* Given a message offset, find its corresponding offset metadata in the log.
- * If the message offset is out of range, throw an OffsetOutOfRangeException
+ * 1. If the message offset is less than the log-start-offset (or) local-log-start-offset, then it returns the
+ * message-only metadata.
+ * 2. If the message offset is beyond the log-end-offset, then it returns the message-only metadata.
+ * 3. For all other cases, it returns the offset metadata from the log.
*/
- private def convertToOffsetMetadataOrThrow(offset: Long): LogOffsetMetadata = {
- checkLogStartOffset(offset)
- localLog.convertToOffsetMetadataOrThrow(offset)
+ private[log] def maybeConvertToOffsetMetadata(offset: Long): LogOffsetMetadata = {
+ try {
+ localLog.convertToOffsetMetadataOrThrow(offset)
+ } catch {
+ case _: OffsetOutOfRangeException =>
+ new LogOffsetMetadata(offset)
+ }
}
/**
diff --git a/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala b/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala
index 15f87b9c247..5e107aa1487 100644
--- a/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala
+++ b/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala
@@ -223,7 +223,7 @@ final class KafkaMetadataLog private (
override def highWatermark: LogOffsetMetadata = {
val hwm = log.fetchOffsetSnapshot.highWatermark
- val segmentPosition: Optional[OffsetMetadata] = if (hwm.messageOffsetOnly) {
+ val segmentPosition: Optional[OffsetMetadata] = if (!hwm.messageOffsetOnly) {
Optional.of(SegmentPosition(hwm.segmentBaseOffset, hwm.relativePositionInSegment))
} else {
Optional.empty()
diff --git a/core/src/main/scala/kafka/server/DelayedFetch.scala b/core/src/main/scala/kafka/server/DelayedFetch.scala
index f8b60b6071d..cc89631ab40 100644
--- a/core/src/main/scala/kafka/server/DelayedFetch.scala
+++ b/core/src/main/scala/kafka/server/DelayedFetch.scala
@@ -91,19 +91,19 @@ class DelayedFetch(
// Go directly to the check for Case G if the message offsets are the same. If the log segment
// has just rolled, then the high watermark offset will remain the same but be on the old segment,
// which would incorrectly be seen as an instance of Case F.
- if (endOffset.messageOffset != fetchOffset.messageOffset) {
- if (endOffset.onOlderSegment(fetchOffset)) {
- // Case F, this can happen when the new fetch operation is on a truncated leader
- debug(s"Satisfying fetch $this since it is fetching later segments of partition $topicIdPartition.")
- return forceComplete()
- } else if (fetchOffset.onOlderSegment(endOffset)) {
+ if (fetchOffset.messageOffset > endOffset.messageOffset) {
+ // Case F, this can happen when the new fetch operation is on a truncated leader
+ debug(s"Satisfying fetch $this since it is fetching later segments of partition $topicIdPartition.")
+ return forceComplete()
+ } else if (fetchOffset.messageOffset < endOffset.messageOffset) {
+ if (fetchOffset.onOlderSegment(endOffset)) {
// Case F, this can happen when the fetch operation is falling behind the current segment
// or the partition has just rolled a new segment
debug(s"Satisfying fetch $this immediately since it is fetching older segments.")
// We will not force complete the fetch request if a replica should be throttled.
if (!params.isFromFollower || !replicaManager.shouldLeaderThrottle(quota, partition, params.replicaId))
return forceComplete()
- } else if (fetchOffset.messageOffset < endOffset.messageOffset) {
+ } else if (fetchOffset.onSameSegment(endOffset)) {
// we take the partition fetch size as upper bound when accumulating the bytes (skip if a throttled partition)
val bytesAvailable = math.min(endOffset.positionDiff(fetchOffset), fetchStatus.fetchInfo.maxBytes)
if (!params.isFromFollower || !replicaManager.shouldLeaderThrottle(quota, partition, params.replicaId))
diff --git a/core/src/test/scala/integration/kafka/server/DelayedFetchTest.scala b/core/src/test/scala/integration/kafka/server/DelayedFetchTest.scala
index f26f7079d4b..ccf4b84b696 100644
--- a/core/src/test/scala/integration/kafka/server/DelayedFetchTest.scala
+++ b/core/src/test/scala/integration/kafka/server/DelayedFetchTest.scala
@@ -28,6 +28,8 @@ import org.apache.kafka.common.requests.FetchRequest
import org.apache.kafka.storage.internals.log.{FetchDataInfo, FetchIsolation, FetchParams, FetchPartitionData, LogOffsetMetadata, LogOffsetSnapshot}
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
import org.mockito.ArgumentMatchers.{any, anyInt}
import org.mockito.Mockito.{mock, when}
@@ -46,7 +48,7 @@ class DelayedFetchTest {
val fetchStatus = FetchPartitionStatus(
startOffsetMetadata = new LogOffsetMetadata(fetchOffset),
- fetchInfo = new FetchRequest.PartitionData(Uuid.ZERO_UUID, fetchOffset, logStartOffset, maxBytes, currentLeaderEpoch))
+ fetchInfo = new FetchRequest.PartitionData(topicIdPartition.topicId(), fetchOffset, logStartOffset, maxBytes, currentLeaderEpoch))
val fetchParams = buildFollowerFetchParams(replicaId, maxWaitMs = 500)
var fetchResultOpt: Option[FetchPartitionData] = None
@@ -92,7 +94,7 @@ class DelayedFetchTest {
val fetchStatus = FetchPartitionStatus(
startOffsetMetadata = new LogOffsetMetadata(fetchOffset),
- fetchInfo = new FetchRequest.PartitionData(Uuid.ZERO_UUID, fetchOffset, logStartOffset, maxBytes, currentLeaderEpoch))
+ fetchInfo = new FetchRequest.PartitionData(topicIdPartition.topicId(), fetchOffset, logStartOffset, maxBytes, currentLeaderEpoch))
val fetchParams = buildFollowerFetchParams(replicaId, maxWaitMs = 500)
var fetchResultOpt: Option[FetchPartitionData] = None
@@ -116,6 +118,9 @@ class DelayedFetchTest {
assertTrue(delayedFetch.tryComplete())
assertTrue(delayedFetch.isCompleted)
assertTrue(fetchResultOpt.isDefined)
+
+ val fetchResult = fetchResultOpt.get
+ assertEquals(Errors.NOT_LEADER_OR_FOLLOWER, fetchResult.error)
}
@Test
@@ -164,18 +169,71 @@ class DelayedFetchTest {
assertTrue(delayedFetch.tryComplete())
assertTrue(delayedFetch.isCompleted)
assertTrue(fetchResultOpt.isDefined)
+
+ val fetchResult = fetchResultOpt.get
+ assertEquals(Errors.NONE, fetchResult.error)
+ }
+
+ @ParameterizedTest(name = "testDelayedFetchWithMessageOnlyHighWatermark endOffset={0}")
+ @ValueSource(longs = Array(0, 500))
+ def testDelayedFetchWithMessageOnlyHighWatermark(endOffset: Long): Unit = {
+ val topicIdPartition = new TopicIdPartition(Uuid.randomUuid(), 0, "topic")
+ val fetchOffset = 450L
+ val logStartOffset = 5L
+ val currentLeaderEpoch = Optional.of[Integer](10)
+ val replicaId = 1
+
+ val fetchStatus = FetchPartitionStatus(
+ startOffsetMetadata = new LogOffsetMetadata(fetchOffset),
+ fetchInfo = new FetchRequest.PartitionData(topicIdPartition.topicId, fetchOffset, logStartOffset, maxBytes, currentLeaderEpoch))
+ val fetchParams = buildFollowerFetchParams(replicaId, maxWaitMs = 500)
+
+ var fetchResultOpt: Option[FetchPartitionData] = None
+ def callback(responses: Seq[(TopicIdPartition, FetchPartitionData)]): Unit = {
+ fetchResultOpt = Some(responses.head._2)
+ }
+
+ val delayedFetch = new DelayedFetch(
+ params = fetchParams,
+ fetchPartitionStatus = Seq(topicIdPartition -> fetchStatus),
+ replicaManager = replicaManager,
+ quota = replicaQuota,
+ responseCallback = callback
+ )
+
+ val partition: Partition = mock(classOf[Partition])
+ when(replicaManager.getPartitionOrException(topicIdPartition.topicPartition)).thenReturn(partition)
+ // Note that the high-watermark does not contain the complete metadata
+ val endOffsetMetadata = new LogOffsetMetadata(endOffset, -1L, -1)
+ when(partition.fetchOffsetSnapshot(
+ currentLeaderEpoch,
+ fetchOnlyFromLeader = true))
+ .thenReturn(new LogOffsetSnapshot(0L, endOffsetMetadata, endOffsetMetadata, endOffsetMetadata))
+ when(replicaManager.isAddingReplica(any(), anyInt())).thenReturn(false)
+ expectReadFromReplica(fetchParams, topicIdPartition, fetchStatus.fetchInfo, Errors.NONE)
+
+ // 1. When `endOffset` is 0, it refers to the truncation case
+ // 2. When `endOffset` is 500, we won't complete because it doesn't contain offset metadata
+ val expected = endOffset == 0
+ assertEquals(expected, delayedFetch.tryComplete())
+ assertEquals(expected, delayedFetch.isCompleted)
+ assertEquals(expected, fetchResultOpt.isDefined)
+ if (fetchResultOpt.isDefined) {
+ assertEquals(Errors.NONE, fetchResultOpt.get.error)
+ }
}
private def buildFollowerFetchParams(
replicaId: Int,
- maxWaitMs: Int
+ maxWaitMs: Int,
+ minBytes: Int = 1,
): FetchParams = {
new FetchParams(
ApiKeys.FETCH.latestVersion,
replicaId,
1,
maxWaitMs,
- 1,
+ minBytes,
maxBytes,
FetchIsolation.LOG_END,
Optional.empty()
diff --git a/core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala b/core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala
index 5bf8f328845..4f2daf3bfa6 100644
--- a/core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala
+++ b/core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala
@@ -184,6 +184,24 @@ final class KafkaMetadataLogTest {
)
}
+ @Test
+ def testHighWatermarkOffsetMetadata(): Unit = {
+ val numberOfRecords = 10
+ val epoch = 1
+ val log = buildMetadataLog(tempDir, mockTime)
+
+ append(log, numberOfRecords, epoch)
+ log.updateHighWatermark(new LogOffsetMetadata(numberOfRecords))
+
+ val highWatermarkMetadata = log.highWatermark
+ assertEquals(numberOfRecords, highWatermarkMetadata.offset)
+ assertTrue(highWatermarkMetadata.metadata.isPresent)
+
+ val segmentPosition = highWatermarkMetadata.metadata.get().asInstanceOf[SegmentPosition]
+ assertEquals(0, segmentPosition.baseOffset)
+ assertTrue(segmentPosition.relativePosition > 0)
+ }
+
@Test
def testCreateSnapshotBeforeLogStartOffset(): Unit = {
val numberOfRecords = 10
diff --git a/core/src/test/scala/unit/kafka/log/LocalLogTest.scala b/core/src/test/scala/unit/kafka/log/LocalLogTest.scala
index a0ba0c478c7..cf1f1a06177 100644
--- a/core/src/test/scala/unit/kafka/log/LocalLogTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LocalLogTest.scala
@@ -372,6 +372,43 @@ class LocalLogTest {
assertTrue(fetchDataInfo.records.records.asScala.isEmpty)
}
+ @Test
+ def testWhenFetchOffsetHigherThanMaxOffset(): Unit = {
+ val record = new SimpleRecord(mockTime.milliseconds, "a".getBytes)
+ for (offset <- 0 to 4) {
+ appendRecords(List(record), initialOffset = offset)
+ if (offset % 2 != 0)
+ log.roll()
+ }
+ assertEquals(3, log.segments.numberOfSegments)
+
+ // case-0: valid case, `startOffset` < `maxOffsetMetadata.offset`
+ var fetchDataInfo = readRecords(startOffset = 3L, maxOffsetMetadata = new LogOffsetMetadata(4L, 4L, 0))
+ assertEquals(1, fetchDataInfo.records.records.asScala.size)
+ assertEquals(new LogOffsetMetadata(3, 2L, 69), fetchDataInfo.fetchOffsetMetadata)
+
+ // case-1: `startOffset` == `maxOffsetMetadata.offset`
+ fetchDataInfo = readRecords(startOffset = 4L, maxOffsetMetadata = new LogOffsetMetadata(4L, 4L, 0))
+ assertTrue(fetchDataInfo.records.records.asScala.isEmpty)
+ assertEquals(new LogOffsetMetadata(4L, 4L, 0), fetchDataInfo.fetchOffsetMetadata)
+
+ // case-2: `startOffset` > `maxOffsetMetadata.offset`
+ fetchDataInfo = readRecords(startOffset = 5L, maxOffsetMetadata = new LogOffsetMetadata(4L, 4L, 0))
+ assertTrue(fetchDataInfo.records.records.asScala.isEmpty)
+ assertEquals(new LogOffsetMetadata(5L, 4L, 69), fetchDataInfo.fetchOffsetMetadata)
+
+ // case-3: `startOffset` < `maxMessageOffset.offset` but `maxMessageOffset.messageOnlyOffset` is true
+ fetchDataInfo = readRecords(startOffset = 3L, maxOffsetMetadata = new LogOffsetMetadata(4L, -1L, -1))
+ assertTrue(fetchDataInfo.records.records.asScala.isEmpty)
+ assertEquals(new LogOffsetMetadata(3L, 2L, 69), fetchDataInfo.fetchOffsetMetadata)
+
+ // case-4: `startOffset` < `maxMessageOffset.offset`, `maxMessageOffset.messageOnlyOffset` is false, but
+ // `maxOffsetMetadata.segmentBaseOffset` < `startOffset.segmentBaseOffset`
+ fetchDataInfo = readRecords(startOffset = 3L, maxOffsetMetadata = new LogOffsetMetadata(4L, 0L, 40))
+ assertTrue(fetchDataInfo.records.records.asScala.isEmpty)
+ assertEquals(new LogOffsetMetadata(3L, 2L, 69), fetchDataInfo.fetchOffsetMetadata)
+ }
+
@Test
def testTruncateTo(): Unit = {
for (offset <- 0 to 11) {
diff --git a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala b/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
index e1c204bf871..1a781a93ea6 100644
--- a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
@@ -358,7 +358,7 @@ class LogLoaderTest {
val wrapper = Mockito.spy(segment)
Mockito.doAnswer { in =>
segmentsWithReads += wrapper
- segment.read(in.getArgument(0, classOf[java.lang.Long]), in.getArgument(1, classOf[java.lang.Integer]), in.getArgument(2, classOf[java.lang.Long]), in.getArgument(3, classOf[java.lang.Boolean]))
+ segment.read(in.getArgument(0, classOf[java.lang.Long]), in.getArgument(1, classOf[java.lang.Integer]), in.getArgument(2, classOf[java.util.Optional[java.lang.Long]]), in.getArgument(3, classOf[java.lang.Boolean]))
}.when(wrapper).read(ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any(), ArgumentMatchers.any())
Mockito.doAnswer { in =>
recoveredSegments += wrapper
diff --git a/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala b/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala
index abfa4810160..25d4fc8626b 100644
--- a/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala
@@ -30,7 +30,7 @@ import org.apache.kafka.storage.internals.log._
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import org.junit.jupiter.params.ParameterizedTest
-import org.junit.jupiter.params.provider.CsvSource
+import org.junit.jupiter.params.provider.{CsvSource, ValueSource}
import java.io.{File, RandomAccessFile}
import java.util
@@ -144,6 +144,34 @@ class LogSegmentTest {
checkEquals(ms2.records.iterator, read.records.records.iterator)
}
+ @ParameterizedTest(name = "testReadWhenNoMaxPosition minOneMessage = {0}")
+ @ValueSource(booleans = Array(true, false))
+ def testReadWhenNoMaxPosition(minOneMessage: Boolean): Unit = {
+ val maxPosition: Optional[java.lang.Long] = Optional.empty()
+ val maxSize = 1
+ val seg = createSegment(40)
+ val ms = records(50, "hello", "there")
+ seg.append(51, RecordBatch.NO_TIMESTAMP, -1L, ms)
+ // read before first offset
+ var read = seg.read(48, maxSize, maxPosition, minOneMessage)
+ assertEquals(new LogOffsetMetadata(48, 40, 0), read.fetchOffsetMetadata)
+ assertTrue(read.records.records().iterator().asScala.isEmpty)
+ // read at first offset
+ read = seg.read(50, maxSize, maxPosition, minOneMessage)
+ assertEquals(new LogOffsetMetadata(50, 40, 0), read.fetchOffsetMetadata)
+ assertTrue(read.records.records().iterator().asScala.isEmpty)
+ // read at last offset
+ read = seg.read(51, maxSize, maxPosition, minOneMessage)
+ assertEquals(new LogOffsetMetadata(51, 40, 39), read.fetchOffsetMetadata)
+ assertTrue(read.records.records().iterator().asScala.isEmpty)
+ // read at log-end-offset
+ read = seg.read(52, maxSize, maxPosition, minOneMessage)
+ assertNull(read)
+ // read beyond log-end-offset
+ read = seg.read(53, maxSize, maxPosition, minOneMessage)
+ assertNull(read)
+ }
+
/**
* In a loop append two messages then truncate off the second of those messages and check that we can read
* the first but not the second message.
@@ -331,7 +359,7 @@ class LogSegmentTest {
writeNonsenseToFile(indexFile, 5, indexFile.length.toInt)
seg.recover(newProducerStateManager(), Optional.empty())
for (i <- 0 until 100) {
- val records = seg.read(i, 1, seg.size(), true).records.records
+ val records = seg.read(i, 1, Optional.of(seg.size()), true).records.records
assertEquals(i, records.iterator.next().offset)
}
}
diff --git a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
index 2394bc2d028..c31e1739f31 100755
--- a/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
+++ b/core/src/test/scala/unit/kafka/log/UnifiedLogTest.scala
@@ -3246,7 +3246,7 @@ class UnifiedLogTest {
val readInfo = segment.read(offsetMetadata.messageOffset,
2048,
- segment.size,
+ Optional.of(segment.size),
false)
if (offsetMetadata.relativePositionInSegment < segment.size)
@@ -4211,6 +4211,46 @@ class UnifiedLogTest {
assertEquals(31, log.localLogStartOffset())
}
+ @Test
+ def testConvertToOffsetMetadataDoesNotThrowOffsetOutOfRangeError(): Unit = {
+ val logConfig = LogTestUtils.createLogConfig(localRetentionBytes = 1, fileDeleteDelayMs = 0, remoteLogStorageEnable = true)
+ val log = createLog(logDir, logConfig, remoteStorageSystemEnable = true)
+
+ var offset = 0L
+ for(_ <- 0 until 50) {
+ val records = TestUtils.singletonRecords("test".getBytes())
+ val info = log.appendAsLeader(records, leaderEpoch = 0)
+ offset = info.lastOffset
+ if (offset != 0 && offset % 10 == 0)
+ log.roll()
+ }
+ assertEquals(5, log.logSegments.size)
+ log.updateHighWatermark(log.logEndOffset)
+ // simulate calls to upload 3 segments to remote storage
+ log.updateHighestOffsetInRemoteStorage(30)
+
+ log.deleteOldSegments()
+ assertEquals(2, log.logSegments.size())
+ assertEquals(0, log.logStartOffset)
+ assertEquals(31, log.localLogStartOffset())
+
+ log.updateLogStartOffsetFromRemoteTier(15)
+ assertEquals(15, log.logStartOffset)
+
+ // case-1: offset is higher than the local-log-start-offset.
+ // log-start-offset < local-log-start-offset < offset-to-be-converted < log-end-offset
+ assertEquals(new LogOffsetMetadata(35, 31, 288), log.maybeConvertToOffsetMetadata(35))
+ // case-2: offset is less than the local-log-start-offset
+ // log-start-offset < offset-to-be-converted < local-log-start-offset < log-end-offset
+ assertEquals(new LogOffsetMetadata(29, -1L, -1), log.maybeConvertToOffsetMetadata(29))
+ // case-3: offset is higher than the log-end-offset
+ // log-start-offset < local-log-start-offset < log-end-offset < offset-to-be-converted
+ assertEquals(new LogOffsetMetadata(log.logEndOffset + 1, -1L, -1), log.maybeConvertToOffsetMetadata(log.logEndOffset + 1))
+ // case-4: offset is less than the log-start-offset
+ // offset-to-be-converted < log-start-offset < local-log-start-offset < log-end-offset
+ assertEquals(new LogOffsetMetadata(14, -1L, -1), log.maybeConvertToOffsetMetadata(14))
+ }
+
private def appendTransactionalToBuffer(buffer: ByteBuffer,
producerId: Long,
producerEpoch: Short,
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogOffsetMetadata.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogOffsetMetadata.java
index 97e96928b2c..71f5f67c972 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogOffsetMetadata.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogOffsetMetadata.java
@@ -28,11 +28,10 @@ public final class LogOffsetMetadata {
//TODO KAFKA-14484 remove once UnifiedLog has been moved to the storage module
private static final long UNIFIED_LOG_UNKNOWN_OFFSET = -1L;
-
- public static final LogOffsetMetadata UNKNOWN_OFFSET_METADATA = new LogOffsetMetadata(-1L, 0L, 0);
-
private static final int UNKNOWN_FILE_POSITION = -1;
+ public static final LogOffsetMetadata UNKNOWN_OFFSET_METADATA = new LogOffsetMetadata(-1L, UNIFIED_LOG_UNKNOWN_OFFSET, UNKNOWN_FILE_POSITION);
+
public final long messageOffset;
public final long segmentBaseOffset;
public final int relativePositionInSegment;
@@ -51,21 +50,22 @@ public final class LogOffsetMetadata {
// check if this offset is already on an older segment compared with the given offset
public boolean onOlderSegment(LogOffsetMetadata that) {
- if (messageOffsetOnly())
- throw new KafkaException(this + " cannot compare its segment info with " + that + " since it only has message offset info");
-
+ if (messageOffsetOnly() || that.messageOffsetOnly())
+ return false;
return this.segmentBaseOffset < that.segmentBaseOffset;
}
// check if this offset is on the same segment with the given offset
- private boolean onSameSegment(LogOffsetMetadata that) {
+ public boolean onSameSegment(LogOffsetMetadata that) {
+ if (messageOffsetOnly() || that.messageOffsetOnly())
+ return false;
return this.segmentBaseOffset == that.segmentBaseOffset;
}
// compute the number of bytes between this offset to the given offset
// if they are on the same segment and this offset precedes the given offset
public int positionDiff(LogOffsetMetadata that) {
- if (messageOffsetOnly())
+ if (messageOffsetOnly() || that.messageOffsetOnly())
throw new KafkaException(this + " cannot compare its segment position with " + that + " since it only has message offset info");
if (!onSameSegment(that))
throw new KafkaException(this + " cannot compare its segment position with " + that + " since they are not on the same segment");
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
index 104eb9a18b3..100f8f303d8 100644
--- 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
@@ -398,7 +398,7 @@ public class LogSegment implements Closeable {
/**
* Equivalent to {@code read(startOffset, maxSize, size())}.
*
- * See {@link #read(long, int, long, boolean)} for details.
+ * See {@link #read(long, int, Optional, boolean)} for details.
*/
public FetchDataInfo read(long startOffset, int maxSize) throws IOException {
return read(startOffset, maxSize, size());
@@ -407,10 +407,10 @@ public class LogSegment implements Closeable {
/**
* Equivalent to {@code read(startOffset, maxSize, maxPosition, false)}.
*
- * See {@link #read(long, int, long, boolean)} for details.
+ * See {@link #read(long, int, Optional, boolean)} for details.
*/
public FetchDataInfo read(long startOffset, int maxSize, long maxPosition) throws IOException {
- return read(startOffset, maxSize, maxPosition, false);
+ return read(startOffset, maxSize, Optional.of(maxPosition), false);
}
/**
@@ -421,13 +421,13 @@ public class LogSegment implements Closeable {
*
* @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 maxPositionOpt 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 {
+ public FetchDataInfo read(long startOffset, int maxSize, Optional<Long> maxPositionOpt, boolean minOneMessage) throws IOException {
if (maxSize < 0)
throw new IllegalArgumentException("Invalid max size " + maxSize + " for log read from segment " + log);
@@ -444,12 +444,14 @@ public class LogSegment implements Closeable {
if (minOneMessage)
adjustedMaxSize = Math.max(maxSize, startOffsetAndSize.size);
- // return a log segment but with zero size in the case below
- if (adjustedMaxSize == 0)
+ // return empty records in the fetch-data-info when:
+ // 1. adjustedMaxSize is 0 (or)
+ // 2. maxPosition to read is unavailable
+ if (adjustedMaxSize == 0 || !maxPositionOpt.isPresent())
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);
+ int fetchSize = Math.min((int) (maxPositionOpt.get() - startPosition), adjustedMaxSize);
return new FetchDataInfo(offsetMetadata, log.slice(startPosition, fetchSize),
adjustedMaxSize < startOffsetAndSize.size, Optional.empty());
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/LogOffsetMetadataTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogOffsetMetadataTest.java
new file mode 100644
index 00000000000..0f1da1e4153
--- /dev/null
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogOffsetMetadataTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.KafkaException;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.kafka.storage.internals.log.LogOffsetMetadata.UNKNOWN_OFFSET_METADATA;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class LogOffsetMetadataTest {
+
+ @Test
+ void testOnOlderSegment() {
+ LogOffsetMetadata metadata1 = new LogOffsetMetadata(1L, 0L, 1);
+ LogOffsetMetadata metadata2 = new LogOffsetMetadata(5L, 4L, 2);
+ LogOffsetMetadata messageOnlyMetadata = new LogOffsetMetadata(1L);
+ assertFalse(UNKNOWN_OFFSET_METADATA.onOlderSegment(UNKNOWN_OFFSET_METADATA));
+ assertFalse(metadata1.onOlderSegment(messageOnlyMetadata));
+ assertFalse(messageOnlyMetadata.onOlderSegment(metadata1));
+ assertFalse(metadata1.onOlderSegment(metadata1));
+ assertFalse(metadata2.onOlderSegment(metadata1));
+ assertTrue(metadata1.onOlderSegment(metadata2));
+ }
+
+ @Test
+ void testPositionDiff() {
+ LogOffsetMetadata metadata1 = new LogOffsetMetadata(1L);
+ LogOffsetMetadata metadata2 = new LogOffsetMetadata(5L, 0L, 5);
+ KafkaException exception = assertThrows(KafkaException.class, () -> metadata1.positionDiff(metadata2));
+ assertTrue(exception.getMessage().endsWith("since it only has message offset info"));
+
+ exception = assertThrows(KafkaException.class, () -> metadata2.positionDiff(metadata1));
+ assertTrue(exception.getMessage().endsWith("since it only has message offset info"));
+
+ LogOffsetMetadata metadata3 = new LogOffsetMetadata(15L, 10L, 5);
+ exception = assertThrows(KafkaException.class, () -> metadata3.positionDiff(metadata2));
+ assertTrue(exception.getMessage().endsWith("since they are not on the same segment"));
+
+ LogOffsetMetadata metadata4 = new LogOffsetMetadata(40L, 10L, 100);
+ assertEquals(95, metadata4.positionDiff(metadata3));
+ }
+
+ @Test
+ void testMessageOffsetOnly() {
+ LogOffsetMetadata metadata1 = new LogOffsetMetadata(1L);
+ LogOffsetMetadata metadata2 = new LogOffsetMetadata(1L, 0L, 1);
+ assertTrue(UNKNOWN_OFFSET_METADATA.messageOffsetOnly());
+ assertFalse(metadata2.messageOffsetOnly());
+ assertTrue(metadata1.messageOffsetOnly());
+ }
+
+ @Test
+ void testOnSameSegment() {
+ LogOffsetMetadata metadata1 = new LogOffsetMetadata(1L, 0L, 1);
+ LogOffsetMetadata metadata2 = new LogOffsetMetadata(5L, 4L, 2);
+ LogOffsetMetadata metadata3 = new LogOffsetMetadata(10L, 4L, 200);
+ assertFalse(metadata1.onSameSegment(metadata2));
+ assertTrue(metadata2.onSameSegment(metadata3));
+
+ LogOffsetMetadata metadata4 = new LogOffsetMetadata(50);
+ LogOffsetMetadata metadata5 = new LogOffsetMetadata(100);
+ assertFalse(metadata4.onSameSegment(metadata5));
+ }
+}
\ No newline at end of file
diff --git a/storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestBuilder.java b/storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestBuilder.java
index 6272d0e6863..dff637cf85e 100644
--- a/storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestBuilder.java
+++ b/storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestBuilder.java
@@ -55,6 +55,7 @@ import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.server.log.remote.storage.LocalTieredStorageEvent;
import org.apache.kafka.storage.internals.log.EpochEntry;
+import java.io.FilenameFilter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -290,6 +291,13 @@ public final class TieredStorageTestBuilder {
return this;
}
+ public TieredStorageTestBuilder eraseBrokerStorage(Integer brokerId,
+ FilenameFilter filenameFilter,
+ boolean isStopped) {
+ actions.add(new EraseBrokerStorageAction(brokerId, filenameFilter, isStopped));
+ return this;
+ }
+
public TieredStorageTestBuilder expectEmptyRemoteStorage(String topic,
Integer partition) {
TopicPartition topicPartition = new TopicPartition(topic, partition);
diff --git a/storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestContext.java b/storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestContext.java
index b3e2312145d..059e33f3525 100644
--- a/storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestContext.java
+++ b/storage/src/test/java/org/apache/kafka/tiered/storage/TieredStorageTestContext.java
@@ -49,6 +49,7 @@ import org.apache.kafka.server.log.remote.storage.LocalTieredStorageSnapshot;
import scala.Function0;
import scala.Function1;
+import java.io.FilenameFilter;
import java.io.IOException;
import java.io.PrintStream;
import java.util.ArrayList;
@@ -259,8 +260,20 @@ public final class TieredStorageTestContext implements AutoCloseable {
initContext();
}
- public void eraseBrokerStorage(int brokerId) throws IOException {
- localStorages.get(brokerId).eraseStorage();
+ public void eraseBrokerStorage(int brokerId,
+ FilenameFilter filter,
+ boolean isStopped) throws IOException {
+ BrokerLocalStorage brokerLocalStorage;
+ if (isStopped) {
+ brokerLocalStorage = TieredStorageTestHarness.localStorages(harness.brokers())
+ .stream()
+ .filter(bls -> bls.getBrokerId() == brokerId)
+ .findFirst()
+ .orElseThrow(() -> new IllegalArgumentException("No local storage found for broker " + brokerId));
+ } else {
+ brokerLocalStorage = localStorages.get(brokerId);
+ }
+ brokerLocalStorage.eraseStorage(filter);
}
public TopicSpec topicSpec(String topicName) {
diff --git a/storage/src/test/java/org/apache/kafka/tiered/storage/actions/EraseBrokerStorageAction.java b/storage/src/test/java/org/apache/kafka/tiered/storage/actions/EraseBrokerStorageAction.java
index 3514630cb82..1439a4b2670 100644
--- a/storage/src/test/java/org/apache/kafka/tiered/storage/actions/EraseBrokerStorageAction.java
+++ b/storage/src/test/java/org/apache/kafka/tiered/storage/actions/EraseBrokerStorageAction.java
@@ -19,24 +19,36 @@ package org.apache.kafka.tiered.storage.actions;
import org.apache.kafka.tiered.storage.TieredStorageTestAction;
import org.apache.kafka.tiered.storage.TieredStorageTestContext;
+import java.io.FilenameFilter;
import java.io.IOException;
import java.io.PrintStream;
public final class EraseBrokerStorageAction implements TieredStorageTestAction {
private final int brokerId;
+ private final FilenameFilter filenameFilter;
+ private final boolean isStopped;
public EraseBrokerStorageAction(int brokerId) {
+ this(brokerId, (dir, name) -> true, false);
+ }
+
+ public EraseBrokerStorageAction(int brokerId,
+ FilenameFilter filenameFilter,
+ boolean isStopped) {
this.brokerId = brokerId;
+ this.filenameFilter = filenameFilter;
+ this.isStopped = isStopped;
}
@Override
public void doExecute(TieredStorageTestContext context) throws IOException {
- context.eraseBrokerStorage(brokerId);
+ context.eraseBrokerStorage(brokerId, filenameFilter, isStopped);
}
@Override
public void describe(PrintStream output) {
- output.println("erase-broker-storage: " + brokerId);
+ output.println("erase-broker-storage: " + brokerId + ", isStopped: " + isStopped +
+ ", filenameFilter: " + filenameFilter);
}
}
diff --git a/storage/src/test/java/org/apache/kafka/tiered/storage/integration/FetchFromLeaderWithCorruptedCheckpointTest.java b/storage/src/test/java/org/apache/kafka/tiered/storage/integration/FetchFromLeaderWithCorruptedCheckpointTest.java
new file mode 100644
index 00000000000..556977e5497
--- /dev/null
+++ b/storage/src/test/java/org/apache/kafka/tiered/storage/integration/FetchFromLeaderWithCorruptedCheckpointTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.tiered.storage.integration;
+
+import kafka.log.LogManager;
+import kafka.server.ReplicaManager;
+import org.apache.kafka.storage.internals.checkpoint.CleanShutdownFileHandler;
+import org.apache.kafka.tiered.storage.TieredStorageTestBuilder;
+import org.apache.kafka.tiered.storage.TieredStorageTestHarness;
+import org.apache.kafka.tiered.storage.specs.KeyValueSpec;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.kafka.common.utils.Utils.mkEntry;
+import static org.apache.kafka.common.utils.Utils.mkMap;
+
+public class FetchFromLeaderWithCorruptedCheckpointTest extends TieredStorageTestHarness {
+
+ @Override
+ public int brokerCount() {
+ return 2;
+ }
+
+ @Override
+ protected void writeTestSpecifications(TieredStorageTestBuilder builder) {
+ final Integer broker0 = 0;
+ final Integer broker1 = 1;
+ final String topicA = "topicA";
+ final Integer p0 = 0;
+ final Integer partitionCount = 1;
+ final Integer replicationFactor = 2;
+ final Integer maxBatchCountPerSegment = 1;
+ final boolean enableRemoteLogStorage = true;
+ final Map<Integer, List<Integer>> assignment = mkMap(mkEntry(p0, Arrays.asList(broker0, broker1)));
+ final List<String> checkpointFiles = Arrays.asList(
+ ReplicaManager.HighWatermarkFilename(),
+ LogManager.RecoveryPointCheckpointFile(),
+ CleanShutdownFileHandler.CLEAN_SHUTDOWN_FILE_NAME);
+
+ builder.createTopic(topicA, partitionCount, replicationFactor, maxBatchCountPerSegment, assignment,
+ enableRemoteLogStorage)
+ // send records to partition 0
+ .expectSegmentToBeOffloaded(broker0, topicA, p0, 0, new KeyValueSpec("k0", "v0"))
+ .expectSegmentToBeOffloaded(broker0, topicA, p0, 1, new KeyValueSpec("k1", "v1"))
+ .expectEarliestLocalOffsetInLogDirectory(topicA, p0, 2L)
+ .produce(topicA, p0, new KeyValueSpec("k0", "v0"), new KeyValueSpec("k1", "v1"),
+ new KeyValueSpec("k2", "v2"))
+ .expectFetchFromTieredStorage(broker0, topicA, p0, 2)
+ .consume(topicA, p0, 0L, 3, 2)
+ // shutdown the brokers
+ .stop(broker1)
+ .stop(broker0)
+ // delete the checkpoint files
+ .eraseBrokerStorage(broker0, (dir, name) -> checkpointFiles.contains(name), true)
+ // start the broker first whose checkpoint files were deleted.
+ .start(broker0)
+ .start(broker1)
+ // send some records to partition 0
+ // Note that the segment 2 gets offloaded for p0, but we cannot expect those events deterministically
+ // because the rlm-task-thread runs in background and this framework doesn't support it.
+ .expectSegmentToBeOffloaded(broker0, topicA, p0, 3, new KeyValueSpec("k3", "v3"))
+ .expectEarliestLocalOffsetInLogDirectory(topicA, p0, 4L)
+ .produce(topicA, p0, new KeyValueSpec("k3", "v3"), new KeyValueSpec("k4", "v4"))
+ .expectFetchFromTieredStorage(broker0, topicA, p0, 4)
+ .consume(topicA, p0, 0L, 5, 4);
+
+ }
+}
diff --git a/storage/src/test/java/org/apache/kafka/tiered/storage/utils/BrokerLocalStorage.java b/storage/src/test/java/org/apache/kafka/tiered/storage/utils/BrokerLocalStorage.java
index 2d26e43f8d7..e303a25a5d3 100644
--- a/storage/src/test/java/org/apache/kafka/tiered/storage/utils/BrokerLocalStorage.java
+++ b/storage/src/test/java/org/apache/kafka/tiered/storage/utils/BrokerLocalStorage.java
@@ -25,6 +25,7 @@ import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.storage.internals.log.LogFileUtils;
import java.io.File;
+import java.io.FilenameFilter;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
@@ -156,8 +157,8 @@ public final class BrokerLocalStorage {
return false;
}
- public void eraseStorage() throws IOException {
- for (File file : Objects.requireNonNull(brokerStorageDirectory.listFiles())) {
+ public void eraseStorage(FilenameFilter filter) throws IOException {
+ for (File file : Objects.requireNonNull(brokerStorageDirectory.listFiles(filter))) {
Utils.delete(file);
}
}