junrao commented on code in PR #23780:
URL: https://github.com/apache/kafka/pull/23780#discussion_r4298114868
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java:
##########
@@ -1770,10 +1785,11 @@ public OffsetResultHolder fetchOffsetByTimestamp(long
targetTimestamp, Optional<
AsyncOffsetReadFutureHolder<OffsetResultHolder.FileRecordsOrError>
asyncOffsetReadFutureHolder =
remoteOffsetReader.get().asyncOffsetRead(topicPartition(), targetTimestamp,
- logStartOffset, leaderEpochCache, () ->
searchOffsetInLocalLog(targetTimestamp, localLogStartOffset()));
+ logStartOffset, leaderEpochCache,
+ () ->
searchOffsetInLocalLog(targetTimestamp, localLogStartOffset(),
lastFetchableOffset));
Review Comment:
indentation
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java:
##########
@@ -750,12 +750,121 @@ public long getFirstBatchTimestamp() {
* @return the timestamp and offset of the first message that meets the
requirements. Empty will be returned if there is no such message.
*/
public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long
timestampMs, long startingOffset, int maxRecordBodySize) throws IOException {
+ return findOffsetByTimestamp(timestampMs, startingOffset,
Long.MAX_VALUE, maxRecordBodySize);
+ }
+
+ /**
+ * Search for the first message whose timestamp is at least {@code
timestampMs} and whose offset is at least
+ * {@code startingOffset} and strictly less than {@code
exclusiveEndOffset}.
+ * {@code exclusiveEndOffset} is a batch boundary, such as the high
watermark or last stable offset.
+ * Pass {@link Long#MAX_VALUE} to search the whole segment.
+ */
+ public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long
timestampMs, long startingOffset,
+ long
exclusiveEndOffset, int maxRecordBodySize) throws IOException {
+ if (exclusiveEndOffset <= baseOffset)
+ return Optional.empty();
+
// Get the index entry with a timestamp less than or equal to the
target timestamp
TimestampOffset timestampOffset = timeIndex().lookup(timestampMs);
int position = offsetIndex().lookup(Math.max(timestampOffset.offset(),
startingOffset)).position();
+ int endPosition = endPosition(exclusiveEndOffset);
+ if (position >= endPosition)
+ return Optional.empty();
+
+ return Optional.ofNullable(log.searchForTimestamp(timestampMs,
position, endPosition, startingOffset, maxRecordBodySize));
+ }
+
+ /**
+ * Largest timestamp in this segment, and the offset of the record that
holds it.
+ * Empty when the segment has no batches or the batch does not expose a
timestamp offset.
+ */
+ public Optional<FileRecords.TimestampAndOffset>
fetchMaxTimestampAndOffset(int maxRecordBodySize) throws IOException {
Review Comment:
Could this be private?
##########
clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java:
##########
@@ -348,7 +348,21 @@ public LogOffsetPosition searchForOffsetFromPosition(long
targetOffset, int star
*/
public TimestampAndOffset searchForTimestamp(long targetTimestamp, int
startingPosition, long startingOffset,
int maxRecordBodySize) {
- for (RecordBatch batch : batchesFrom(startingPosition)) {
+ int endPosition = isSlice ? end : sizeInBytes();
Review Comment:
Could we add a comment that this searchForTimestamp() is for test only?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java:
##########
@@ -1694,6 +1694,16 @@ public List<AbortedTxn> collectAbortedTransactions(long
startOffset, long upperB
* </ul>
*/
public OffsetResultHolder fetchOffsetByTimestamp(long targetTimestamp,
Optional<AsyncOffsetReader> remoteOffsetReader) {
+ return fetchOffsetByTimestamp(targetTimestamp, remoteOffsetReader,
Long.MAX_VALUE);
+ }
+
+ /**
+ * Same as {@link #fetchOffsetByTimestamp(long, Optional)} but ignores
offsets at or after {@code lastFetchableOffset}.
Review Comment:
This method ignores offsets after lastFetchableOffset only for maxTimestamp.
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java:
##########
@@ -750,12 +750,121 @@ public long getFirstBatchTimestamp() {
* @return the timestamp and offset of the first message that meets the
requirements. Empty will be returned if there is no such message.
*/
public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long
timestampMs, long startingOffset, int maxRecordBodySize) throws IOException {
+ return findOffsetByTimestamp(timestampMs, startingOffset,
Long.MAX_VALUE, maxRecordBodySize);
+ }
+
+ /**
+ * Search for the first message whose timestamp is at least {@code
timestampMs} and whose offset is at least
+ * {@code startingOffset} and strictly less than {@code
exclusiveEndOffset}.
+ * {@code exclusiveEndOffset} is a batch boundary, such as the high
watermark or last stable offset.
+ * Pass {@link Long#MAX_VALUE} to search the whole segment.
+ */
+ public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long
timestampMs, long startingOffset,
+ long
exclusiveEndOffset, int maxRecordBodySize) throws IOException {
+ if (exclusiveEndOffset <= baseOffset)
+ return Optional.empty();
+
// Get the index entry with a timestamp less than or equal to the
target timestamp
TimestampOffset timestampOffset = timeIndex().lookup(timestampMs);
int position = offsetIndex().lookup(Math.max(timestampOffset.offset(),
startingOffset)).position();
+ int endPosition = endPosition(exclusiveEndOffset);
+ if (position >= endPosition)
+ return Optional.empty();
+
+ return Optional.ofNullable(log.searchForTimestamp(timestampMs,
position, endPosition, startingOffset, maxRecordBodySize));
+ }
+
+ /**
+ * Largest timestamp in this segment, and the offset of the record that
holds it.
+ * Empty when the segment has no batches or the batch does not expose a
timestamp offset.
+ */
+ public Optional<FileRecords.TimestampAndOffset>
fetchMaxTimestampAndOffset(int maxRecordBodySize) throws IOException {
+ if (size() == 0)
+ return Optional.empty();
+
+ TimestampOffset maxTimestampSoFar = readMaxTimestampAndOffsetSoFar();
+ int position =
offsetIndex().lookup(maxTimestampSoFar.offset()).position();
+ for (RecordBatch batch : log.batchesFrom(position)) {
+ if (batch.maxTimestamp() == maxTimestampSoFar.timestamp())
+ return toTimestampAndOffset(batch, maxRecordBodySize);
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Largest timestamp among batches whose base offset is strictly less than
{@code exclusiveEndOffset}.
+ * The high watermark and last stable offset are batch boundaries, so a
batch is either entirely included or not.
+ * When this segment's own max timestamp is already below the bound, that
value is used. Otherwise the time index
+ * names the last indexed timestamp below the bound and the records after
that entry are scanned up to the bound.
+ */
+ public Optional<FileRecords.TimestampAndOffset> maxTimestampBelow(long
exclusiveEndOffset, int maxRecordBodySize) throws IOException {
+ if (exclusiveEndOffset <= baseOffset || size() == 0)
+ return Optional.empty();
+
+ TimestampOffset segmentMax = readMaxTimestampAndOffsetSoFar();
+ if (segmentMax.offset() < exclusiveEndOffset)
+ return fetchMaxTimestampAndOffset(maxRecordBodySize);
Review Comment:
We call readMaxTimestampAndOffsetSoFar() again inside
fetchMaxTimestampAndOffset(), at that point, maxTimestamp could have moved to
>= exclusiveEndOffset. It's safer to pass in segmentMax.
##########
clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java:
##########
@@ -434,7 +456,12 @@ private AbstractIterator<FileChannelRecordBatch>
batchIterator(int start) {
end = this.end;
else
end = this.sizeInBytes();
- FileLogInputStream inputStream = new FileLogInputStream(this, start,
end);
+ return batchIterator(start, end);
+ }
+
+ private AbstractIterator<FileChannelRecordBatch> batchIterator(int start,
int end) {
+ int boundedEnd = isSlice ? Math.min(end, this.end) : end;
Review Comment:
We probably want to bound the end for non-sliced FileRecords too. Claude
suggested the following.
`Math.min(end, isSlice ? this.end : sizeInBytes())`
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java:
##########
@@ -750,12 +750,121 @@ public long getFirstBatchTimestamp() {
* @return the timestamp and offset of the first message that meets the
requirements. Empty will be returned if there is no such message.
*/
public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long
timestampMs, long startingOffset, int maxRecordBodySize) throws IOException {
+ return findOffsetByTimestamp(timestampMs, startingOffset,
Long.MAX_VALUE, maxRecordBodySize);
+ }
+
+ /**
+ * Search for the first message whose timestamp is at least {@code
timestampMs} and whose offset is at least
+ * {@code startingOffset} and strictly less than {@code
exclusiveEndOffset}.
+ * {@code exclusiveEndOffset} is a batch boundary, such as the high
watermark or last stable offset.
+ * Pass {@link Long#MAX_VALUE} to search the whole segment.
+ */
+ public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long
timestampMs, long startingOffset,
+ long
exclusiveEndOffset, int maxRecordBodySize) throws IOException {
+ if (exclusiveEndOffset <= baseOffset)
+ return Optional.empty();
+
// Get the index entry with a timestamp less than or equal to the
target timestamp
TimestampOffset timestampOffset = timeIndex().lookup(timestampMs);
int position = offsetIndex().lookup(Math.max(timestampOffset.offset(),
startingOffset)).position();
+ int endPosition = endPosition(exclusiveEndOffset);
+ if (position >= endPosition)
+ return Optional.empty();
+
+ return Optional.ofNullable(log.searchForTimestamp(timestampMs,
position, endPosition, startingOffset, maxRecordBodySize));
+ }
+
+ /**
+ * Largest timestamp in this segment, and the offset of the record that
holds it.
+ * Empty when the segment has no batches or the batch does not expose a
timestamp offset.
+ */
+ public Optional<FileRecords.TimestampAndOffset>
fetchMaxTimestampAndOffset(int maxRecordBodySize) throws IOException {
+ if (size() == 0)
+ return Optional.empty();
+
+ TimestampOffset maxTimestampSoFar = readMaxTimestampAndOffsetSoFar();
+ int position =
offsetIndex().lookup(maxTimestampSoFar.offset()).position();
+ for (RecordBatch batch : log.batchesFrom(position)) {
+ if (batch.maxTimestamp() == maxTimestampSoFar.timestamp())
+ return toTimestampAndOffset(batch, maxRecordBodySize);
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Largest timestamp among batches whose base offset is strictly less than
{@code exclusiveEndOffset}.
+ * The high watermark and last stable offset are batch boundaries, so a
batch is either entirely included or not.
+ * When this segment's own max timestamp is already below the bound, that
value is used. Otherwise the time index
+ * names the last indexed timestamp below the bound and the records after
that entry are scanned up to the bound.
+ */
+ public Optional<FileRecords.TimestampAndOffset> maxTimestampBelow(long
exclusiveEndOffset, int maxRecordBodySize) throws IOException {
+ if (exclusiveEndOffset <= baseOffset || size() == 0)
+ return Optional.empty();
+
+ TimestampOffset segmentMax = readMaxTimestampAndOffsetSoFar();
+ if (segmentMax.offset() < exclusiveEndOffset)
+ return fetchMaxTimestampAndOffset(maxRecordBodySize);
+
+ TimestampOffset indexed =
timeIndex().lastEntryWithOffsetBelow(exclusiveEndOffset);
+ int scanStart = offsetIndex().lookup(indexed.offset()).position();
+ int scanEnd = endPosition(exclusiveEndOffset);
+
+ RecordBatch bestBatch = null;
+ long bestTimestamp = RecordBatch.NO_TIMESTAMP;
+ for (RecordBatch batch : log.batchesFrom(scanStart, scanEnd)) {
Review Comment:
Could we reuse FileRecords.largestTimestampAfter with an end-position
overload?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java:
##########
@@ -1738,6 +1748,11 @@ public OffsetResultHolder fetchOffsetByTimestamp(long
targetTimestamp, Optional<
} else if (targetTimestamp ==
ListOffsetsRequest.EARLIEST_PENDING_UPLOAD_TIMESTAMP) {
return
fetchEarliestPendingUploadOffset(remoteOffsetReader);
} else if (targetTimestamp ==
ListOffsetsRequest.MAX_TIMESTAMP) {
+ // A bound below the log end means the segment with
the global max timestamp may hold that
+ // max past the high watermark. Search only the
fetchable prefix. See KAFKA-21204.
+ if (lastFetchableOffset < logEndOffset())
Review Comment:
This check is not very reliable since immediately after this call,
logEndOffset could advance.
##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -1464,7 +1463,7 @@ class Partition(val topicPartition: TopicPartition,
def getOffsetByTimestamp: OffsetResultHolder = {
logManager.getLog(topicPartition)
- .map(log => log.fetchOffsetByTimestamp(timestamp,
remoteLogManager.asInstanceOf[Option[AsyncOffsetReader]].toJava))
+ .map(log => log.fetchOffsetByTimestamp(timestamp,
remoteLogManager.asInstanceOf[Option[AsyncOffsetReader]].toJava,
lastFetchableOffset))
Review Comment:
With this change, we can get rid of the bounding by lastFetchableOffset in
ReplicaManager.fetchOffset().
```
if (resultHolder.lastFetchableOffset.isPresent &&
timestampAndOffsetOpt.offset >=
resultHolder.lastFetchableOffset.get) {
resultHolder.maybeOffsetsError.map(e => throw e)
```
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java:
##########
@@ -750,12 +750,121 @@ public long getFirstBatchTimestamp() {
* @return the timestamp and offset of the first message that meets the
requirements. Empty will be returned if there is no such message.
*/
public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long
timestampMs, long startingOffset, int maxRecordBodySize) throws IOException {
+ return findOffsetByTimestamp(timestampMs, startingOffset,
Long.MAX_VALUE, maxRecordBodySize);
+ }
+
+ /**
+ * Search for the first message whose timestamp is at least {@code
timestampMs} and whose offset is at least
+ * {@code startingOffset} and strictly less than {@code
exclusiveEndOffset}.
+ * {@code exclusiveEndOffset} is a batch boundary, such as the high
watermark or last stable offset.
+ * Pass {@link Long#MAX_VALUE} to search the whole segment.
+ */
+ public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long
timestampMs, long startingOffset,
+ long
exclusiveEndOffset, int maxRecordBodySize) throws IOException {
+ if (exclusiveEndOffset <= baseOffset)
+ return Optional.empty();
+
// Get the index entry with a timestamp less than or equal to the
target timestamp
TimestampOffset timestampOffset = timeIndex().lookup(timestampMs);
int position = offsetIndex().lookup(Math.max(timestampOffset.offset(),
startingOffset)).position();
+ int endPosition = endPosition(exclusiveEndOffset);
+ if (position >= endPosition)
+ return Optional.empty();
+
+ return Optional.ofNullable(log.searchForTimestamp(timestampMs,
position, endPosition, startingOffset, maxRecordBodySize));
+ }
+
+ /**
+ * Largest timestamp in this segment, and the offset of the record that
holds it.
+ * Empty when the segment has no batches or the batch does not expose a
timestamp offset.
+ */
+ public Optional<FileRecords.TimestampAndOffset>
fetchMaxTimestampAndOffset(int maxRecordBodySize) throws IOException {
+ if (size() == 0)
+ return Optional.empty();
+
+ TimestampOffset maxTimestampSoFar = readMaxTimestampAndOffsetSoFar();
+ int position =
offsetIndex().lookup(maxTimestampSoFar.offset()).position();
+ for (RecordBatch batch : log.batchesFrom(position)) {
+ if (batch.maxTimestamp() == maxTimestampSoFar.timestamp())
+ return toTimestampAndOffset(batch, maxRecordBodySize);
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Largest timestamp among batches whose base offset is strictly less than
{@code exclusiveEndOffset}.
+ * The high watermark and last stable offset are batch boundaries, so a
batch is either entirely included or not.
+ * When this segment's own max timestamp is already below the bound, that
value is used. Otherwise the time index
+ * names the last indexed timestamp below the bound and the records after
that entry are scanned up to the bound.
+ */
+ public Optional<FileRecords.TimestampAndOffset> maxTimestampBelow(long
exclusiveEndOffset, int maxRecordBodySize) throws IOException {
+ if (exclusiveEndOffset <= baseOffset || size() == 0)
+ return Optional.empty();
+
+ TimestampOffset segmentMax = readMaxTimestampAndOffsetSoFar();
+ if (segmentMax.offset() < exclusiveEndOffset)
+ return fetchMaxTimestampAndOffset(maxRecordBodySize);
+
+ TimestampOffset indexed =
timeIndex().lastEntryWithOffsetBelow(exclusiveEndOffset);
+ int scanStart = offsetIndex().lookup(indexed.offset()).position();
+ int scanEnd = endPosition(exclusiveEndOffset);
+
+ RecordBatch bestBatch = null;
+ long bestTimestamp = RecordBatch.NO_TIMESTAMP;
+ for (RecordBatch batch : log.batchesFrom(scanStart, scanEnd)) {
+ if (batch.baseOffset() >= exclusiveEndOffset)
+ break;
+ long timestamp = batch.maxTimestamp();
+ if (bestBatch == null || timestamp > bestTimestamp) {
+ bestTimestamp = timestamp;
+ bestBatch = batch;
+ }
+ }
+
+ // The indexed timestamp can sit on a batch that the offset-index
checkpoint starts after.
+ if (indexed.timestamp() > bestTimestamp) {
+ Optional<FileRecords.TimestampAndOffset> indexedResult =
+ timestampAndOffsetAt(indexed.offset(),
indexed.timestamp(), scanEnd, maxRecordBodySize);
+ if (indexedResult.isPresent())
+ return indexedResult;
+ }
+ if (bestBatch == null)
+ return Optional.empty();
+ return toTimestampAndOffset(bestBatch, maxRecordBodySize);
+ }
+
+ /**
+ * File position of the first batch at or after {@code
exclusiveEndOffset}, or the segment size when every
+ * batch starts below it. {@link Long#MAX_VALUE} means the whole segment.
+ */
+ private int endPosition(long exclusiveEndOffset) throws IOException {
+ if (exclusiveEndOffset == Long.MAX_VALUE)
+ return log.sizeInBytes();
+ if (exclusiveEndOffset <= baseOffset)
Review Comment:
This seems redundant since both callers check that already.
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java:
##########
@@ -1812,12 +1828,36 @@ public boolean isEmpty() {
return logStartOffset == logEndOffset();
}
- private Optional<FileRecords.TimestampAndOffset>
searchOffsetInLocalLog(long targetTimestamp, long startOffset) throws
IOException {
+ /**
+ * Largest timestamp among records with offset strictly less than {@code
exclusiveEndOffset}.
+ * Segments that end before the bound contribute their full max. The
segment that contains the bound
+ * contributes only the prefix below it, so a later uncommitted timestamp
cannot hide an earlier one.
+ */
+ private Optional<FileRecords.TimestampAndOffset> maxTimestampBelow(long
exclusiveEndOffset) throws IOException {
Review Comment:
This forces IO scan on every segment. Claude suggested the following
optimization.
```
// bound = lastFetchableOffset, or Long.MAX_VALUE for the unbounded case
for (LogSegment segment : logSegments()) {
if (segment.baseOffset() >= bound) break;
TimestampOffset segMax = segment.readMaxTimestampAndOffsetSoFar();
// cached, read once
if (segMax.offset() < bound) {
// segment's max is below the bound: compare cached values only, no
I/O
if (winnerMax == null || segMax.timestamp() > winnerMax.timestamp())
{ winner = segment; winnerMax = segMax; }
} else {
// only the segment containing the bound: scan its prefix below the
bound
boundCandidate = segment.maxTimestampBelow(bound, maxRecordBodySize);
}
}
```
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java:
##########
@@ -750,12 +750,121 @@ public long getFirstBatchTimestamp() {
* @return the timestamp and offset of the first message that meets the
requirements. Empty will be returned if there is no such message.
*/
public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long
timestampMs, long startingOffset, int maxRecordBodySize) throws IOException {
+ return findOffsetByTimestamp(timestampMs, startingOffset,
Long.MAX_VALUE, maxRecordBodySize);
+ }
+
+ /**
+ * Search for the first message whose timestamp is at least {@code
timestampMs} and whose offset is at least
+ * {@code startingOffset} and strictly less than {@code
exclusiveEndOffset}.
+ * {@code exclusiveEndOffset} is a batch boundary, such as the high
watermark or last stable offset.
+ * Pass {@link Long#MAX_VALUE} to search the whole segment.
+ */
+ public Optional<FileRecords.TimestampAndOffset> findOffsetByTimestamp(long
timestampMs, long startingOffset,
+ long
exclusiveEndOffset, int maxRecordBodySize) throws IOException {
+ if (exclusiveEndOffset <= baseOffset)
+ return Optional.empty();
+
// Get the index entry with a timestamp less than or equal to the
target timestamp
TimestampOffset timestampOffset = timeIndex().lookup(timestampMs);
int position = offsetIndex().lookup(Math.max(timestampOffset.offset(),
startingOffset)).position();
+ int endPosition = endPosition(exclusiveEndOffset);
+ if (position >= endPosition)
+ return Optional.empty();
+
+ return Optional.ofNullable(log.searchForTimestamp(timestampMs,
position, endPosition, startingOffset, maxRecordBodySize));
+ }
+
+ /**
+ * Largest timestamp in this segment, and the offset of the record that
holds it.
+ * Empty when the segment has no batches or the batch does not expose a
timestamp offset.
+ */
+ public Optional<FileRecords.TimestampAndOffset>
fetchMaxTimestampAndOffset(int maxRecordBodySize) throws IOException {
+ if (size() == 0)
+ return Optional.empty();
+
+ TimestampOffset maxTimestampSoFar = readMaxTimestampAndOffsetSoFar();
+ int position =
offsetIndex().lookup(maxTimestampSoFar.offset()).position();
+ for (RecordBatch batch : log.batchesFrom(position)) {
+ if (batch.maxTimestamp() == maxTimestampSoFar.timestamp())
+ return toTimestampAndOffset(batch, maxRecordBodySize);
+ }
+ return Optional.empty();
+ }
+
+ /**
+ * Largest timestamp among batches whose base offset is strictly less than
{@code exclusiveEndOffset}.
+ * The high watermark and last stable offset are batch boundaries, so a
batch is either entirely included or not.
+ * When this segment's own max timestamp is already below the bound, that
value is used. Otherwise the time index
+ * names the last indexed timestamp below the bound and the records after
that entry are scanned up to the bound.
+ */
+ public Optional<FileRecords.TimestampAndOffset> maxTimestampBelow(long
exclusiveEndOffset, int maxRecordBodySize) throws IOException {
+ if (exclusiveEndOffset <= baseOffset || size() == 0)
+ return Optional.empty();
+
+ TimestampOffset segmentMax = readMaxTimestampAndOffsetSoFar();
+ if (segmentMax.offset() < exclusiveEndOffset)
+ return fetchMaxTimestampAndOffset(maxRecordBodySize);
+
+ TimestampOffset indexed =
timeIndex().lastEntryWithOffsetBelow(exclusiveEndOffset);
+ int scanStart = offsetIndex().lookup(indexed.offset()).position();
+ int scanEnd = endPosition(exclusiveEndOffset);
+
+ RecordBatch bestBatch = null;
+ long bestTimestamp = RecordBatch.NO_TIMESTAMP;
+ for (RecordBatch batch : log.batchesFrom(scanStart, scanEnd)) {
+ if (batch.baseOffset() >= exclusiveEndOffset)
+ break;
+ long timestamp = batch.maxTimestamp();
+ if (bestBatch == null || timestamp > bestTimestamp) {
+ bestTimestamp = timestamp;
+ bestBatch = batch;
+ }
+ }
+
+ // The indexed timestamp can sit on a batch that the offset-index
checkpoint starts after.
+ if (indexed.timestamp() > bestTimestamp) {
Review Comment:
It seems that this is always false. The previous scan includes the batch
where indexed is in. So, bestTimestamp should always be larger or equal.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
