This is an automated email from the ASF dual-hosted git repository.
cryptoe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new e421a067931 fix: handle interrupt during partial mmap to not explode
(#20053)
e421a067931 is described below
commit e421a067931d3959071c94fed881a887a1210a1e
Author: Clint Wylie <[email protected]>
AuthorDate: Tue Aug 18 14:32:31 2026 -0700
fix: handle interrupt during partial mmap to not explode (#20053)
---
.../segment/file/PartialSegmentFileMapperV10.java | 171 ++++++++++++++-------
.../file/PartialSegmentFileMapperV10Test.java | 101 ++++++++++++
2 files changed, 220 insertions(+), 52 deletions(-)
diff --git
a/processing/src/main/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10.java
b/processing/src/main/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10.java
index df571ff2647..1c25f089871 100644
---
a/processing/src/main/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10.java
+++
b/processing/src/main/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10.java
@@ -44,6 +44,7 @@ import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.MappedByteBuffer;
+import java.nio.channels.ClosedByInterruptException;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.Collection;
@@ -195,6 +196,11 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
result = parseHeaderFile(headerFile, jsonMapper);
bitmapBuffer = mmapBitmap(headerFile, result);
}
+ catch (ClosedByInterruptException e) {
+ // The header is fine, an interrupt aborted the mapping (see
mapUninterruptibly). Treating this as corruption
+ // would delete a valid local header and force a needless re-download,
so leave the file alone and unwind.
+ throw e;
+ }
catch (Exception e) {
// corrupted file (partial write, truncated bitmap, bad JSON, etc.),
delete and re-fetch
result = null;
@@ -228,47 +234,54 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
maxFetchRunBytes
);
- // bitmap-vs-container repair pre-pass: if the bitmap claims a file is
downloaded but its container file is
- // missing on disk, the bitmap is lying (e.g. partial-cache eviction that
cleared containers but couldn't atomically
- // clear bits, or external file-system damage). Clear those bits before
the restore loop so we don't spuriously
- // sparse-allocate empty containers in the restore loop's
ensureContainerInitialized call and treat their files as
- // downloaded.
- for (int i = 0; i < mapper.sortedFileNames.size(); i++) {
- final int byteIndex = i / 8;
- final int bitMask = 1 << (i % 8);
- if ((bitmapBuffer.get(byteIndex) & bitMask) == 0) {
- continue;
- }
- final String name = mapper.sortedFileNames.get(i);
- final SegmentInternalFileMetadata fileMetadata =
result.getMetadata().getFiles().get(name);
- if (fileMetadata == null) {
- continue;
- }
- final File containerFile = new File(
- localCacheDir,
- StringUtils.format("%s.container.%05d", targetFilename,
fileMetadata.getContainer())
- );
- if (!containerFile.exists()) {
- bitmapBuffer.put(byteIndex, (byte) (bitmapBuffer.get(byteIndex) &
~bitMask));
- }
- }
-
- // restore downloaded files from the (now-repaired) bitmap
- for (int i = 0; i < mapper.sortedFileNames.size(); i++) {
- final int byteIndex = i / 8;
- final int bitIndex = i % 8;
- if ((bitmapBuffer.get(byteIndex) & (1 << bitIndex)) != 0) {
+ try {
+ // bitmap-vs-container repair pre-pass: if the bitmap claims a file is
downloaded but its container file is
+ // missing on disk, the bitmap is lying (e.g. partial-cache eviction
that cleared containers but couldn't
+ // atomically clear bits, or external file-system damage). Clear those
bits before the restore loop so we don't
+ // spuriously sparse-allocate empty containers in the restore loop's
ensureContainerInitialized call and treat
+ // their files as downloaded.
+ for (int i = 0; i < mapper.sortedFileNames.size(); i++) {
+ final int byteIndex = i / 8;
+ final int bitMask = 1 << (i % 8);
+ if ((bitmapBuffer.get(byteIndex) & bitMask) == 0) {
+ continue;
+ }
final String name = mapper.sortedFileNames.get(i);
final SegmentInternalFileMetadata fileMetadata =
result.getMetadata().getFiles().get(name);
- if (fileMetadata != null) {
- mapper.ensureContainerInitialized(fileMetadata.getContainer());
- mapper.downloadedFiles.add(name);
- mapper.downloadedBytes.addAndGet(fileMetadata.getSize());
+ if (fileMetadata == null) {
+ continue;
+ }
+ final File containerFile = new File(
+ localCacheDir,
+ StringUtils.format("%s.container.%05d", targetFilename,
fileMetadata.getContainer())
+ );
+ if (!containerFile.exists()) {
+ bitmapBuffer.put(byteIndex, (byte) (bitmapBuffer.get(byteIndex) &
~bitMask));
+ }
+ }
+
+ // restore downloaded files from the (now-repaired) bitmap
+ for (int i = 0; i < mapper.sortedFileNames.size(); i++) {
+ final int byteIndex = i / 8;
+ final int bitIndex = i % 8;
+ if ((bitmapBuffer.get(byteIndex) & (1 << bitIndex)) != 0) {
+ final String name = mapper.sortedFileNames.get(i);
+ final SegmentInternalFileMetadata fileMetadata =
result.getMetadata().getFiles().get(name);
+ if (fileMetadata != null) {
+ mapper.ensureContainerInitialized(fileMetadata.getContainer());
+ mapper.downloadedFiles.add(name);
+ mapper.downloadedBytes.addAndGet(fileMetadata.getSize());
+ }
}
}
- }
- return mapper;
+ return mapper;
+ }
+ catch (Throwable t) {
+ // close a half-built mapper
+ CloseableUtils.closeAndSuppressExceptions(mapper, t::addSuppressed);
+ throw t;
+ }
}
private final SegmentFileMetadata metadata;
@@ -1072,8 +1085,9 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
/**
* Initialize a local container file if not already done. Creates a sparse
file at the original container size
- * and memory-maps it. The channel is closed immediately after mapping, the
mmap persists independently, backed by
- * the kernel page cache. This avoids the risk of channel closure from
thread interruption.
+ * and memory-maps it via {@link #mapUninterruptibly}. The channel is closed
immediately after mapping, the mmap
+ * persists independently, backed by the kernel page cache, so once
established it is immune to channel closure from
+ * thread interruption.
*/
private void ensureContainerInitialized(int containerIndex) throws
IOException
{
@@ -1095,15 +1109,14 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
);
// create sparse file at original container size, mmap it, then close
the channel immediately
- try (RandomAccessFile raf = new RandomAccessFile(localFile, "rw");
FileChannel channel = raf.getChannel()) {
- raf.setLength(containerMeta.getSize());
- containerFiles[containerIndex] = localFile;
- containers[containerIndex] = channel.map(
- FileChannel.MapMode.READ_ONLY,
- 0,
- containerMeta.getSize()
- );
- }
+ final MappedByteBuffer container = mapUninterruptibly(() -> {
+ try (RandomAccessFile raf = new RandomAccessFile(localFile, "rw");
FileChannel channel = raf.getChannel()) {
+ raf.setLength(containerMeta.getSize());
+ return channel.map(FileChannel.MapMode.READ_ONLY, 0,
containerMeta.getSize());
+ }
+ });
+ containerFiles[containerIndex] = localFile;
+ containers[containerIndex] = container;
}
finally {
containerLocks[containerIndex].unlock();
@@ -1270,15 +1283,69 @@ public class PartialSegmentFileMapperV10 implements
SegmentFileMapper
{
final int numBitmapBytes = (result.getMetadata().getFiles().size() + 7) /
8;
final long expectedSize = result.getHeaderSize() + numBitmapBytes;
- try (RandomAccessFile raf = new RandomAccessFile(headerFile, "rw");
- FileChannel channel = raf.getChannel()) {
- if (raf.length() < expectedSize) {
- raf.setLength(expectedSize);
+ return mapUninterruptibly(() -> {
+ try (RandomAccessFile raf = new RandomAccessFile(headerFile, "rw");
+ FileChannel channel = raf.getChannel()) {
+ if (raf.length() < expectedSize) {
+ raf.setLength(expectedSize);
+ }
+ return channel.map(FileChannel.MapMode.READ_WRITE,
result.getHeaderSize(), numBitmapBytes);
+ }
+ });
+ }
+
+ /**
+ * Establish a memory mapping, shielding it from the calling thread's
interrupt status.
+ * <p>
+ * {@link FileChannel#map} is an interruptible channel operation: an
interrupt (a canceled query, a stage tearing
+ * down, {@code shutdownNow} on the processing pool) closes the channel
mid-call and surfaces as
+ * {@link ClosedByInterruptException}. This class already avoids that hazard
for container data by reading and
+ * writing through plain {@link RandomAccessFile} rather than NIO channels,
but mapping has no non-NIO equivalent, so
+ * it is handled here instead: the interrupt status is parked for the
duration of the call and the mapping is retried
+ * once on a fresh channel if an interrupt lands inside the call itself.
+ * <p>
+ * The flag is always restored before returning, so an interrupt is never
swallowed, the caller still observes it at
+ * its next cancellation checkpoint.
+ */
+ private static MappedByteBuffer mapUninterruptibly(MmapOperation operation)
throws IOException
+ {
+ boolean interrupted = Thread.interrupted();
+ boolean retried = false;
+
+ try {
+ while (true) {
+ try {
+ return operation.run();
+ }
+ catch (ClosedByInterruptException e) {
+ // An interrupt arrived after the clear above, retry once
+ interrupted = true;
+
+ if (retried) {
+ throw e;
+ }
+
+ retried = true;
+ Thread.interrupted();
+ }
+ }
+ }
+ finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt();
}
- return channel.map(FileChannel.MapMode.READ_WRITE,
result.getHeaderSize(), numBitmapBytes);
}
}
+ /**
+ * A mapping attempt for {@link #mapUninterruptibly}. Opens its own channel
so that each attempt is independent.
+ */
+ @FunctionalInterface
+ private interface MmapOperation
+ {
+ MappedByteBuffer run() throws IOException;
+ }
+
/**
* One planned coalesced range read: {@code containerIndex} identifies the
container within the mapper that planned
* it (runs are only meaningful to their planning mapper), {@code
startOffset}/{@code length} are container-local,
diff --git
a/processing/src/test/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10Test.java
b/processing/src/test/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10Test.java
index 21e730ca954..3925976b88b 100644
---
a/processing/src/test/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10Test.java
+++
b/processing/src/test/java/org/apache/druid/segment/file/PartialSegmentFileMapperV10Test.java
@@ -826,6 +826,107 @@ class PartialSegmentFileMapperV10Test
}
}
+ @Test
+ void testFetchWithPendingInterruptStillMapsContainer() throws IOException
+ {
+ final File segmentFile = buildTestSegment(10, CompressionStrategy.NONE);
+ final File cacheDir = newCacheDir("interrupted-fetch");
+ final DirectoryBackedRangeReader rangeReader = new
DirectoryBackedRangeReader(segmentFile.getParentFile());
+
+ try (PartialSegmentFileMapperV10 mapper = createMapper(rangeReader,
cacheDir)) {
+ // A pending interrupt (canceled query, stage teardown, shutdownNow on
the processing pool) must not abort the
+ // container mmap: FileChannel.map is an interruptible channel
operation, so without the interrupt parking in
+ // mapUninterruptibly this fails with ClosedByInterruptException.
+ Thread.currentThread().interrupt();
+ try {
+ mapper.fetchFiles(Set.of("5"));
+
+ final ByteBuffer buf = mapper.mapFile("5");
+ Assertions.assertNotNull(buf);
+ Assertions.assertEquals(5, buf.getInt());
+
+ // preserved rather than swallowed, so the caller still unwinds at its
next cancellation checkpoint
+ Assertions.assertTrue(Thread.currentThread().isInterrupted());
+ }
+ finally {
+ // don't leak the flag into the mapper close below, or into other
tests on this thread
+ Thread.interrupted();
+ }
+ }
+ }
+
+ @Test
+ void testRestoreWithPendingInterruptKeepsHeader() throws IOException
+ {
+ final File segmentFile = buildTestSegment(10, CompressionStrategy.NONE);
+ final File cacheDir = newCacheDir("interrupted-restore");
+ final File headerFile = new File(
+ cacheDir,
+ IndexIO.V10_FILE_NAME +
PartialSegmentFileMapperV10.METADATA_HEADER_SUFFIX
+ );
+
+ try (PartialSegmentFileMapperV10 mapper =
+ createMapper(new
DirectoryBackedRangeReader(segmentFile.getParentFile()), cacheDir)) {
+ mapper.fetchFiles(Set.of("5"));
+ Assertions.assertEquals(4, mapper.getDownloadedBytes());
+ }
+
+ final CountingRangeReader freshReader = new
CountingRangeReader(segmentFile.getParentFile());
+ try {
+ Thread.currentThread().interrupt();
+ try (PartialSegmentFileMapperV10 restored = createMapper(freshReader,
cacheDir)) {
+ // The bitmap mmap is interrupt-immune too, so a valid local header is
restored as-is rather than mistaken for
+ // a corrupt file, deleted, and re-fetched from deep storage.
+ Assertions.assertTrue(Thread.currentThread().isInterrupted());
+
+ // Clear here rather than in the finally below, so the mapper close at
the end of this block runs with the
+ // same clear flag its counterpart in
testFetchWithPendingInterruptStillMapsContainer does; the finally is
+ // then only a backstop for a failure before this point.
+ Thread.interrupted();
+
+ Assertions.assertTrue(headerFile.exists());
+ Assertions.assertEquals(0, freshReader.getHeaderReadCount(), "header
must not have been re-downloaded");
+ Assertions.assertEquals(4, restored.getDownloadedBytes());
+
+ final ByteBuffer buf = restored.mapFile("5");
+ Assertions.assertNotNull(buf);
+ Assertions.assertEquals(5, buf.getInt());
+ }
+ }
+ finally {
+ Thread.interrupted();
+ }
+ }
+
+ @Test
+ void testRestoreFailurePropagatesOriginalException() throws IOException
+ {
+ final File segmentFile = buildTestSegment(10, CompressionStrategy.NONE);
+ final File cacheDir = newCacheDir("restore-failure");
+
+ // Persist a header + bitmap with one file resident, so the restore loop
has a container to initialize.
+ try (PartialSegmentFileMapperV10 mapper =
+ createMapper(new
DirectoryBackedRangeReader(segmentFile.getParentFile()), cacheDir)) {
+ mapper.fetchFiles(Set.of("5"));
+ }
+
+ // Replace the container file with a directory of the same name: it still
passes the repair pre-pass's exists()
+ // check, so the bitmap bit survives, but the restore loop's
RandomAccessFile open fails. Doing it this way rather
+ // than with permissions keeps the test deterministic when it runs as root.
+ final File containerFile = new File(cacheDir,
StringUtils.format("%s.container.%05d", IndexIO.V10_FILE_NAME, 0));
+ Assertions.assertTrue(containerFile.isFile());
+ Assertions.assertTrue(containerFile.delete());
+ FileUtils.mkdirp(containerFile);
+
+ // createForFile closes the half-built mapper on the way out (releasing
the bitmap mmap and any containers already
+ // initialized). The unmap itself isn't observable in-process; what this
pins is that the cleanup neither swallows
+ // the failure nor rewraps it into some other type, which is what would
hide the real cause from callers.
+ Assertions.assertThrows(
+ IOException.class,
+ () -> createMapper(new
DirectoryBackedRangeReader(segmentFile.getParentFile()), cacheDir)
+ );
+ }
+
@Test
void testCreateWithExternals() throws IOException
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
