bbotella commented on code in PR #148: URL: https://github.com/apache/cassandra-analytics/pull/148#discussion_r2470685922
########## cassandra-four-zero-bridge/src/test/java/org/apache/cassandra/io/util/CdcRandomAccessReaderTest.java: ########## @@ -0,0 +1,455 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.cassandra.io.util; + +import java.io.IOException; +import io.vertx.core.buffer.Buffer; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.function.Consumer; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import org.apache.cassandra.cdc.api.CommitLog; +import org.apache.cassandra.cdc.stats.ICdcStats; +import org.apache.cassandra.spark.data.FileType; +import org.apache.cassandra.spark.data.partitioner.CassandraInstance; +import org.apache.cassandra.spark.utils.streaming.CassandraFileSource; +import org.apache.cassandra.spark.utils.streaming.StreamBuffer; +import org.apache.cassandra.spark.utils.streaming.StreamConsumer; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + + +public class CdcRandomAccessReaderTest +{ + private TestCommitLog testCommitLog; + private TestCassandraFileSource testSource; + private TestCdcStats testStats; + + private CdcRandomAccessReader reader; + + @BeforeEach + public void setUp() + { + testSource = new TestCassandraFileSource(); + testStats = new TestCdcStats(); + testCommitLog = new TestCommitLog("/test/path/commitlog", testSource, testStats, 1024L); + testSource.setCommitLog(testCommitLog); + } + + @Test + public void testConstructorInitialization() + { + // Verify reader is properly initialized with commit log + reader = new CdcRandomAccessReader(testCommitLog); + + assertThat(reader).isNotNull(); + assertThat(reader.getPath()).isEqualTo("/test/path/commitlog"); + } + + @Test + public void testCDCRebufferConstructorWithInvalidChunkSize() + { + // Verify constructor rejects zero chunk size + assertThatThrownBy(() -> new CdcRandomAccessReader.CDCRebuffer(testCommitLog, 0)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Chunk size must be a positive integer"); + + // Verify constructor rejects negative chunk size + assertThatThrownBy(() -> new CdcRandomAccessReader.CDCRebuffer(testCommitLog, -1)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Chunk size must be a positive integer"); + } + + @Test + public void testCDCRebufferSequentialReading() throws IOException + { + // Setup: 100 bytes total, 50-byte buffer chunks + testCommitLog.setMaxOffset(100L); + final int bufferSize = 50; + testSource.setRequestHandler(call -> { + // BufferingInputStream requests a range - we must provide ALL requested data + long actualEnd = Math.min(call.end, testCommitLog.maxOffset()); + long position = call.start; + + // Deliver data in chunks until request is fulfilled + while (position < actualEnd) + { + int chunkSize = (int) Math.min(actualEnd - position, bufferSize); + Buffer data = Buffer.buffer(); + for (int i = 0; i < chunkSize; i++) + { + data.appendByte((byte) (position + i)); + } + + TestStreamBuffer streamBuffer = new TestStreamBuffer(data); + call.consumer.onRead(streamBuffer); + position += chunkSize; + } + + // Signal end of stream when reaching EOF + if (actualEnd >= testCommitLog.maxOffset()) + { + call.consumer.onEnd(); + } + }); + CdcRandomAccessReader.CDCRebuffer rebuffer = new CdcRandomAccessReader.CDCRebuffer(testCommitLog, bufferSize); + + // First read: bytes 0-49 + Rebufferer.BufferHolder holder = rebuffer.rebuffer(0); + assertThat(holder).isNotNull(); + assertThat(rebuffer.offset()).isEqualTo(0L); + + // Verify buffer was fully written (position at end, remaining = 0) + ByteBuffer buffer = holder.buffer(); + assertThat(buffer.remaining()).isEqualTo(0); + assertThat(buffer.position()).isEqualTo(50); + + // Flip to read mode and verify actual byte values + buffer.flip(); + for (int i = 0; i < 50; i++) + { + assertThat(buffer.get()).isEqualTo((byte) i); + } + + // Second read: bytes 50-99 (sequential) + holder = rebuffer.rebuffer(50); + assertThat(holder).isNotNull(); + assertThat(rebuffer.offset()).isEqualTo(50L); + + // Verify buffer was fully written (position at end, remaining = 0) + buffer = holder.buffer(); + assertThat(buffer.remaining()).isEqualTo(0); + assertThat(buffer.position()).isEqualTo(50); + + // Flip to read mode and verify actual byte values + buffer.flip(); + for (int i = 0; i < 50; i++) + { + assertThat(buffer.get()).isEqualTo((byte) (50 + i)); + } + } + + @Test + public void testCDCRebufferBackwardSeek() throws IOException + { + // Setup: 100 bytes total, 50-byte buffer chunks + 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 position = call.start; + + // Cap delivery to buffer capacity to avoid BufferOverflowException from + 1 bug + long bytesToDeliver = Math.min(requestedBytes, bufferSize); + + // Deliver capped amount + while (position < call.start + bytesToDeliver) + { + int chunkSize = (int) Math.min(call.start + bytesToDeliver - position, bufferSize); + Buffer data = Buffer.buffer(); + for (int i = 0; i < chunkSize; i++) + { + data.appendByte((byte) (position + i)); + } + + TestStreamBuffer streamBuffer = new TestStreamBuffer(data); + call.consumer.onRead(streamBuffer); + position += chunkSize; + } + + // Signal end when complete + call.consumer.onEnd(); + }); + CdcRandomAccessReader.CDCRebuffer rebuffer = new CdcRandomAccessReader.CDCRebuffer(testCommitLog, bufferSize); + + // First, advance to position 50 (sequential read) + Rebufferer.BufferHolder holder = rebuffer.rebuffer(50); + assertThat(holder).isNotNull(); + assertThat(rebuffer.offset()).isEqualTo(50L); + + // Now seek backward to position 0 - triggers backward seek code path + holder = rebuffer.rebuffer(0); + assertThat(holder).isNotNull(); + assertThat(rebuffer.offset()).isEqualTo(0L); + + // Verify buffer is in write mode + ByteBuffer buffer = holder.buffer(); + assertThat(buffer.remaining()).isEqualTo(0); + assertThat(buffer.position()).isEqualTo(50); + + // TEST flips to verify byte values are correct + buffer.flip(); + for (int i = 0; i < 50; i++) + { + assertThat(buffer.get()).isEqualTo((byte) i); + } + } + + + @Test + public void testCdcRandomAccessReaderEndToEnd() + { + // Setup commit log with 1024 bytes + testCommitLog.setMaxOffset(1024L); + + // Configure source to provide sequential data + testSource.setRequestHandler(call -> { + int dataSize = (int) (call.end - call.start); + Buffer data = Buffer.buffer(dataSize); + for (int i = 0; i < dataSize; i++) + { + data.appendByte((byte) (call.start + i)); + } + + TestStreamBuffer streamBuffer = new TestStreamBuffer(data); + + // Deliver data and signal completion + call.consumer.onRead(streamBuffer); + call.consumer.onEnd(); + }); + + // Create reader + reader = new CdcRandomAccessReader(testCommitLog); + + assertThat(reader).isNotNull(); + assertThat(reader.getPath()).isEqualTo("/test/path/commitlog"); + + // Verify no premature calls before rebuffer is used + assertThat(testSource.requestCalls.size()).isEqualTo(0); + } + + // Test stub classes + + private static class TestCommitLog implements CommitLog Review Comment: I don't think that is possible. cassandra-analytics-cdc depends on `fourzeroBridge(project(path: ':cassandra-four-zero-bridge'))`, which is where this class is being implemented. Re-using it would introduce a circular dependency. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
