You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by lu...@apache.org on 2026/04/16 08:16:20 UTC
(cassandra-analytics) branch trunk updated: CASSANALYTICS-147: BufferingInputStream fails to read last unaligned chunk Patch by Lukasz Antoniak; Reviewed by Shailaja Koppu and Yifan Cai for CASSANALYTICS-147
This is an automated email from the ASF dual-hosted git repository.
lukasz pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-analytics.git
The following commit(s) were added to refs/heads/trunk by this push:
new 7f7eb6d4 CASSANALYTICS-147: BufferingInputStream fails to read last unaligned chunk Patch by Lukasz Antoniak; Reviewed by Shailaja Koppu and Yifan Cai for CASSANALYTICS-147
7f7eb6d4 is described below
commit 7f7eb6d4024b8b165f9f3d0881485cdcb650df3e
Author: Lukasz Antoniak <Lu...@gmail.com>
AuthorDate: Thu Apr 2 06:56:41 2026 +0200
CASSANALYTICS-147: BufferingInputStream fails to read last unaligned chunk
Patch by Lukasz Antoniak; Reviewed by Shailaja Koppu and Yifan Cai for CASSANALYTICS-147
---
CHANGES.txt | 1 +
.../cassandra/spark/data/FileSystemSource.java | 7 +-
.../utils/streaming/BufferingInputStream.java | 6 +-
.../spark/utils/BufferingInputStreamTests.java | 77 +++++++++++++++++++++-
.../cassandra/io/util/CdcRandomAccessReader.java | 4 +-
.../io/util/CdcRandomAccessReaderTest.java | 12 ++--
6 files changed, 90 insertions(+), 17 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index b1ca9640..cf4540d5 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,6 @@
0.4.0
-----
+ * BufferingInputStream fails to read last unaligned chunk (CASSANALYTICS-147)
* Support per-instance sidecar port resolution in CDC client (CASSANALYTICS-130)
* Change log level for logs in CassandraSchema to debug level to avoid log spamming (CASSANALYTICS-149)
* Pass SidecarCdcClient as a constructor parameter to avoid thread/resource leaks (CASSANALYTICS-142)
diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/FileSystemSource.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/FileSystemSource.java
index f18a3be3..5fc736e6 100644
--- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/FileSystemSource.java
+++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/data/FileSystemSource.java
@@ -91,12 +91,11 @@ public class FileSystemSource<T extends CassandraFile> implements CassandraFileS
public void request(long start, long end, StreamConsumer consumer)
{
executor().submit(() -> {
- boolean close = length <= end;
+ boolean close = length <= end + 1;
try
{
- // Start-end range is inclusive but on the final request end == length so we need to exclude
- int increment = close ? 0 : 1;
- byte[] bytes = new byte[(int) (end - start + increment)];
+ // start-end range is inclusive
+ byte[] bytes = new byte[(int) (end - start + 1)];
if (file.getChannel().read(ByteBuffer.wrap(bytes), start) >= 0)
{
consumer.onRead(StreamBuffer.wrap(bytes));
diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/streaming/BufferingInputStream.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/streaming/BufferingInputStream.java
index ecfdb82d..0b1d84f7 100644
--- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/streaming/BufferingInputStream.java
+++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/streaming/BufferingInputStream.java
@@ -107,7 +107,7 @@ public class BufferingInputStream<T extends CassandraFile> extends InputStream i
this.rangeStart = position;
// adjusting bytes read and written so that isFinished() returns correct output and we know
// when we reached the end of the stream
- this.bytesRead = position == 0 ? 0 : (position + 1);
+ this.bytesRead = position;
this.bytesWritten.set(this.bytesRead);
}
@@ -199,12 +199,12 @@ public class BufferingInputStream<T extends CassandraFile> extends InputStream i
}
long chunkSize = rangeStart == 0 ? source.headerChunkSize() : source.chunkBufferSize();
- long rangeEnd = Math.min(source.size(), rangeStart + chunkSize);
+ long rangeEnd = Math.min(source.size(), rangeStart + chunkSize) - 1; // start and end range are inclusive
if (rangeEnd >= rangeStart)
{
activeRequest = true;
source.request(rangeStart, rangeEnd, this);
- rangeStart += chunkSize + 1; // Increment range start pointer for next request
+ rangeStart += chunkSize; // Increment range start pointer for next request
}
else
{
diff --git a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/BufferingInputStreamTests.java b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/BufferingInputStreamTests.java
index 1b0aace3..fc41d7fb 100644
--- a/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/BufferingInputStreamTests.java
+++ b/cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/utils/BufferingInputStreamTests.java
@@ -20,7 +20,10 @@
package org.apache.cassandra.spark.utils;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -33,6 +36,7 @@ import java.util.stream.IntStream;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.google.common.util.concurrent.Uninterruptibles;
+import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.mutable.MutableInt;
import org.junit.jupiter.api.Test;
@@ -289,7 +293,7 @@ public class BufferingInputStreamTests
};
int bytesToRead = chunkSize * numChunks;
- long skipAhead = size - bytesToRead + 1;
+ long skipAhead = size - bytesToRead;
try (BufferingInputStream<SSTable> stream = new BufferingInputStream<>(source, STATS.bufferingInputStreamStats()))
{
// Skip ahead so we only read the final chunks
@@ -346,6 +350,77 @@ public class BufferingInputStreamTests
}
}
+ @Test
+ public void testUnalignedEndReading() throws IOException
+ {
+ int dataSize = 8192;
+ int chunkSize = 4096;
+ int remainingReadBytes = 2;
+ List<byte[]> returnedBuffers = new ArrayList<>();
+ CassandraFileSource<SSTable> source = new CassandraFileSource<SSTable>()
+ {
+ @Override
+ public void request(long start, long end, StreamConsumer consumer)
+ {
+ byte[] bytes = RandomUtils.randomBytes((int) (end - start + 1));
+ StreamBuffer buffer = StreamBuffer.wrap(bytes);
+ returnedBuffers.add(bytes);
+ consumer.onRead(buffer);
+ consumer.onEnd();
+ }
+
+ @Override
+ public SSTable cassandraFile()
+ {
+ return null;
+ }
+
+ @Override
+ public FileType fileType()
+ {
+ return FileType.PARTITIONS_INDEX;
+ }
+
+ @Override
+ public long size()
+ {
+ return dataSize;
+ }
+
+ @Override
+ @Nullable
+ public Duration timeout()
+ {
+ return Duration.ofSeconds(5);
+ }
+
+ @Override
+ public long chunkBufferSize()
+ {
+ return chunkSize;
+ }
+ };
+
+ try (BufferingInputStream<SSTable> stream1 = new BufferingInputStream<>(source, STATS.bufferingInputStreamStats()))
+ {
+ // move left from the file end by (chunkSize + remainingReadBytes)
+ try (BufferingInputStream<SSTable> stream2 = stream1.reBuffer(dataSize - chunkSize - remainingReadBytes))
+ {
+ ByteBuffer buffer = ByteBuffer.allocate(chunkSize);
+ int read = stream2.read(buffer); // read last full chunk
+ assertThat(returnedBuffers).hasSize(2);
+ assertThat(read).isEqualTo(chunkSize);
+ assertThat(buffer.array()).isEqualTo(returnedBuffers.get(0));
+
+ buffer.position(0);
+ buffer.limit(remainingReadBytes);
+ read = stream2.read(buffer); // read remaining bytes
+ assertThat(read).isEqualTo(remainingReadBytes);
+ assertThat(ArrayUtils.subarray(buffer.array(), 0, remainingReadBytes)).isEqualTo(returnedBuffers.get(1));
+ }
+ }
+ }
+
// Utils
private static ImmutableList<StreamBuffer> randomBuffers(int count)
diff --git a/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/io/util/CdcRandomAccessReader.java b/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/io/util/CdcRandomAccessReader.java
index ec487fc5..581ec4d1 100644
--- a/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/io/util/CdcRandomAccessReader.java
+++ b/cassandra-four-zero-bridge/src/main/java/org/apache/cassandra/io/util/CdcRandomAccessReader.java
@@ -111,8 +111,8 @@ public class CdcRandomAccessReader extends RandomAccessReader
// attempting to read bytes previously read
// in practice we read the Commit Logs sequentially
// but we still need to respect random access reader API, it will just require blocking
- int requestLen = buffer.remaining() + 1;
- long end = offset + requestLen;
+ int requestLen = buffer.remaining();
+ long end = offset + requestLen - 1;
BlockingStreamConsumer streamConsumer = new BlockingStreamConsumer();
source.request(offset, end, streamConsumer);
streamConsumer.getBytes(buffer);
diff --git a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/io/util/CdcRandomAccessReaderTest.java b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/io/util/CdcRandomAccessReaderTest.java
index 28a9f902..d81f72c6 100644
--- a/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/io/util/CdcRandomAccessReaderTest.java
+++ b/cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/io/util/CdcRandomAccessReaderTest.java
@@ -92,9 +92,9 @@ public class CdcRandomAccessReaderTest
long position = call.start;
// Deliver data in chunks until request is fulfilled
- while (position < actualEnd)
+ while (position <= actualEnd) // range boundaries are inclusive
{
- int chunkSize = (int) Math.min(actualEnd - position, bufferSize);
+ int chunkSize = (int) Math.min(actualEnd - position + 1, bufferSize);
Buffer data = Buffer.buffer();
for (int i = 0; i < chunkSize; i++)
{
@@ -154,13 +154,10 @@ public class CdcRandomAccessReaderTest
testCommitLog.setMaxOffset(100L);
final int bufferSize = 50;
testSource.setRequestHandler(call -> {
- // Backward seek path has a bug: requests buffer.remaining() + 1 bytes
// We cap delivery at buffer capacity to work around this and test flip() behavior
long actualEnd = Math.min(call.end, testCommitLog.maxOffset());
- long requestedBytes = actualEnd - call.start;
+ long requestedBytes = actualEnd - call.start + 1; // range boundaries are inclusive
long position = call.start;
-
- // Cap delivery to buffer capacity to avoid BufferOverflowException from + 1 bug
long bytesToDeliver = Math.min(requestedBytes, bufferSize);
// Deliver capped amount
@@ -214,7 +211,7 @@ public class CdcRandomAccessReaderTest
// Configure source to provide sequential data
testSource.setRequestHandler(call -> {
- int dataSize = (int) (call.end - call.start);
+ int dataSize = (int) (call.end - call.start + 1);
Buffer data = Buffer.buffer(dataSize);
for (int i = 0; i < dataSize; i++)
{
@@ -389,6 +386,7 @@ public class CdcRandomAccessReaderTest
static class RequestCall
{
+ // start and end offsets are considered inclusive
final long start;
final long end;
final StreamConsumer consumer;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org