From fa93639eb84ad4384996d9fb5eafe6e2bf0ea66c Mon Sep 17 00:00:00 2001 From: nivy Date: Mon, 13 Jul 2026 23:03:01 -0700 Subject: [PATCH 1/6] Fix fill() to return CorruptSSTableException if chunk metadata and file size are out of sync --- .../io/util/ThreadLocalReadAheadBuffer.java | 8 ++- .../StandardCompressedChunkReaderTest.java | 57 +++++++++++++++++++ 2 files changed, 62 insertions(+), 3 deletions(-) diff --git a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java index c8ef6cf1e3..57490f404a 100644 --- a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java +++ b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java @@ -104,8 +104,10 @@ public class ThreadLocalReadAheadBuffer implements Closeable { Block block = getBlock(); ByteBuffer blockBuffer = block.buffer; - long realPosition = Math.min(channelSize, position); - int blockNo = (int) (realPosition / bufferSize); + if (position >= channelSize) + throw new CorruptSSTableException(null, channel.filePath()); + + int blockNo = (int) (position / bufferSize); long blockPosition = blockNo * (long) bufferSize; long remaining = channelSize - blockPosition; @@ -119,7 +121,7 @@ public class ThreadLocalReadAheadBuffer implements Closeable blockBuffer.flip(); blockBuffer.limit(sizeToRead); - blockBuffer.position((int) (realPosition - blockPosition)); + blockBuffer.position((int) (position - blockPosition)); } protected void loadBlock(ByteBuffer blockBuffer, long blockPosition, int sizeToRead) diff --git a/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java b/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java index afad1cd117..f38b2bf66c 100644 --- a/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java +++ b/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java @@ -19,7 +19,9 @@ package org.apache.cassandra.io.util; import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; import java.nio.file.Files; +import java.nio.file.StandardOpenOption; import java.util.concurrent.atomic.AtomicInteger; import org.assertj.core.api.Assertions; @@ -33,6 +35,7 @@ import org.apache.cassandra.db.ClusteringComparator; import org.apache.cassandra.io.compress.CompressedSequentialWriter; import org.apache.cassandra.io.compress.CompressionMetadata; import org.apache.cassandra.io.filesystem.ListenableFileSystem; +import org.apache.cassandra.io.sstable.CorruptSSTableException; import org.apache.cassandra.io.sstable.metadata.MetadataCollector; import org.apache.cassandra.schema.CompressionParams; import org.apache.cassandra.utils.memory.MemoryUtil; @@ -112,4 +115,58 @@ public class StandardCompressedChunkReaderTest extends CompressedChunkReaderTest MemoryUtil.clean(buffer); } } + + + @Test(timeout = 10_000) + public void scanReaderShouldNotHangOnTruncatedFile() throws Exception + { + SequentialWriterOption writerOption = SequentialWriterOption.newBuilder().finishOnClose(false).bufferSize(1 << 10).build(); + CompressionParams params = CompressionParams.snappy(4096, 1.1); + + FileSystems.newGlobalInMemoryFileSystem(); + File f = new File("/truncated_hang_repro.db"); + File offsets = new File("/truncated_hang_repro.offset"); + File digest = new File("/truncated_hang_repro.digest"); + + long longsToWrite = 600; // 4800 uncompressed bytes -> 2 compressed chunks (second one partial) + CompressionMetadata metadata; + try (CompressedSequentialWriter writer = new CompressedSequentialWriter(f, offsets, digest, writerOption, params,new MetadataCollector(new ClusteringComparator()))) + { + for (long i = 0; i < longsToWrite; i++) + writer.writeLong(i); + + writer.sync(); + metadata = writer.open(0); + } + + // Sanity: read-ahead must actually engage for this test to exercise the scan path + Assert.assertTrue("read-ahead must be larger than chunk length for scanReader to be non-null", + DatabaseDescriptor.getCompressedReadAheadBufferSize() > metadata.chunkLength()); + + // Truncate file so that chunk metadata expects a chunk to extend further than the actual file size + long originalSize = Files.size(f.toPath()); + long truncatedSize = originalSize - (params.chunkLength() / 2); + try (FileChannel fc = FileChannel.open(f.toPath(), StandardOpenOption.WRITE)) + { + fc.truncate(truncatedSize); + } + long uncompressedTotal = longsToWrite * Long.BYTES; + long lastChunkUncompressedStart = ((uncompressedTotal - 1) / metadata.chunkLength()) * metadata.chunkLength(); + + ByteBuffer buffer = ByteBuffer.allocateDirect(metadata.chunkLength()); + try (ChannelProxy channel = new ChannelProxy(f); + CompressedChunkReader reader = new CompressedChunkReader.Standard(channel, metadata, () -> 1.1); + metadata) + { + reader.forScan(); + + Assertions.assertThatThrownBy(() -> reader.readChunk(lastChunkUncompressedStart, buffer)) + .as("readChunk() reading past truncated EOF via the scan path") + .isInstanceOf(CorruptSSTableException.class); + } + finally + { + MemoryUtil.clean(buffer); + } + } } \ No newline at end of file From e83c85aed014f586dbc42c6b29504e40903596d8 Mon Sep 17 00:00:00 2001 From: nivy Date: Tue, 21 Jul 2026 21:32:56 -0700 Subject: [PATCH 2/6] Add exception message --- .../apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java index 57490f404a..5b55816f53 100644 --- a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java +++ b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java @@ -18,6 +18,7 @@ package org.apache.cassandra.io.util; +import java.io.IOException; import java.nio.ByteBuffer; import java.util.HashMap; import java.util.Map; @@ -105,7 +106,9 @@ public class ThreadLocalReadAheadBuffer implements Closeable Block block = getBlock(); ByteBuffer blockBuffer = block.buffer; if (position >= channelSize) - throw new CorruptSSTableException(null, channel.filePath()); + throw new CorruptSSTableException(new IOException(String.format("Chunk read past EOF: requested position %d must be less than file size %d", + position, channelSize)), + channel.filePath()); int blockNo = (int) (position / bufferSize); long blockPosition = blockNo * (long) bufferSize; From 8214ae208df2b7106ba7e7a7244a930a6fa1bea3 Mon Sep 17 00:00:00 2001 From: nivy Date: Tue, 28 Jul 2026 13:21:26 -0700 Subject: [PATCH 3/6] Fix formatting and set buffer size in test --- .../cassandra/io/util/ThreadLocalReadAheadBuffer.java | 7 +++---- .../io/util/StandardCompressedChunkReaderTest.java | 6 ++---- 2 files changed, 5 insertions(+), 8 deletions(-) diff --git a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java index 5b55816f53..c77e44ec60 100644 --- a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java +++ b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java @@ -18,13 +18,13 @@ package org.apache.cassandra.io.util; -import java.io.IOException; import java.nio.ByteBuffer; import java.util.HashMap; import java.util.Map; import java.util.function.Supplier; import org.apache.cassandra.io.compress.BufferType; +import org.apache.cassandra.io.compress.CorruptBlockException; import org.apache.cassandra.io.sstable.CorruptSSTableException; import org.apache.cassandra.utils.Closeable; import org.apache.cassandra.utils.memory.MemoryUtil; @@ -106,9 +106,8 @@ public class ThreadLocalReadAheadBuffer implements Closeable Block block = getBlock(); ByteBuffer blockBuffer = block.buffer; if (position >= channelSize) - throw new CorruptSSTableException(new IOException(String.format("Chunk read past EOF: requested position %d must be less than file size %d", - position, channelSize)), - channel.filePath()); + throw new CorruptSSTableException(new CorruptBlockException(channel.filePath(), position, bufferSize), + channel.filePath()); int blockNo = (int) (position / bufferSize); long blockPosition = blockNo * (long) bufferSize; diff --git a/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java b/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java index f38b2bf66c..0906f39c71 100644 --- a/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java +++ b/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java @@ -130,7 +130,7 @@ public class StandardCompressedChunkReaderTest extends CompressedChunkReaderTest long longsToWrite = 600; // 4800 uncompressed bytes -> 2 compressed chunks (second one partial) CompressionMetadata metadata; - try (CompressedSequentialWriter writer = new CompressedSequentialWriter(f, offsets, digest, writerOption, params,new MetadataCollector(new ClusteringComparator()))) + try (CompressedSequentialWriter writer = new CompressedSequentialWriter(f, offsets, digest, writerOption, params, new MetadataCollector(new ClusteringComparator()))) { for (long i = 0; i < longsToWrite; i++) writer.writeLong(i); @@ -139,9 +139,7 @@ public class StandardCompressedChunkReaderTest extends CompressedChunkReaderTest metadata = writer.open(0); } - // Sanity: read-ahead must actually engage for this test to exercise the scan path - Assert.assertTrue("read-ahead must be larger than chunk length for scanReader to be non-null", - DatabaseDescriptor.getCompressedReadAheadBufferSize() > metadata.chunkLength()); + DatabaseDescriptor.setCompressedReadAheadBufferSizeInKb(256); // Truncate file so that chunk metadata expects a chunk to extend further than the actual file size long originalSize = Files.size(f.toPath()); From 258cacbef6f8843ba656f7364dc3dc59d2dae1a8 Mon Sep 17 00:00:00 2001 From: nivy Date: Wed, 29 Jul 2026 13:34:29 -0700 Subject: [PATCH 4/6] Throw CorruptBlockException directly --- .../apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java index c77e44ec60..ff59f7cf96 100644 --- a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java +++ b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java @@ -101,13 +101,12 @@ public class ThreadLocalReadAheadBuffer implements Closeable return blockMap.get().computeIfAbsent(channel.filePath(), k -> new Block()); } - public void fill(long position) + public void fill(long position) throws CorruptBlockException { Block block = getBlock(); ByteBuffer blockBuffer = block.buffer; if (position >= channelSize) - throw new CorruptSSTableException(new CorruptBlockException(channel.filePath(), position, bufferSize), - channel.filePath()); + throw new CorruptBlockException(channel.filePath(), position, bufferSize); int blockNo = (int) (position / bufferSize); long blockPosition = blockNo * (long) bufferSize; From a5386becd94737f6c137901fb9ba51c08dfe506e Mon Sep 17 00:00:00 2001 From: nivy Date: Wed, 29 Jul 2026 14:39:41 -0700 Subject: [PATCH 5/6] Catch exception --- .../cassandra/io/util/ThreadLocalReadAheadBufferTest.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java b/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java index a4a7f8d519..47bba7b6c5 100644 --- a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java +++ b/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java @@ -37,6 +37,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.config.DataStorageSpec; import org.apache.cassandra.io.compress.BufferType; +import org.apache.cassandra.io.compress.CorruptBlockException; import org.apache.cassandra.io.sstable.CorruptSSTableException; import org.apache.cassandra.utils.Pair; @@ -125,7 +126,7 @@ public class ThreadLocalReadAheadBufferTest implements WithQuickTheories copied += tlrab.read(buf2, tlrab.remaining()); } } - catch (CorruptSSTableException e) + catch (CorruptSSTableException | CorruptBlockException e) { throw new RuntimeException(e); } From 80c071cc0d8b0f218ad10dd188908be2843fd5d3 Mon Sep 17 00:00:00 2001 From: nivy Date: Thu, 30 Jul 2026 23:41:39 -0700 Subject: [PATCH 6/6] Remove newline --- .../cassandra/io/util/StandardCompressedChunkReaderTest.java | 1 - 1 file changed, 1 deletion(-) diff --git a/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java b/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java index 0906f39c71..436a3e3aa2 100644 --- a/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java +++ b/test/unit/org/apache/cassandra/io/util/StandardCompressedChunkReaderTest.java @@ -116,7 +116,6 @@ public class StandardCompressedChunkReaderTest extends CompressedChunkReaderTest } } - @Test(timeout = 10_000) public void scanReaderShouldNotHangOnTruncatedFile() throws Exception {