You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mi...@apache.org on 2026/09/15 08:01:24 UTC
(kafka) branch 4.4 updated: KAFKA-20979: Fix another gap in index metadata sync logic (#23316)
This is an automated email from the ASF dual-hosted git repository.
mimaison pushed a commit to branch 4.4
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.4 by this push:
new 72b21c706e8 KAFKA-20979: Fix another gap in index metadata sync logic (#23316)
72b21c706e8 is described below
commit 72b21c706e8bc3f9fde86d3bcfdd3e7f13bff4fd
Author: Mickael Maison <mi...@users.noreply.github.com>
AuthorDate: Tue Sep 15 09:44:18 2026 +0200
KAFKA-20979: Fix another gap in index metadata sync logic (#23316)
Addressing comments from https://github.com/apache/kafka/pull/23258
Reviewers: Chia-Ping Tsai <ch...@gmail.com>, Jun Rao
<ju...@gmail.com>
---
.../kafka/common/record/internal/FileRecords.java | 17 +-------
.../common/record/internal/FileRecordsTest.java | 2 +
.../kafka/metadata/util/BatchFileReader.java | 2 +-
.../kafka/storage/internals/log/AbstractIndex.java | 32 ++++-----------
.../kafka/storage/internals/log/LazyIndex.java | 17 --------
.../kafka/storage/internals/log/LocalLog.java | 38 +++++++++---------
.../kafka/storage/internals/log/LogManager.java | 18 ++++-----
.../kafka/storage/internals/log/LogSegment.java | 37 +++++++----------
.../kafka/storage/internals/log/LogSegments.java | 6 +--
.../kafka/storage/internals/log/UnifiedLog.java | 46 ++++++++++++++--------
.../storage/internals/log/AbstractIndexTest.java | 16 --------
.../kafka/storage/internals/log/LocalLogTest.java | 32 ++++++---------
.../kafka/storage/internals/log/LogLoaderTest.java | 2 +-
.../storage/internals/log/LogSegmentTest.java | 5 ++-
.../storage/internals/log/OffsetIndexTest.java | 1 +
.../internals/log/RemoteIndexCacheTest.java | 2 +-
.../kafka/storage/internals/log/TimeIndexTest.java | 1 +
.../storage/internals/log/UnifiedLogTest.java | 22 +++++++++--
.../org/apache/kafka/tools/DumpLogSegments.java | 18 +--------
19 files changed, 127 insertions(+), 187 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java b/clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java
index ec4580ec033..9547afc9d68 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java
@@ -210,22 +210,9 @@ public class FileRecords extends AbstractRecords implements Closeable {
}
/**
- * Close this record set
+ * Close this record set. Callers are responsible for flushing and trimming before closing.
*/
public void close() throws IOException {
- if (!channel.isOpen()) {
- return;
- }
-
- flush();
- trim();
- channel.close();
- }
-
- /**
- * Close file handlers used by the FileChannel but don't write to disk. This is used when the disk may have failed
- */
- public void closeHandlers() throws IOException {
channel.close();
}
@@ -241,7 +228,7 @@ public class FileRecords extends AbstractRecords implements Closeable {
}
/**
- * Trim file when close or roll to next file
+ * Trim file when rolling to the next segment or preparing for shutdown
*/
public void trim() throws IOException {
truncateTo(sizeInBytes());
diff --git a/clients/src/test/java/org/apache/kafka/common/record/internal/FileRecordsTest.java b/clients/src/test/java/org/apache/kafka/common/record/internal/FileRecordsTest.java
index 9082db17195..9e30143bfc5 100644
--- a/clients/src/test/java/org/apache/kafka/common/record/internal/FileRecordsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/record/internal/FileRecordsTest.java
@@ -402,6 +402,8 @@ public class FileRecordsTest {
int oldSize = fileRecords.sizeInBytes();
assertEquals(this.fileRecords.sizeInBytes(), oldPosition);
assertEquals(this.fileRecords.sizeInBytes(), oldSize);
+ fileRecords.trim();
+ fileRecords.flush();
fileRecords.close();
File tempReopen = new File(temp.getAbsolutePath());
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/util/BatchFileReader.java b/metadata/src/main/java/org/apache/kafka/metadata/util/BatchFileReader.java
index 70e5b8b8920..c24eb8b6c50 100644
--- a/metadata/src/main/java/org/apache/kafka/metadata/util/BatchFileReader.java
+++ b/metadata/src/main/java/org/apache/kafka/metadata/util/BatchFileReader.java
@@ -171,7 +171,7 @@ public final class BatchFileReader implements Iterator<BatchFileReader.BatchAndT
@Override
public void close() {
try {
- fileRecords.closeHandlers();
+ fileRecords.close();
} catch (Exception e) {
log.error("Error closing fileRecords", e);
}
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/AbstractIndex.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/AbstractIndex.java
index 4af93d52a40..cb32b8ba6b3 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/AbstractIndex.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/AbstractIndex.java
@@ -190,7 +190,7 @@ public abstract class AbstractIndex implements Closeable {
/**
* Reset the size of the memory map and the underneath file. This is used in two kinds of cases: (1) in
- * trimToValidSize() which is called at closing the segment or new segment being rolled; (2) at
+ * trimToValidSize() which is called when a segment becomes inactive (e.g. rolled or prepared for shutdown); (2) at
* loading segments from disk or truncating back to an old segment where a new log segment became active;
* we want to reset the index size to maximum index size to avoid rolling new segment.
*
@@ -198,15 +198,6 @@ public abstract class AbstractIndex implements Closeable {
* @return a boolean indicating whether the size of the memory map and the underneath file is changed or not.
*/
public boolean resize(int newSize) throws IOException {
- return resize(newSize, false);
- }
-
- /**
- * @param newSize new size of the index file
- * @param sync if true, fsync the file after resizing to ensure the size is durable
- * @return true if the index was resized, false otherwise
- */
- public boolean resize(int newSize, boolean sync) throws IOException {
return inLock(() ->
inRemapWriteLock(() -> {
int roundedNewSize = roundDownToExactMultiple(newSize, entrySize());
@@ -225,9 +216,6 @@ public abstract class AbstractIndex implements Closeable {
mmap = raf.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, roundedNewSize);
this.maxEntries = mmap.limit() / entrySize();
mmap.position(position);
- if (sync) {
- raf.getChannel().force(true);
- }
log.debug("Resized {} to {}, position is {} and limit is {}", file.getAbsolutePath(), roundedNewSize,
mmap.position(), mmap.limit());
return true;
@@ -252,12 +240,13 @@ public abstract class AbstractIndex implements Closeable {
}
/**
- * Flush the data in the index to disk
+ * Flush the data in the index and its metadata to disk
*/
- public void flush() {
+ public void flush() throws IOException {
inLock(() -> {
if (mmap != null) {
mmap.force();
+ Utils.flushFileIfExists(file.toPath());
}
});
}
@@ -270,18 +259,17 @@ public abstract class AbstractIndex implements Closeable {
* not exist
*/
public boolean deleteIfExists() throws IOException {
- closeHandler();
+ close();
return Files.deleteIfExists(file.toPath());
}
/**
* Trim this index to fit just the valid entries, deleting all trailing unwritten bytes from the file.
- * @param sync if true, fsync the file after resizing to ensure both content and size are durable
*/
- public void trimToValidSize(boolean sync) throws IOException {
+ public void trimToValidSize() throws IOException {
inLock(() -> {
if (mmap != null) {
- resize(entrySize() * entries, sync);
+ resize(entrySize() * entries);
}
});
}
@@ -294,12 +282,6 @@ public abstract class AbstractIndex implements Closeable {
}
public void close() throws IOException {
- flush(); // Ensure the index content is flushed to disk as LogSegment.close may append an entry
- trimToValidSize(true);
- closeHandler();
- }
-
- public void closeHandler() {
// On JVM, a memory mapping is typically unmapped by garbage collector.
// However, in some cases it can pause application threads(STW) for a long moment reading metadata from a physical disk.
// To prevent this, we forcefully cleanup memory mapping within proper execution which never affects API responsiveness.
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LazyIndex.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LazyIndex.java
index afc1c679c29..198b1e94e4a 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LazyIndex.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LazyIndex.java
@@ -54,7 +54,6 @@ public class LazyIndex<T extends AbstractIndex> implements Closeable {
void renameTo(File file) throws IOException;
boolean deleteIfExists() throws IOException;
void close() throws IOException;
- void closeHandler();
}
private static class IndexFile implements IndexWrapper {
@@ -95,9 +94,6 @@ public class LazyIndex<T extends AbstractIndex> implements Closeable {
@Override
public void close() { }
- @Override
- public void closeHandler() { }
-
}
private static class IndexValue<T extends AbstractIndex> implements IndexWrapper {
@@ -132,11 +128,6 @@ public class LazyIndex<T extends AbstractIndex> implements Closeable {
public void close() throws IOException {
index.close();
}
-
- @Override
- public void closeHandler() {
- index.closeHandler();
- }
}
private final Lock lock = new ReentrantLock();
@@ -224,14 +215,6 @@ public class LazyIndex<T extends AbstractIndex> implements Closeable {
}
}
- public void closeHandler() {
- lock.lock();
- try {
- indexWrapper.closeHandler();
- } finally {
- lock.unlock();
- }
- }
@SuppressWarnings("unchecked")
private T loadIndex(File file) throws IOException {
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java
index 071dd2c3f5a..36f3c3dbae1 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LocalLog.java
@@ -86,9 +86,9 @@ public class LocalLog {
private final Logger logger;
private volatile LogOffsetMetadata nextOffsetMetadata;
- // The memory mapped buffer for index files of this log will be closed with either delete() or closeHandlers()
- // After memory mapped buffer is closed, no disk IO operation should be performed for this log.
- private volatile boolean isMemoryMappedBufferClosed = false;
+ // Set to true when the log is closed with either delete(), close() or closeQuietly().
+ // After the log is closed, no disk IO operation should be performed for this log.
+ private volatile boolean isClosed = false;
// Cache value of parent directory to avoid allocations in hot paths like ReplicaManager.checkpointHighWatermarks
private volatile String parentDir;
private volatile LogConfig config;
@@ -222,9 +222,9 @@ public class LocalLog {
config = newConfig;
}
- public void checkIfMemoryMappedBufferClosed() {
- if (isMemoryMappedBufferClosed) {
- throw new KafkaStorageException("The memory mapped buffer for log of " + topicPartition + " is already closed");
+ public void checkIfClosed() {
+ if (isClosed) {
+ throw new KafkaStorageException("The log for " + topicPartition + " is already closed");
}
}
@@ -239,7 +239,7 @@ public class LocalLog {
* @param offset the offset to be updated
*/
public void markFlushed(long offset) {
- checkIfMemoryMappedBufferClosed();
+ checkIfClosed();
if (offset > recoveryPoint) {
updateRecoveryPoint(offset);
lastFlushedTime.set(time.milliseconds());
@@ -309,23 +309,25 @@ public class LocalLog {
}
/**
- * Close file handlers used by log but don't write to disk.
+ * Close the segments of the log, swallowing any exceptions.
* This is called if the log directory is offline.
*/
- public void closeHandlers() {
- segments.closeHandlers();
- isMemoryMappedBufferClosed = true;
+ public void closeQuietly() {
+ if (isClosed) return;
+ segments.closeQuietly();
+ isClosed = true;
}
/**
- * Closes the segments of the log.
+ * Close the segments of the log.
*/
public void close() {
+ if (isClosed) return;
maybeHandleIOException(
() -> "Error while closing log segments for " + topicPartition + " in dir " + dir.getParent(),
() -> {
- checkIfMemoryMappedBufferClosed();
segments.close();
+ isClosed = true;
return null;
}
);
@@ -341,8 +343,8 @@ public class LocalLog {
if (!segments.isEmpty()) {
throw new IllegalStateException("Can not delete directory when " + segments.numberOfSegments() + " segments are still present");
}
- if (!isMemoryMappedBufferClosed) {
- throw new IllegalStateException("Can not delete directory when memory mapped buffer for log of " + topicPartition + " is still open.");
+ if (!isClosed) {
+ throw new IllegalStateException("Can not delete directory when log of " + topicPartition + " is still open.");
}
Utils.delete(dir);
return null;
@@ -365,7 +367,7 @@ public class LocalLog {
toDelete -> logger.info("Deleting segments as the log has been deleted: {}", toDelete.stream()
.map(LogSegment::toString)
.collect(Collectors.joining(", "))));
- isMemoryMappedBufferClosed = true;
+ isClosed = true;
return deletableSegments;
}
);
@@ -586,7 +588,7 @@ public class LocalLog {
() -> "Error while rolling log segment for " + topicPartition + " in dir " + dir.getParent(),
() -> {
long start = time.hiResClockMs();
- checkIfMemoryMappedBufferClosed();
+ checkIfClosed();
long newOffset = Math.max(expectedNextOffset, logEndOffset());
File logFile = LogFileUtils.logFile(dir, newOffset, "");
LogSegment activeSegment = segments.activeSegment();
@@ -659,7 +661,7 @@ public class LocalLog {
() -> "Error while truncating the entire log for " + topicPartition + " in dir " + dir.getParent(),
() -> {
logger.debug("Truncate and start at offset {}", newOffset);
- checkIfMemoryMappedBufferClosed();
+ checkIfClosed();
List<LogSegment> segmentsToDelete = new ArrayList<>(segments.values());
if (!segmentsToDelete.isEmpty()) {
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogManager.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogManager.java
index f60486a9825..095e1e86560 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogManager.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogManager.java
@@ -393,7 +393,7 @@ public class LogManager {
.collect(Collectors.toUnmodifiableSet());
offlineTopicPartitions.forEach(topicPartition -> {
Optional<UnifiedLog> removedLog = removeLogAndMetrics(logs, topicPartition);
- removedLog.ifPresent(UnifiedLog::closeHandlers);
+ removedLog.ifPresent(UnifiedLog::closeQuietly);
});
return offlineTopicPartitions;
@@ -877,13 +877,11 @@ public class LogManager {
Collection<UnifiedLog> logs = logsInDir(localLogsByDir, dir).values();
- List<Runnable> jobsForDir = logs.stream().map(log -> {
+ List<Runnable> jobsForDir = logs.stream().map(log -> (Runnable) () -> {
+ log.prepareActiveSegmentForClose();
// flush the log to ensure latest possible recovery point
- return (Runnable) () -> {
- // flush the log to ensure latest possible recovery point
- log.flush(true);
- log.close();
- };
+ log.flush(true);
+ log.close();
}).toList();
jobs.put(dir, jobsForDir.stream().map(pool::submit).collect(Collectors.toList()));
@@ -1561,10 +1559,10 @@ public class LogManager {
});
destLog.newMetrics();
} catch (KafkaStorageException kse) {
- // If sourceLog's log directory is offline, we need close its handlers here.
- // handleLogDirFailure() will not close handlers of sourceLog because it has been removed from currentLogs map
+ // If sourceLog's log directory is offline, we need to quietly close it here.
+ // handleLogDirFailure() will not close sourceLog because it has been removed from currentLogs map
sourceLog.ifPresent(srcLog -> {
- srcLog.closeHandlers();
+ srcLog.closeQuietly();
srcLog.removeLogMetrics();
});
throw kse;
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 7169a9f9f0e..b1c00af5c7a 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
@@ -34,7 +34,6 @@ import com.yammer.metrics.core.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.slf4j.event.Level;
import java.io.Closeable;
import java.io.File;
@@ -521,10 +520,10 @@ public class LogSegment implements Closeable {
LOGGER.debug("Truncated {} invalid bytes at the end of segment {} during recovery", truncated, log.file().getAbsolutePath());
log.truncateTo(validBytes);
- offsetIndex().trimToValidSize(false);
+ offsetIndex().trimToValidSize();
// A normally closed segment always appends the biggest timestamp ever seen into log segment, we do this as well.
timeIndex().maybeAppend(maxTimestampSoFar(), shallowOffsetOfMaxTimestampSoFar(), true);
- timeIndex().trimToValidSize(false);
+ timeIndex().trimToValidSize();
return truncated;
}
@@ -628,16 +627,12 @@ public class LogSegment implements Closeable {
*/
public void flush() throws IOException {
try {
- LOG_FLUSH_TIMER.time(new Callable<Void>() {
- // lambdas cannot declare a more specific exception type, so we use an anonymous inner class
- @Override
- public Void call() throws IOException {
- log.flush();
- offsetIndex().flush();
- timeIndex().flush();
- txnIndex.flush();
- return null;
- }
+ LOG_FLUSH_TIMER.time((Callable<Void>) () -> {
+ log.flush();
+ offsetIndex().flush();
+ timeIndex().flush();
+ txnIndex.flush();
+ return null;
});
} catch (Exception e) {
if (e instanceof IOException)
@@ -685,8 +680,8 @@ public class LogSegment implements Closeable {
*/
public void onBecomeInactiveSegment() throws IOException {
timeIndex().maybeAppend(maxTimestampSoFar(), shallowOffsetOfMaxTimestampSoFar(), true);
- offsetIndex().trimToValidSize(false);
- timeIndex().trimToValidSize(false);
+ offsetIndex().trimToValidSize();
+ timeIndex().trimToValidSize();
log.trim();
}
@@ -768,18 +763,16 @@ public class LogSegment implements Closeable {
*/
@Override
public void close() throws IOException {
- if (maxTimestampAndOffsetSoFar != TimestampOffset.UNKNOWN)
- Utils.swallow(LOGGER, Level.WARN, "maybeAppend", () -> timeIndex().maybeAppend(maxTimestampSoFar(), shallowOffsetOfMaxTimestampSoFar(), true));
Utils.closeAll(lazyOffsetIndex, lazyTimeIndex, log, txnIndex);
}
/**
- * Close file handlers used by the log segment but don't write to disk. This is used when the disk may have failed
+ * Close the log segment, swallowing any exceptions. This is used when the disk may have failed.
*/
- void closeHandlers() {
- Utils.swallow(LOGGER, Level.WARN, "offsetIndex", lazyOffsetIndex::closeHandler);
- Utils.swallow(LOGGER, Level.WARN, "timeIndex", lazyTimeIndex::closeHandler);
- Utils.swallow(LOGGER, Level.WARN, "log", log::closeHandlers);
+ void closeQuietly() {
+ Utils.closeQuietly(lazyOffsetIndex, "offsetIndex", LOGGER);
+ Utils.closeQuietly(lazyTimeIndex, "timeIndex", LOGGER);
+ Utils.closeQuietly(log, "log", LOGGER);
Utils.closeQuietly(txnIndex, "txnIndex", LOGGER);
}
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegments.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegments.java
index 4b89d7115a4..587349513c8 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegments.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegments.java
@@ -109,11 +109,11 @@ public class LogSegments implements Closeable {
}
/**
- * Close the handlers for all segments.
+ * Close all segments, swallowing any exceptions.
*/
- public void closeHandlers() {
+ public void closeQuietly() {
for (LogSegment s : values())
- s.closeHandlers();
+ s.closeQuietly();
}
/**
diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java
index 3f49fccfb23..8331db61690 100644
--- a/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java
+++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java
@@ -605,7 +605,7 @@ public class UnifiedLog implements AutoCloseable {
* known, this will do a lookup in the index and cache the result.
*/
private LogOffsetMetadata fetchHighWatermarkMetadata() throws IOException {
- localLog.checkIfMemoryMappedBufferClosed();
+ localLog.checkIfClosed();
LogOffsetMetadata offsetMetadata = highWatermarkMetadata;
if (offsetMetadata.messageOffsetOnly()) {
synchronized (lock) {
@@ -646,7 +646,7 @@ public class UnifiedLog implements AutoCloseable {
}
private LogOffsetMetadata fetchLastStableOffsetMetadata() throws IOException {
- localLog.checkIfMemoryMappedBufferClosed();
+ localLog.checkIfClosed();
// cache the current high watermark and the first unstable offset metadata to avoid a concurrent update
// invalidating the range check breaking the isPresent check
@@ -825,7 +825,7 @@ public class UnifiedLog implements AutoCloseable {
// free of all side effects, i.e. it must not update any log-specific state.
private void rebuildProducerState(long lastOffset, ProducerStateManager producerStateManager) throws IOException {
synchronized (lock) {
- localLog.checkIfMemoryMappedBufferClosed();
+ localLog.checkIfClosed();
UnifiedLog.rebuildProducerState(producerStateManager, localLog.segments(), logStartOffset, lastOffset, time(), false, logIdent);
}
}
@@ -944,9 +944,22 @@ public class UnifiedLog implements AutoCloseable {
return localLog.segments().numberOfSegments();
}
+ /**
+ * Append the largest time index entry to the time index of the active segment and trim the log and indexes.
+ * This is the same operation performed when rolling a segment and should be called before flushing and
+ * closing the log during shutdown.
+ */
+ public void prepareActiveSegmentForClose() {
+ maybeHandleIOException(
+ () -> "Error while preparing active segment for close for " + topicPartition() + " in dir " + dir().getParent(),
+ () -> {
+ localLog.segments().activeSegment().onBecomeInactiveSegment();
+ return null;
+ });
+ }
+
/**
* Close this log.
- * The memory mapped buffer for index files of this log will be left open until the log is deleted.
*/
@Override
public void close() {
@@ -954,7 +967,7 @@ public class UnifiedLog implements AutoCloseable {
synchronized (lock) {
logOffsetsListener = LogOffsetsListener.NO_OP_OFFSETS_LISTENER;
maybeFlushMetadataFile();
- localLog.checkIfMemoryMappedBufferClosed();
+ localLog.checkIfClosed();
producerExpireCheck.cancel(true);
maybeHandleIOException(
() -> "Error while taking producer state snapshot for " + topicPartition() + " in dir " + dir().getParent(),
@@ -1002,12 +1015,12 @@ public class UnifiedLog implements AutoCloseable {
}
/**
- * Close file handlers used by this log but don't write to disk. This is called if the log directory is offline
+ * Close the log, swallowing any exceptions. This is called if the log directory is offline.
*/
- public void closeHandlers() {
- logger.debug("Closing handlers");
+ public void closeQuietly() {
+ logger.debug("Closing quietly");
synchronized (lock) {
- localLog.closeHandlers();
+ localLog.closeQuietly();
}
}
@@ -1139,7 +1152,7 @@ public class UnifiedLog implements AutoCloseable {
() -> "Error while appending records to " + topicPartition() + " in dir " + dir().getParent(),
() -> {
MemoryRecords validRecords = trimmedRecords;
- localLog.checkIfMemoryMappedBufferClosed();
+ localLog.checkIfClosed();
if (validateAndAssignOffsets) {
// assign offsets to the message set
PrimitiveRef.LongRef offset = PrimitiveRef.ofLong(localLog.logEndOffset());
@@ -1313,7 +1326,7 @@ public class UnifiedLog implements AutoCloseable {
private void maybeIncrementFirstUnstableOffset() throws IOException {
synchronized (lock) {
- localLog.checkIfMemoryMappedBufferClosed();
+ localLog.checkIfClosed();
Optional<LogOffsetMetadata> updatedFirstUnstableOffset = producerStateManager.firstUnstableOffset();
if (updatedFirstUnstableOffset.isPresent() &&
@@ -1366,7 +1379,7 @@ public class UnifiedLog implements AutoCloseable {
localLogStartOffset = Math.max(newLogStartOffset, localLogStartOffset());
}
- localLog.checkIfMemoryMappedBufferClosed();
+ localLog.checkIfClosed();
if (newLogStartOffset > logStartOffset) {
updateLogStartOffset(newLogStartOffset);
logger.info("Incremented log start offset to {} due to {}", newLogStartOffset, reason);
@@ -1941,7 +1954,7 @@ public class UnifiedLog implements AutoCloseable {
segmentsToDelete = List.copyOf(deletable);
}
}
- localLog.checkIfMemoryMappedBufferClosed();
+ localLog.checkIfClosed();
if (!segmentsToDelete.isEmpty()) {
// increment the local-log-start-offset or log-start-offset before removing the segment for lookups
long newLocalLogStartOffset = localLog.segments().higherSegment(segmentsToDelete.get(segmentsToDelete.size() - 1).baseOffset()).get().baseOffset();
@@ -2290,7 +2303,6 @@ public class UnifiedLog implements AutoCloseable {
() -> "Error while deleting log for " + topicPartition() + " in dir " + dir().getParent(),
() -> {
synchronized (lock) {
- localLog.checkIfMemoryMappedBufferClosed();
producerExpireCheck.cancel(true);
leaderEpochCache.clear();
List<LogSegment> deletedSegments = localLog.deleteAllSegments();
@@ -2304,7 +2316,7 @@ public class UnifiedLog implements AutoCloseable {
// visible for testing
public void takeProducerSnapshot() throws IOException {
synchronized (lock) {
- localLog.checkIfMemoryMappedBufferClosed();
+ localLog.checkIfClosed();
producerStateManager.takeSnapshot();
}
}
@@ -2372,7 +2384,7 @@ public class UnifiedLog implements AutoCloseable {
} else {
logger.info("Truncating to offset {}", targetOffset);
synchronized (lock) {
- localLog.checkIfMemoryMappedBufferClosed();
+ localLog.checkIfClosed();
if (localLog.segments().firstSegmentBaseOffset().getAsLong() > targetOffset) {
truncateFullyAndStartAt(targetOffset, Optional.empty());
} else {
@@ -2479,7 +2491,7 @@ public class UnifiedLog implements AutoCloseable {
public void replaceSegments(List<LogSegment> newSegments, List<LogSegment> oldSegments) throws IOException {
synchronized (lock) {
- localLog.checkIfMemoryMappedBufferClosed();
+ localLog.checkIfClosed();
List<LogSegment> deletedSegments = LocalLog.replaceSegments(localLog.segments(), newSegments, oldSegments, dir(), topicPartition(),
config(), scheduler(), logDirFailureChannel(), logIdent, false);
deleteProducerSnapshots(deletedSegments, true);
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/AbstractIndexTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/AbstractIndexTest.java
index 1378577fa95..ba5af7db264 100644
--- a/storage/src/test/java/org/apache/kafka/storage/internals/log/AbstractIndexTest.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/AbstractIndexTest.java
@@ -35,18 +35,11 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
public class AbstractIndexTest {
private static class TestIndex extends AbstractIndex {
private boolean unmapInvoked = false;
- private boolean flushInvoked = false;
private MappedByteBuffer unmappedBuffer = null;
public TestIndex(File file, long baseOffset, int maxIndexSize, boolean writable) throws IOException {
super(file, baseOffset, maxIndexSize, writable);
}
- @Override
- public void flush() {
- flushInvoked = true;
- super.flush();
- }
-
@Override
protected int entrySize() {
return 1;
@@ -94,13 +87,4 @@ public class AbstractIndexTest {
assertNotSame(idx.unmappedBuffer, idx.mmap());
}
- @Test
- public void testCloseInvokeFlush() throws IOException {
- File f = new File(TestUtils.tempDirectory(), "test-index");
- TestIndex idx = new TestIndex(f, 0L, 100, true);
- assertFalse(idx.flushInvoked, "Flush should not have been invoked yet");
-
- idx.close();
- assertTrue(idx.flushInvoked, "Flush should have been invoked during close");
- }
}
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/LocalLogTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/LocalLogTest.java
index e91058e6b14..a13a5efbb84 100644
--- a/storage/src/test/java/org/apache/kafka/storage/internals/log/LocalLogTest.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/LocalLogTest.java
@@ -162,7 +162,7 @@ class LocalLogTest {
List<LogSegment> deletedSegments = log.deleteAllSegments();
assertTrue(log.segments().isEmpty());
assertEquals(segmentsBeforeDelete, deletedSegments);
- assertThrows(KafkaStorageException.class, () -> log.checkIfMemoryMappedBufferClosed());
+ assertThrows(KafkaStorageException.class, () -> log.checkIfClosed());
assertTrue(logDir.exists());
}
@@ -257,43 +257,33 @@ class LocalLogTest {
}
@Test
- public void testLogCloseSuccess() throws IOException {
- List<KeyValue> keyValues = List.of(new KeyValue("abc", "ABC"), new KeyValue("de", "DE"));
- appendRecords(kvsToRecords(keyValues), 0L);
+ public void testCloseIdempotent() {
+ log.close();
log.close();
- assertThrows(ClosedChannelException.class, () -> appendRecords(kvsToRecords(keyValues), 2L));
}
@Test
- public void testLogCloseIdempotent() {
- log.close();
- // Check that LocalLog.close() is idempotent
- log.close();
+ public void testCloseQuietlyIdempotent() {
+ log.closeQuietly();
+ log.closeQuietly();
}
@Test
- public void testLogCloseFailureWhenInMemoryBufferClosed() throws IOException {
+ public void testClosePreventsAppend() throws IOException {
List<KeyValue> keyValues = List.of(new KeyValue("abc", "ABC"), new KeyValue("de", "DE"));
appendRecords(kvsToRecords(keyValues), 0L);
- log.closeHandlers();
- assertThrows(KafkaStorageException.class, () -> log.close());
+ log.close();
+ assertThrows(ClosedChannelException.class, () -> appendRecords(kvsToRecords(keyValues), 2L));
}
@Test
- public void testLogCloseHandlers() throws IOException {
+ public void testCloseQuietlyPreventsAppend() throws IOException {
List<KeyValue> keyValues = List.of(new KeyValue("abc", "ABC"), new KeyValue("de", "DE"));
appendRecords(kvsToRecords(keyValues), 0L);
- log.closeHandlers();
+ log.closeQuietly();
assertThrows(ClosedChannelException.class, () -> appendRecords(kvsToRecords(keyValues), 2L));
}
- @Test
- public void testLogCloseHandlersIdempotent() {
- log.closeHandlers();
- // Check that LocalLog.closeHandlers() is idempotent
- log.closeHandlers();
- }
-
static class TestDeletionReason implements SegmentDeletionReason {
private Collection<LogSegment> deletedSegments = new ArrayList<>();
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/LogLoaderTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogLoaderTest.java
index 7d9c2547829..b596e859b4b 100644
--- a/storage/src/test/java/org/apache/kafka/storage/internals/log/LogLoaderTest.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogLoaderTest.java
@@ -1788,7 +1788,7 @@ public class LogLoaderTest {
log.flush(false);
assertThrows(NoSuchFileException.class, () -> log.activeSegment().sanityCheck(true));
long lastOffset = log.logEndOffset();
- log.closeHandlers();
+ log.closeQuietly();
UnifiedLog log2 = createLog(logDir, logConfig, lastOffset, false);
assertEquals(lastOffset, log2.recoveryPoint(), "Unexpected recovery point");
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/LogSegmentTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogSegmentTest.java
index f2a8829f721..2a10a81e67a 100644
--- a/storage/src/test/java/org/apache/kafka/storage/internals/log/LogSegmentTest.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/LogSegmentTest.java
@@ -294,6 +294,8 @@ public class LogSegmentTest {
try (LogSegment seg = createSegment(0L, time)) {
seg.timeIndex(); // Force load indexes before closing the segment
seg.offsetIndex();
+ // Trim indexes to simulate a segment from a previous clean shutdown
+ seg.onBecomeInactiveSegment();
seg.close();
LogSegment reopened = createSegment(0L, time);
@@ -676,8 +678,9 @@ public class LogSegmentTest {
long oldPosition = seg.log().channel().position();
long oldFileSize = seg.log().file().length();
assertEquals(512 * 1024 * 1024, oldFileSize);
+ seg.onBecomeInactiveSegment();
seg.close();
- // After close, file should be trimmed
+ // After onBecomeInactiveSegment, file should be trimmed
assertEquals(oldSize, seg.log().file().length());
LogSegment segReopen = LogSegment.open(tempDir, 40, logConfig, Time.SYSTEM,
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/OffsetIndexTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/OffsetIndexTest.java
index 578cb4ac243..6eb58b7a950 100644
--- a/storage/src/test/java/org/apache/kafka/storage/internals/log/OffsetIndexTest.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/OffsetIndexTest.java
@@ -169,6 +169,7 @@ public class OffsetIndexTest {
OffsetPosition sec = new OffsetPosition(52, 1);
index.append(first.offset(), first.position());
index.append(sec.offset(), sec.position());
+ index.trimToValidSize();
index.close();
OffsetIndex idxRo = new OffsetIndex(index.file(), index.baseOffset());
assertEquals(first, idxRo.lookup(first.offset()));
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/RemoteIndexCacheTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/RemoteIndexCacheTest.java
index dc9c07839c9..1204ba84b4a 100644
--- a/storage/src/test/java/org/apache/kafka/storage/internals/log/RemoteIndexCacheTest.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/RemoteIndexCacheTest.java
@@ -1221,7 +1221,7 @@ public class RemoteIndexCacheTest {
return metadataList;
}
- private void maybeAppendIndexEntries(OffsetIndex offsetIndex, TimeIndex timeIndex) {
+ private void maybeAppendIndexEntries(OffsetIndex offsetIndex, TimeIndex timeIndex) throws IOException {
if (!offsetIndex.isFull()) {
long curTime = time.milliseconds();
for (int i = 0; i < offsetIndex.maxEntries(); i++) {
diff --git a/storage/src/test/java/org/apache/kafka/storage/internals/log/TimeIndexTest.java b/storage/src/test/java/org/apache/kafka/storage/internals/log/TimeIndexTest.java
index 7cb7a0d5896..c346c5c268a 100644
--- a/storage/src/test/java/org/apache/kafka/storage/internals/log/TimeIndexTest.java
+++ b/storage/src/test/java/org/apache/kafka/storage/internals/log/TimeIndexTest.java
@@ -102,6 +102,7 @@ public class TimeIndexTest {
appendEntries(5);
TimestampOffset firstEntry = idx.entry(0);
idx.sanityCheck();
+ idx.trimToValidSize();
idx.close();
class MockTimeIndex extends TimeIndex {
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 bb4ba94246b..0f8752630b2 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
@@ -2412,10 +2412,10 @@ public class UnifiedLogTest {
}
@Test
- public void testLogRollAfterLogHandlerClosed() throws IOException {
+ public void testLogRollAfterCloseQuietly() throws IOException {
LogConfig logConfig = new LogTestUtils.LogConfigBuilder().build();
UnifiedLog log = createLog(logDir, logConfig);
- log.closeHandlers();
+ log.closeQuietly();
assertThrows(KafkaStorageException.class, () -> log.roll(Optional.of(1L)));
}
@@ -6087,7 +6087,6 @@ public class UnifiedLogTest {
assertTrue(exception.getMessage().contains(String.valueOf(bumpedEpoch)));
}
-
@Test
public void testFetchOffsetByTimestampRejectsCompressedRecordExceedingMaxDecompressedMessageBytes() throws IOException {
LogConfig logConfig = new LogTestUtils.LogConfigBuilder()
@@ -6135,4 +6134,21 @@ public class UnifiedLogTest {
() -> log.fetchOffsetByTimestamp(ListOffsetsRequest.MAX_TIMESTAMP, Optional.empty()));
assertTrue(e.getMessage().contains("exceeds the configured maximum record size of 100"), e.getMessage());
}
+
+ @Test
+ public void testPrepareActiveSegmentForCloseAppendsTimeIndexAndTrimsIndexes() throws IOException {
+ log = createLog(logDir, new LogConfig(new Properties()));
+ log.appendAsLeader(MemoryRecords.withRecords(
+ Compression.NONE,
+ new SimpleRecord(mockTime.milliseconds(), "key".getBytes(), "value".getBytes())
+ ), 0);
+
+ LogSegment activeSegment = log.activeSegment();
+ assertTrue(activeSegment.timeIndex().sizeInBytes() < activeSegment.timeIndex().maxIndexSize());
+
+ log.prepareActiveSegmentForClose();
+
+ assertEquals(activeSegment.timeIndex().entrySize(), activeSegment.timeIndex().sizeInBytes());
+ assertEquals(activeSegment.timeIndex().sizeInBytes(), activeSegment.timeIndex().length());
+ }
}
diff --git a/tools/src/main/java/org/apache/kafka/tools/DumpLogSegments.java b/tools/src/main/java/org/apache/kafka/tools/DumpLogSegments.java
index f968aeea167..040c5741bc7 100644
--- a/tools/src/main/java/org/apache/kafka/tools/DumpLogSegments.java
+++ b/tools/src/main/java/org/apache/kafka/tools/DumpLogSegments.java
@@ -311,15 +311,7 @@ public class DumpLogSegments {
prevTimestamp = entry.timestamp();
}
} finally {
- if (fileRecords != null) {
- fileRecords.closeHandlers();
- }
- if (index != null) {
- index.closeHandler();
- }
- if (timeIndex != null) {
- timeIndex.closeHandler();
- }
+ Utils.closeAll(fileRecords, index, timeIndex);
}
}
@@ -376,9 +368,7 @@ public class DumpLogSegments {
}
}
- FileRecords fileRecords = null;
- try {
- fileRecords = FileRecords.open(file, false).slice(0, maxBytes);
+ try (FileRecords fileRecords = FileRecords.open(file, false).slice(0, maxBytes)) {
long validBytes = 0L;
AtomicLong lastOffset = new AtomicLong(-1L);
@@ -392,10 +382,6 @@ public class DumpLogSegments {
}
printTrailingBytes(fileRecords, validBytes, maxBytes, file);
- } finally {
- if (fileRecords != null) {
- fileRecords.closeHandlers();
- }
}
}