This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch cassandra-5.0
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cassandra-5.0 by this push:
new 58863f59db Fix ThreadLocalReadAheadBuffer#fill() to throw a
CorruptBlockException if chunk metadata and file size are out of sync
58863f59db is described below
commit 58863f59db06698330f01b8fdb883e933343043e
Author: nivy <[email protected]>
AuthorDate: Wed Jul 29 19:40:05 2026 -0700
Fix ThreadLocalReadAheadBuffer#fill() to throw a CorruptBlockException if
chunk metadata and file size are out of sync
patch by Nivy Kani; reviewed by Caleb Rackliffe and Francisco Guerrero for
CASSANDRA-21519
---
CHANGES.txt | 1 +
.../io/util/ThreadLocalReadAheadBuffer.java | 11 +++--
.../io/util/CompressedChunkReaderTest.java | 54 ++++++++++++++++++++++
.../io/util/ThreadLocalReadAheadBufferTest.java | 3 +-
4 files changed, 64 insertions(+), 5 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 310e651373..3d12be0786 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
5.0.9
+ * Return CorruptSSTableException if chunk metadata and file size are out of
sync (CASSANDRA-21519)
* Coordinator load-shedding returns OverloadedException without setting
streamId, misrouting query responses (CASSANDRA-21508)
* SAI Component Checksum Validation Should be Segment-Aware (CASSANDRA-21516)
* Support Python 3.12 and 3.13 in cqlsh (CASSANDRA-20997)
diff --git
a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java
b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java
index bc92407bef..5e78697686 100644
--- a/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java
+++ b/src/java/org/apache/cassandra/io/util/ThreadLocalReadAheadBuffer.java
@@ -24,6 +24,7 @@ import java.util.Map;
import io.netty.util.concurrent.FastThreadLocal;
import org.apache.cassandra.io.compress.BufferType;
+import org.apache.cassandra.io.compress.CorruptBlockException;
import org.apache.cassandra.io.sstable.CorruptSSTableException;
public final class ThreadLocalReadAheadBuffer
@@ -92,12 +93,14 @@ public final class ThreadLocalReadAheadBuffer
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;
- long realPosition = Math.min(channelSize, position);
- int blockNo = (int) (realPosition / bufferSize);
+ if (position >= channelSize)
+ throw new CorruptBlockException(channel.filePath(), position,
bufferSize);
+
+ int blockNo = (int) (position / bufferSize);
long blockPosition = blockNo * (long) bufferSize;
long remaining = channelSize - blockPosition;
@@ -114,7 +117,7 @@ public final class ThreadLocalReadAheadBuffer
blockBuffer.flip();
blockBuffer.limit(sizeToRead);
- blockBuffer.position((int) (realPosition - blockPosition));
+ blockBuffer.position((int) (position - blockPosition));
}
public int read(ByteBuffer dest, int length)
diff --git
a/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java
b/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java
index af4b458fec..6d1cc35821 100644
--- a/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java
+++ b/test/unit/org/apache/cassandra/io/util/CompressedChunkReaderTest.java
@@ -25,6 +25,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.assertj.core.api.Assertions;
@@ -33,6 +34,8 @@ import org.junit.Assert;
import org.junit.Test;
import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.StandardOpenOption;
import java.nio.file.Files;
import java.util.concurrent.atomic.AtomicInteger;
@@ -138,4 +141,55 @@ public class CompressedChunkReaderTest
}
};
}
+
+ @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);
+ }
+
+ 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());
+ 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
+ {
+ FileUtils.clean(buffer);
+ }
+ }
}
\ No newline at end of file
diff --git
a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java
b/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java
index e374c6303f..2d29504d6f 100644
--- a/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java
+++ b/test/unit/org/apache/cassandra/io/util/ThreadLocalReadAheadBufferTest.java
@@ -36,6 +36,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;
import org.quicktheories.WithQuickTheories;
@@ -119,7 +120,7 @@ public class ThreadLocalReadAheadBufferTest implements
WithQuickTheories
copied += trlab.read(buf2, trlab.remaining());
}
}
- catch (CorruptSSTableException e)
+ catch (CorruptSSTableException | CorruptBlockException e)
{
throw new RuntimeException(e);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
