diff --git a/storage/inkless/src/main/java/io/aiven/inkless/consume/FetchCompleter.java b/storage/inkless/src/main/java/io/aiven/inkless/consume/FetchCompleter.java index 97ee7975ec9..018b67cd09d 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/consume/FetchCompleter.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/consume/FetchCompleter.java @@ -287,7 +287,9 @@ private static MemoryRecords constructRecordsFromFile( return null; // Doesn't cover entire batch range - incomplete batch } - // All extents cover the batch range, safe to allocate full buffer + // All extents cover the batch range, allocate buffer and copy data. + // Note: We always copy because createMemoryRecords mutates the buffer (setLastOffset, + // setMaxTimestamp), and FileExtent data is cached/shared across concurrent fetches. final byte[] buffer = new byte[Math.toIntExact(batchRange.bufferSize())]; for (FileExtent file : files) { diff --git a/storage/inkless/src/main/java/io/aiven/inkless/consume/FileFetchJob.java b/storage/inkless/src/main/java/io/aiven/inkless/consume/FileFetchJob.java index 3be049fb3a2..a4b2f782ff6 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/consume/FileFetchJob.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/consume/FileFetchJob.java @@ -56,12 +56,27 @@ public FileFetchJob(Time time, // visible for testing static FileExtent createFileExtent(ObjectKey object, ByteRange byteRange, ByteBuffer buffer) { + // Handle both heap and direct/read-only ByteBuffers + // buffer.array() returns the entire backing array and ignores position/limit/arrayOffset, + // so we can only use it directly when the buffer spans the entire array. + byte[] data; + if (buffer.hasArray() + && buffer.arrayOffset() == 0 + && buffer.position() == 0 + && buffer.remaining() == buffer.array().length) { + // Buffer spans the entire backing array - use directly without copy + data = buffer.array(); + } else { + // Copy from direct/read-only buffer or buffer with non-zero position/arrayOffset + data = new byte[buffer.remaining()]; + buffer.get(data); + } return new FileExtent() .setObject(object.value()) .setRange(new FileExtent.ByteRange() .setOffset(byteRange.offset()) - .setLength(buffer.limit())) - .setData(buffer.array()); + .setLength(data.length)) + .setData(data); } @Override @@ -70,7 +85,8 @@ public FileExtent call() throws Exception { } private FileExtent doWork() throws IOException, StorageBackendException { - final ByteBuffer byteBuffer = objectFetcher.readToByteBuffer(objectFetcher.fetch(key, range)); + // Use fetchToByteBuffer for direct ByteBuffer access (avoids channel/stream overhead) + final ByteBuffer byteBuffer = objectFetcher.fetchToByteBuffer(key, range); return createFileExtent(key, range, byteBuffer); } diff --git a/storage/inkless/src/main/java/io/aiven/inkless/produce/FileCommitter.java b/storage/inkless/src/main/java/io/aiven/inkless/produce/FileCommitter.java index 0f54613448c..0a5e32bcf83 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/produce/FileCommitter.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/produce/FileCommitter.java @@ -26,6 +26,7 @@ import java.io.Closeable; import java.io.IOException; +import java.nio.ByteBuffer; import java.time.Duration; import java.time.Instant; import java.util.Collections; @@ -168,15 +169,18 @@ void commit(final ClosedFile file) throws InterruptedException { totalBytesInProgress.addAndGet(file.size()); // Start uploading and add to the commit queue (as Runnable). - // This ensures files are uploaded in concurrently, but committed to the control plane sequentially, + // This ensures files are uploaded concurrently, but committed to the control plane sequentially, // because `executorServiceCommit` is single-threaded. - final FileUploadJob uploadJob = FileUploadJob.createFromByteArray( + // Use ByteBuffer upload path for zero-copy S3 uploads (avoids internal byte[] copy). + // asReadOnlyBuffer() provides defense against accidental modification while + // preserving zero-copy semantics (no memory allocation beyond the view). + final FileUploadJob uploadJob = FileUploadJob.createFromByteBuffer( objectKeyCreator, storage, time, maxFileUploadAttempts, fileUploadRetryBackoff, - file.data(), + ByteBuffer.wrap(file.data()).asReadOnlyBuffer(), metrics::fileUploadFinished ); final Future uploadFuture = executorServiceUpload.submit(uploadJob); diff --git a/storage/inkless/src/main/java/io/aiven/inkless/produce/FileUploadJob.java b/storage/inkless/src/main/java/io/aiven/inkless/produce/FileUploadJob.java index fdec9842b81..9a5de02ae0e 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/produce/FileUploadJob.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/produce/FileUploadJob.java @@ -26,6 +26,7 @@ import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; +import java.nio.ByteBuffer; import java.time.Duration; import java.util.Objects; import java.util.concurrent.Callable; @@ -50,10 +51,14 @@ public class FileUploadJob implements Callable { private final Time time; private final int attempts; private final Duration retryBackoff; - private final Supplier data; + private final Supplier dataStream; + private final ByteBuffer dataBuffer; private final long length; private final Consumer durationCallback; + /** + * Constructor for InputStream-based uploads. + */ public FileUploadJob(final ObjectKeyCreator objectKeyCreator, final ObjectUploader objectUploader, final Time time, @@ -70,11 +75,41 @@ public FileUploadJob(final ObjectKeyCreator objectKeyCreator, } this.attempts = attempts; this.retryBackoff = Objects.requireNonNull(retryBackoff, "retryBackoff cannot be null"); - this.data = Objects.requireNonNull(data, "data cannot be null"); + this.dataStream = Objects.requireNonNull(data, "data cannot be null"); + this.dataBuffer = null; this.length = length; this.durationCallback = Objects.requireNonNull(durationCallback, "durationCallback cannot be null"); } + /** + * Constructor for ByteBuffer-based uploads (zero-copy path). + * Note: The buffer is stored by reference, not copied. The caller must ensure the buffer + * is not modified between construction and when call() completes. The buffer's position + * and limit are captured at upload time via duplicate() in the ObjectUploader implementation, + * which preserves retry support without modifying the original buffer. + */ + private FileUploadJob(final ObjectKeyCreator objectKeyCreator, + final ObjectUploader objectUploader, + final Time time, + final int attempts, + final Duration retryBackoff, + final ByteBuffer dataBuffer, + final Consumer durationCallback) { + this.objectKeyCreator = Objects.requireNonNull(objectKeyCreator, "objectKeyCreator cannot be null"); + this.objectUploader = Objects.requireNonNull(objectUploader, "objectUploader cannot be null"); + this.time = Objects.requireNonNull(time, "time cannot be null"); + if (attempts <= 0) { + throw new IllegalArgumentException("attempts must be positive"); + } + this.attempts = attempts; + this.retryBackoff = Objects.requireNonNull(retryBackoff, "retryBackoff cannot be null"); + this.dataStream = null; + // Store the buffer reference - position/limit are preserved via duplicate() during upload + this.dataBuffer = Objects.requireNonNull(dataBuffer, "dataBuffer cannot be null"); + this.length = dataBuffer.remaining(); + this.durationCallback = Objects.requireNonNull(durationCallback, "durationCallback cannot be null"); + } + public static FileUploadJob createFromByteArray(final ObjectKeyCreator objectKeyCreator, final ObjectUploader objectUploader, final Time time, @@ -93,7 +128,32 @@ public static FileUploadJob createFromByteArray(final ObjectKeyCreator objectKey data.length, durationCallback ); + } + /** + * Creates a FileUploadJob for ByteBuffer data using the zero-copy upload path. + * The ByteBuffer's position will not be modified (uses duplicate internally for retries). + */ + public static FileUploadJob createFromByteBuffer(final ObjectKeyCreator objectKeyCreator, + final ObjectUploader objectUploader, + final Time time, + final int attempts, + final Duration retryBackoff, + final ByteBuffer data, + final Consumer durationCallback) { + Objects.requireNonNull(data, "data cannot be null"); + if (data.remaining() <= 0) { + throw new IllegalArgumentException("data must have remaining bytes"); + } + return new FileUploadJob( + objectKeyCreator, + objectUploader, + time, + attempts, + retryBackoff, + data, + durationCallback + ); } @Override @@ -106,7 +166,17 @@ private ObjectKey callInternal() throws Exception { final Exception uploadError; try { objectKey = objectKeyCreator.create(Uuid.randomUuid().toString()); - uploadError = uploadWithRetry(objectKey, data, length); + if (dataBuffer != null) { + LOGGER.debug("Uploading {} via ByteBuffer (zero-copy)", objectKey); + uploadError = uploadWithRetry(objectKey, () -> objectUploader.upload(objectKey, dataBuffer)); + } else { + LOGGER.debug("Uploading {} via InputStream", objectKey); + uploadError = uploadWithRetry(objectKey, () -> { + try (InputStream stream = dataStream.get()) { + objectUploader.upload(objectKey, stream, length); + } + }); + } } catch (final Exception e) { LOGGER.error("Unexpected exception", e); throw e; @@ -119,32 +189,24 @@ private ObjectKey callInternal() throws Exception { } } - private Exception uploadWithRetry(final ObjectKey objectKey, final Supplier data, final long length) { - LOGGER.debug("Uploading {}", objectKey); + /** + * Executes the upload operation with retry logic. + * @param objectKey the object key being uploaded (for logging) + * @param uploadOperation the upload operation to execute + * @return null on success, or the last exception on failure after all retries exhausted + */ + private Exception uploadWithRetry(final ObjectKey objectKey, final UploadOperation uploadOperation) { Exception error = null; for (int attempt = 0; attempt < attempts; attempt++) { - try (InputStream stream = data.get()) { - objectUploader.upload(objectKey, stream, length); + try { + uploadOperation.execute(); LOGGER.debug("Successfully uploaded {}", objectKey); return null; } catch (final StorageBackendException | IOException e) { error = e; - // Sleep on all attempts but last. final boolean lastAttempt = attempt == attempts - 1; - if (lastAttempt) { - if (e instanceof StorageBackendTimeoutException) { - LOGGER.error("Error uploading {} due to timeout, giving up: {}", objectKey, safeGetCauseMessage(e)); - } else { - LOGGER.error("Error uploading {}, giving up", objectKey, e); - } - } else { - if (e instanceof StorageBackendTimeoutException) { - LOGGER.error("Error uploading {} due to timeout, retrying in {} ms: {}", - objectKey, retryBackoff.toMillis(), safeGetCauseMessage(e)); - } else { - LOGGER.error("Error uploading {}, retrying in {} ms", - objectKey, retryBackoff.toMillis(), e); - } + logRetryableError(objectKey, lastAttempt, e); + if (!lastAttempt) { time.sleep(retryBackoff.toMillis()); } } @@ -152,6 +214,29 @@ private Exception uploadWithRetry(final ObjectKey objectKey, final SupplierThe default implementation falls back to fetch() + readToByteBuffer() for compatibility. + * + * @param key the object key to fetch + * @param range the byte range to fetch, or null for entire object + * @return ByteBuffer containing the fetched data with position at 0 and limit at data length + */ + default ByteBuffer fetchToByteBuffer(ObjectKey key, ByteRange range) throws StorageBackendException, IOException { + try (ReadableByteChannel channel = fetch(key, range)) { + return readToByteBuffer(channel); + } + } + default ByteBuffer readToByteBuffer(final ReadableByteChannel readableByteChannel) throws IOException { final ByteBuffer byteBuffer; final List buffers = new ArrayList<>(5); diff --git a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/common/ObjectUploader.java b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/common/ObjectUploader.java index b540809220d..f401fa2d299 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/common/ObjectUploader.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/common/ObjectUploader.java @@ -17,8 +17,12 @@ */ package io.aiven.inkless.storage_backend.common; +import org.apache.kafka.common.utils.ByteBufferInputStream; + import java.io.Closeable; import java.io.InputStream; +import java.nio.ByteBuffer; +import java.util.Objects; import io.aiven.inkless.common.ObjectKey; @@ -35,4 +39,21 @@ public interface ObjectUploader extends Closeable { */ void upload(ObjectKey key, InputStream inputStream, long length) throws StorageBackendException; + /** + * Uploads an object to object storage from a ByteBuffer. + * The buffer's position will not be modified (uses duplicate internally). + * @param key key of the object to upload. + * @param byteBuffer data of the object that will be uploaded. + * @throws StorageBackendException if there are errors during the upload. + */ + default void upload(ObjectKey key, ByteBuffer byteBuffer) throws StorageBackendException { + Objects.requireNonNull(key, "key cannot be null"); + Objects.requireNonNull(byteBuffer, "byteBuffer cannot be null"); + if (byteBuffer.remaining() <= 0) { + throw new IllegalArgumentException("byteBuffer must have remaining bytes"); + } + final ByteBuffer duplicate = byteBuffer.duplicate(); + upload(key, new ByteBufferInputStream(duplicate), duplicate.remaining()); + } + } diff --git a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/S3Storage.java b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/S3Storage.java index 73bcf886659..96330bdd17a 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/S3Storage.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/S3Storage.java @@ -24,6 +24,7 @@ import java.io.IOException; import java.io.InputStream; +import java.nio.ByteBuffer; import java.nio.channels.Channels; import java.nio.channels.ReadableByteChannel; import java.util.ArrayList; @@ -104,12 +105,63 @@ public void upload(final ObjectKey key, final InputStream inputStream, final lon } @Override - public ReadableByteChannel fetch(final ObjectKey key, final ByteRange range) throws StorageBackendException, IOException { + public void upload(final ObjectKey key, final ByteBuffer byteBuffer) throws StorageBackendException { + Objects.requireNonNull(key, "key cannot be null"); + Objects.requireNonNull(byteBuffer, "byteBuffer cannot be null"); + if (byteBuffer.remaining() <= 0) { + throw new IllegalArgumentException("byteBuffer must have remaining bytes"); + } + final PutObjectRequest putObjectRequest = PutObjectRequest.builder() + .bucket(bucketName) + .key(key.value()) + .build(); + // Use ByteBufferInputStream with duplicate() to avoid: + // 1. The byte[] copy in RequestBody.fromByteBuffer() -> BinaryUtils.copyAllBytesFrom() + // 2. Modifying the original buffer's position (preserves retry support in FileUploadJob) + final ByteBuffer duplicate = byteBuffer.duplicate(); + final long length = duplicate.remaining(); + final RequestBody requestBody = RequestBody.fromInputStream( + new ByteBufferInputStream(duplicate), + length + ); try { - if (range != null && range.empty()) { - return Channels.newChannel(InputStream.nullInputStream()); - } + s3Client.putObject(putObjectRequest, requestBody); + } catch (final ApiCallTimeoutException | ApiCallAttemptTimeoutException e) { + throw new StorageBackendTimeoutException("Failed to upload " + key, e); + } catch (final SdkException e) { + throw new StorageBackendException("Failed to upload " + key, e); + } + } + @Override + public ReadableByteChannel fetch(final ObjectKey key, final ByteRange range) throws StorageBackendException, IOException { + if (range != null && range.empty()) { + return Channels.newChannel(InputStream.nullInputStream()); + } + final var buffer = doFetch(key, range); + return Channels.newChannel(new ByteBufferInputStream(buffer)); + } + + /** + * Optimized fetch that returns ByteBuffer directly without intermediate channel/stream copies. + * S3 SDK already provides the data as a ByteBuffer via getObjectAsBytes(), so we return it directly. + */ + @Override + public ByteBuffer fetchToByteBuffer(final ObjectKey key, final ByteRange range) throws StorageBackendException { + if (range != null && range.empty()) { + return ByteBuffer.allocate(0); + } + return doFetch(key, range); + } + + /** + * Shared fetch implementation that retrieves object data as ByteBuffer. + * For the small 4-8MiB blobs expected here, reading the whole object into memory is more efficient + * than streaming it via S3ObjectInputStream which has significant overhead per read call. + */ + private ByteBuffer doFetch(final ObjectKey key, final ByteRange range) throws StorageBackendException { + Objects.requireNonNull(key, "key cannot be null"); + try { var builder = GetObjectRequest.builder() .bucket(bucketName) .key(key.value()); @@ -117,11 +169,7 @@ public ReadableByteChannel fetch(final ObjectKey key, final ByteRange range) thr builder = builder.range(formatRange(range)); } final GetObjectRequest getRequest = builder.build(); - // for the small 4-8MiB blobs expected here, reading the whole object into memory is more efficient - // than streaming it via S3ObjectInputStream which has significant overhead per read call - // and does not play well with the buffering done in ObjectFetcher.readToByteBuffer() - final var buffer = s3Client.getObjectAsBytes(getRequest).asByteBuffer(); - return Channels.newChannel(new ByteBufferInputStream(buffer)); + return s3Client.getObjectAsBytes(getRequest).asByteBuffer(); } catch (final AwsServiceException e) { if (e.statusCode() == 404) { throw new KeyNotFoundException(this, key, e); @@ -129,7 +177,6 @@ public ReadableByteChannel fetch(final ObjectKey key, final ByteRange range) thr if (e.statusCode() == 416) { throw new InvalidRangeException("Failed to fetch " + key + ": Invalid range " + range, e); } - throw new StorageBackendException("Failed to fetch " + key, e); } catch (final ApiCallTimeoutException | ApiCallAttemptTimeoutException e) { throw new StorageBackendTimeoutException("Failed to fetch " + key, e); diff --git a/storage/inkless/src/test/java/io/aiven/inkless/consume/FetchCompleterTest.java b/storage/inkless/src/test/java/io/aiven/inkless/consume/FetchCompleterTest.java index a6da71dd61f..5e54c824aaf 100644 --- a/storage/inkless/src/test/java/io/aiven/inkless/consume/FetchCompleterTest.java +++ b/storage/inkless/src/test/java/io/aiven/inkless/consume/FetchCompleterTest.java @@ -259,6 +259,7 @@ public void testFetchMultipleFilesForSameBatch() { var endOffset = startOffset + length; ByteBuffer copy = ByteBuffer.allocate(length); copy.put(records.buffer().duplicate().position(startOffset).limit(endOffset).slice()); + copy.flip(); // Reset position to 0 for reading fileExtents.add(new FileExtentResult.Success(OBJECT_KEY_A, range, FileFetchJob.createFileExtent(OBJECT_KEY_A, range, copy))); } @@ -297,6 +298,7 @@ public void testFetchMultipleBatches() { ByteBuffer concatenatedBuffer = ByteBuffer.allocate(totalSize); concatenatedBuffer.put(recordsA.buffer()); concatenatedBuffer.put(recordsB.buffer()); + concatenatedBuffer.flip(); // Reset position to 0 for reading Map fetchInfos = Map.of( partition0, new FetchRequest.PartitionData(topicId, 0, 0, 1000, Optional.empty()) @@ -378,6 +380,7 @@ public void testFetchMultipleFilesForMultipleBatches() { var endOffset = startOffset + length; ByteBuffer copy = ByteBuffer.allocate(blockSize); copy.put(concatenatedBuffer.duplicate().position(startOffset).limit(endOffset).slice()); + copy.flip(); // Reset position to 0 for reading fileExtents.add(new FileExtentResult.Success(OBJECT_KEY_A, range, FileFetchJob.createFileExtent(OBJECT_KEY_A, range, copy))); } @@ -592,6 +595,7 @@ public void testSingleBatchWithMultipleExtentsAllSucceed() { var endOffset = startOffset + length; ByteBuffer copy = ByteBuffer.allocate(length); copy.put(records.buffer().duplicate().position(startOffset).limit(endOffset).slice()); + copy.flip(); // Reset position to 0 for reading fileExtents.add(new FileExtentResult.Success(OBJECT_KEY_A, range, FileFetchJob.createFileExtent(OBJECT_KEY_A, range, copy))); } @@ -662,6 +666,7 @@ public void testSingleBatchWithMissingMiddleExtentFails() { var endOffset = startOffset + length; ByteBuffer copy = ByteBuffer.allocate(length); copy.put(records.buffer().duplicate().position(startOffset).limit(endOffset).slice()); + copy.flip(); // Reset position to 0 for reading fileExtents.add(new FileExtentResult.Success(OBJECT_KEY_A, range, FileFetchJob.createFileExtent(OBJECT_KEY_A, range, copy))); } @@ -730,6 +735,7 @@ public void testSingleBatchWithFailedMiddleExtentFails() { var endOffset = startOffset + length; ByteBuffer copy = ByteBuffer.allocate(length); copy.put(records.buffer().duplicate().position(startOffset).limit(endOffset).slice()); + copy.flip(); // Reset position to 0 for reading fileExtents.add(new FileExtentResult.Success(OBJECT_KEY_A, range, FileFetchJob.createFileExtent(OBJECT_KEY_A, range, copy))); } diff --git a/storage/inkless/src/test/java/io/aiven/inkless/consume/FetchPlannerTest.java b/storage/inkless/src/test/java/io/aiven/inkless/consume/FetchPlannerTest.java index fced294f9ad..7dd881a8c0d 100644 --- a/storage/inkless/src/test/java/io/aiven/inkless/consume/FetchPlannerTest.java +++ b/storage/inkless/src/test/java/io/aiven/inkless/consume/FetchPlannerTest.java @@ -350,17 +350,10 @@ public void testMultipleAsyncFetchOperations() throws Exception { final byte[] dataA = "data-for-a".getBytes(); final byte[] dataB = "data-for-b".getBytes(); - // Mock the fetcher's two-step process: fetch() is called first, then readToByteBuffer() - // For this test, we only care about the final data returned by readToByteBuffer() - when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))) - .thenReturn(null); // Return value doesn't matter, readToByteBuffer() is also mocked - when(fetcher.fetch(eq(OBJECT_KEY_B), any(ByteRange.class))) - .thenReturn(null); // Return value doesn't matter, readToByteBuffer() is also mocked - - // Mock readToByteBuffer to return the test data we want to verify - // Order matters: first call returns dataA, second call returns dataB - when(fetcher.readToByteBuffer(any())) - .thenReturn(ByteBuffer.wrap(dataA)) + // Mock fetchToByteBuffer to return the test data directly + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class))) + .thenReturn(ByteBuffer.wrap(dataA)); + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_B), any(ByteRange.class))) .thenReturn(ByteBuffer.wrap(dataB)); final Map coordinates = Map.of( @@ -387,8 +380,8 @@ public void testMultipleAsyncFetchOperations() throws Exception { CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get(); // Verify both were fetched - verify(fetcher).fetch(eq(OBJECT_KEY_A), any(ByteRange.class)); - verify(fetcher).fetch(eq(OBJECT_KEY_B), any(ByteRange.class)); + verify(fetcher).fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class)); + verify(fetcher).fetchToByteBuffer(eq(OBJECT_KEY_B), any(ByteRange.class)); // Verify correct data for each final List results = futures.stream() @@ -419,10 +412,8 @@ public void testCacheMiss() throws Exception { final byte[] expectedData = "test-data".getBytes(); final ByteBuffer byteBuffer = ByteBuffer.wrap(expectedData); - // Mock the fetcher to return data via ByteBuffer - when(fetcher.fetch(any(ObjectKey.class), any(ByteRange.class))) - .thenReturn(null); // channel not used directly - when(fetcher.readToByteBuffer(any())) + // Mock fetchToByteBuffer to return data directly + when(fetcher.fetchToByteBuffer(any(ObjectKey.class), any(ByteRange.class))) .thenReturn(byteBuffer); final Map coordinates = Map.of( @@ -449,7 +440,7 @@ public void testCacheMiss() throws Exception { assertThat(result.data()).isEqualTo(expectedData); // Verify remote fetch was called (cache miss) - verify(fetcher).fetch(any(ObjectKey.class), any(ByteRange.class)); + verify(fetcher).fetchToByteBuffer(any(ObjectKey.class), any(ByteRange.class)); // Verify the result is now in cache final ObjectFetchRequest request = new ObjectFetchRequest( @@ -498,7 +489,7 @@ public void testCacheHit() throws Exception { assertThat(result.data()).isEqualTo(expectedData); // Verify remote fetch was NOT called (cache hit) - verify(fetcher, never()).fetch(any(ObjectKey.class), any(ByteRange.class)); + verify(fetcher, never()).fetchToByteBuffer(any(ObjectKey.class), any(ByteRange.class)); } } @@ -508,7 +499,7 @@ public void testFetchFailure() throws Exception { try (CaffeineCache caffeineCache = new CaffeineCache(100, 3600, 180)) { // Mock fetcher to throw exception - when(fetcher.fetch(any(ObjectKey.class), any(ByteRange.class))) + when(fetcher.fetchToByteBuffer(any(ObjectKey.class), any(ByteRange.class))) .thenThrow(new RuntimeException("S3 unavailable")); final Map coordinates = Map.of( @@ -535,7 +526,7 @@ public void testFetchFailure() throws Exception { .hasCauseInstanceOf(FileFetchException.class); // Verify remote fetch was attempted - verify(fetcher).fetch(any(ObjectKey.class), any(ByteRange.class)); + verify(fetcher).fetchToByteBuffer(any(ObjectKey.class), any(ByteRange.class)); } } @@ -550,10 +541,8 @@ public void testFailedFetchesAreRetried() throws Exception { final byte[] expectedData = "recovered-data".getBytes(); // First call fails, second call succeeds - when(fetcher.fetch(any(ObjectKey.class), any(ByteRange.class))) + when(fetcher.fetchToByteBuffer(any(ObjectKey.class), any(ByteRange.class))) .thenThrow(new RuntimeException("Transient S3 error")) - .thenReturn(null); - when(fetcher.readToByteBuffer(any())) .thenReturn(ByteBuffer.wrap(expectedData)); final Map coordinates = Map.of( @@ -587,7 +576,7 @@ public void testFailedFetchesAreRetried() throws Exception { assertThat(result.data()).isEqualTo(expectedData); // Verify fetch was called twice (once for failure, once for success) - verify(fetcher, times(2)).fetch(any(ObjectKey.class), any(ByteRange.class)); + verify(fetcher, times(2)).fetchToByteBuffer(any(ObjectKey.class), any(ByteRange.class)); } } @@ -603,10 +592,8 @@ public void testConcurrentRequestsToSameKeyFetchOnlyOnce() throws Exception { try (CaffeineCache caffeineCache = new CaffeineCache(100, 3600, 180)) { final byte[] expectedData = "shared-data".getBytes(); - // Mock fetcher to return data - when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))) - .thenReturn(null); - when(fetcher.readToByteBuffer(any())) + // Mock fetchToByteBuffer to return data directly + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class))) .thenReturn(ByteBuffer.wrap(expectedData)); // Create coordinates with TWO batches that map to the SAME cache key @@ -640,7 +627,7 @@ public void testConcurrentRequestsToSameKeyFetchOnlyOnce() throws Exception { assertThat(result.data()).isEqualTo(expectedData); // Verify fetch was called **only once** despite multiple requests - verify(fetcher).fetch(eq(OBJECT_KEY_A), any(ByteRange.class)); + verify(fetcher).fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class)); } } @@ -653,10 +640,9 @@ public void testMetricsAreRecordedCorrectly() throws Exception { final byte[] dataA = "data-a".getBytes(); final byte[] dataB = "data-bb".getBytes(); - when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(null); - when(fetcher.fetch(eq(OBJECT_KEY_B), any(ByteRange.class))).thenReturn(null); - when(fetcher.readToByteBuffer(any())) - .thenReturn(ByteBuffer.wrap(dataA)) + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class))) + .thenReturn(ByteBuffer.wrap(dataA)); + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_B), any(ByteRange.class))) .thenReturn(ByteBuffer.wrap(dataB)); final Map coordinates = Map.of( @@ -700,8 +686,8 @@ public void testOldDataUsesHotPathWhenLaggingConsumerFeatureDisabled() throws Ex try (CaffeineCache caffeineCache = new CaffeineCache(100, 3600, 180)) { final byte[] expectedData = "old-data-but-hot-path".getBytes(); - when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(null); - when(fetcher.readToByteBuffer(any())).thenReturn(ByteBuffer.wrap(expectedData)); + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class))) + .thenReturn(ByteBuffer.wrap(expectedData)); // Very old timestamp - would be "lagging" if feature was enabled final long veryOldTimestamp = time.milliseconds() - 3600_000L; // 1 hour ago @@ -730,7 +716,7 @@ public void testOldDataUsesHotPathWhenLaggingConsumerFeatureDisabled() throws Ex verify(metrics, never()).recordRateLimitWaitTime(any(Long.class)); // Verify data was fetched successfully - verify(fetcher).fetch(eq(OBJECT_KEY_A), any(ByteRange.class)); + verify(fetcher).fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class)); } } @@ -757,7 +743,7 @@ public void testExecutionWithEmptyBatches() throws Exception { assertThat(futures).isEmpty(); // Verify no fetch operations were attempted - verify(fetcher, never()).fetch(any(ObjectKey.class), any(ByteRange.class)); + verify(fetcher, never()).fetchToByteBuffer(any(ObjectKey.class), any(ByteRange.class)); // Verify metrics were still recorded (batch size = 0) verify(metrics).recordFetchBatchSize(0); @@ -782,8 +768,8 @@ public void recentDataUsesRecentExecutorWithoutRateLimit() throws Exception { .addLimit(limit -> limit.capacity(1).refillGreedy(1, java.time.Duration.ofSeconds(1))) .build(); - when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(null); - when(fetcher.readToByteBuffer(any())).thenReturn(ByteBuffer.wrap(expectedData)); + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class))) + .thenReturn(ByteBuffer.wrap(expectedData)); final long recentTimestamp = time.milliseconds() - 30000L; // 30 seconds ago (recent) final Map coordinates = Map.of( @@ -822,8 +808,8 @@ public void boundaryConditionExactlyAtThreshold() throws Exception { final byte[] expectedData = "boundary-data".getBytes(); final long threshold = 60 * 1000L; - when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(null); - when(fetcher.readToByteBuffer(any())).thenReturn(ByteBuffer.wrap(expectedData)); + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class))) + .thenReturn(ByteBuffer.wrap(expectedData)); final long exactThresholdTimestamp = time.milliseconds() - threshold; final Map coordinates = Map.of( @@ -864,8 +850,8 @@ public void laggingDataUsesLaggingExecutorWithRateLimit() throws Exception { .addLimit(limit -> limit.capacity(10).refillGreedy(10, java.time.Duration.ofSeconds(1))) .build(); - when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(null); - when(fetcher.readToByteBuffer(any())).thenReturn(ByteBuffer.wrap(expectedData)); + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class))) + .thenReturn(ByteBuffer.wrap(expectedData)); final long oldTimestamp = time.milliseconds() - 120000L; // 2 minutes ago (old) final Map coordinates = Map.of( @@ -904,8 +890,8 @@ public void laggingDataWithoutRateLimiter() throws Exception { try (CaffeineCache caffeineCache = new CaffeineCache(100, 3600, 180)) { final byte[] expectedData = "old-data-no-limit".getBytes(); - when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(null); - when(fetcher.readToByteBuffer(any())).thenReturn(ByteBuffer.wrap(expectedData)); + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class))) + .thenReturn(ByteBuffer.wrap(expectedData)); final long oldTimestamp = time.milliseconds() - 120000L; final Map coordinates = Map.of( @@ -942,7 +928,7 @@ public void fetchFailureInColdPathPropagatesException() throws Exception { // Test that fetch failures in the cold path are properly wrapped and propagated try (CaffeineCache caffeineCache = new CaffeineCache(100, 3600, 180)) { // Mock fetcher to throw exception - when(fetcher.fetch(any(ObjectKey.class), any(ByteRange.class))) + when(fetcher.fetchToByteBuffer(any(ObjectKey.class), any(ByteRange.class))) .thenThrow(new RuntimeException("S3 unavailable")); final long oldTimestamp = time.milliseconds() - 120000L; @@ -976,7 +962,7 @@ public void fetchFailureInColdPathPropagatesException() throws Exception { verify(metrics, never()).recordRecentDataRequest(); // Verify remote fetch was attempted - verify(fetcher).fetch(any(ObjectKey.class), any(ByteRange.class)); + verify(fetcher).fetchToByteBuffer(any(ObjectKey.class), any(ByteRange.class)); } } @@ -1159,10 +1145,9 @@ public void multipleRequestsMixedHotAndColdPaths() throws Exception { final byte[] oldData = "old".getBytes(); final long threshold = 60 * 1000L; - when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(null); - when(fetcher.fetch(eq(OBJECT_KEY_B), any(ByteRange.class))).thenReturn(null); - when(fetcher.readToByteBuffer(any())) - .thenReturn(ByteBuffer.wrap(recentData)) + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class))) + .thenReturn(ByteBuffer.wrap(recentData)); + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_B), any(ByteRange.class))) .thenReturn(ByteBuffer.wrap(oldData)); final long recentTimestamp = time.milliseconds() - 30000L; // 30s ago (recent) @@ -1210,10 +1195,9 @@ public void hotAndColdPathsExecuteConcurrently() throws Exception { final ExecutorService coldExecutor = Executors.newFixedThreadPool(2); try { - when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(null); - when(fetcher.fetch(eq(OBJECT_KEY_B), any(ByteRange.class))).thenReturn(null); - when(fetcher.readToByteBuffer(any())) - .thenReturn(ByteBuffer.wrap(recentData)) + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class))) + .thenReturn(ByteBuffer.wrap(recentData)); + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_B), any(ByteRange.class))) .thenReturn(ByteBuffer.wrap(oldData)); final long recentTimestamp = time.milliseconds() - 30000L; // 30s ago (recent) @@ -1287,8 +1271,8 @@ public void allRequestsUseRecentPathWhenFeatureDisabled() throws Exception { // Validates: laggingConsumerExecutor = null → feature disabled, all use hot path try (CaffeineCache caffeineCache = new CaffeineCache(100, 3600, 180)) { final byte[] expectedData = "all-recent".getBytes(); - when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(null); - when(fetcher.readToByteBuffer(any())).thenReturn(ByteBuffer.wrap(expectedData)); + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class))) + .thenReturn(ByteBuffer.wrap(expectedData)); final long oldTimestamp = time.milliseconds() - 120000L; // Would be lagging if feature enabled final Map coordinates = Map.of( @@ -1320,8 +1304,8 @@ public void rateLimiterCanBeDisabledIndependently() throws Exception { // Validates: rateLimiter = null → cold path without rate limiting try (CaffeineCache caffeineCache = new CaffeineCache(100, 3600, 180)) { final byte[] expectedData = "cold-no-limit".getBytes(); - when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(null); - when(fetcher.readToByteBuffer(any())).thenReturn(ByteBuffer.wrap(expectedData)); + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class))) + .thenReturn(ByteBuffer.wrap(expectedData)); final long oldTimestamp = time.milliseconds() - 120000L; final Map coordinates = Map.of( @@ -1357,8 +1341,8 @@ public void bothFeaturesCanBeEnabled() throws Exception { final Bucket rateLimiter = Bucket.builder() .addLimit(limit -> limit.capacity(10).refillGreedy(10, java.time.Duration.ofSeconds(1))) .build(); - when(fetcher.fetch(eq(OBJECT_KEY_A), any(ByteRange.class))).thenReturn(null); - when(fetcher.readToByteBuffer(any())).thenReturn(ByteBuffer.wrap(expectedData)); + when(fetcher.fetchToByteBuffer(eq(OBJECT_KEY_A), any(ByteRange.class))) + .thenReturn(ByteBuffer.wrap(expectedData)); final long oldTimestamp = time.milliseconds() - 120000L; final Map coordinates = Map.of( diff --git a/storage/inkless/src/test/java/io/aiven/inkless/consume/FileFetchJobTest.java b/storage/inkless/src/test/java/io/aiven/inkless/consume/FileFetchJobTest.java index a7cda9ecc5f..4f5acb82484 100644 --- a/storage/inkless/src/test/java/io/aiven/inkless/consume/FileFetchJobTest.java +++ b/storage/inkless/src/test/java/io/aiven/inkless/consume/FileFetchJobTest.java @@ -28,7 +28,6 @@ import org.mockito.quality.Strictness; import java.nio.ByteBuffer; -import java.nio.channels.ReadableByteChannel; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -42,7 +41,6 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertThrows; -import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @ExtendWith(MockitoExtension.class) @@ -71,9 +69,8 @@ public void testFetch() throws Exception { FileFetchJob job = new FileFetchJob(time, fetcher, objectA, range, durationMs -> { }); FileExtent expectedFile = FileFetchJob.createFileExtent(objectA, range, ByteBuffer.wrap(array)); - final ReadableByteChannel channel = mock(ReadableByteChannel.class); - when(fetcher.fetch(objectA, range)).thenReturn(channel); - when(fetcher.readToByteBuffer(channel)).thenReturn(ByteBuffer.wrap(array)); + // FileFetchJob now uses fetchToByteBuffer directly for better performance + when(fetcher.fetchToByteBuffer(objectA, range)).thenReturn(ByteBuffer.wrap(array)); FileExtent actualFile = job.call(); assertThat(actualFile).isEqualTo(expectedFile); @@ -150,4 +147,81 @@ public void testSingleFileExtentLessThanBlockSize() { assertThat(fileRanges).containsExactlyInAnyOrderElementsOf(expectedRanges); } + @Test + public void testCreateFileExtentWithDirectByteBuffer() { + // Direct ByteBuffers don't support .array() - this test verifies the fix handles them correctly + byte[] expectedData = {1, 2, 3, 4, 5}; + ByteBuffer directBuffer = ByteBuffer.allocateDirect(expectedData.length); + directBuffer.put(expectedData); + directBuffer.flip(); + + ByteRange range = new ByteRange(100, expectedData.length); + FileExtent extent = FileFetchJob.createFileExtent(objectA, range, directBuffer); + + assertThat(extent.object()).isEqualTo(objectA.value()); + assertThat(extent.range().offset()).isEqualTo(100); + assertThat(extent.range().length()).isEqualTo(expectedData.length); + assertThat(extent.data()).isEqualTo(expectedData); + } + + @Test + public void testCreateFileExtentWithReadOnlyByteBuffer() { + // Read-only ByteBuffers also don't support .array() + byte[] expectedData = {10, 20, 30, 40}; + ByteBuffer readOnlyBuffer = ByteBuffer.wrap(expectedData).asReadOnlyBuffer(); + + ByteRange range = new ByteRange(50, expectedData.length); + FileExtent extent = FileFetchJob.createFileExtent(objectA, range, readOnlyBuffer); + + assertThat(extent.object()).isEqualTo(objectA.value()); + assertThat(extent.range().offset()).isEqualTo(50); + assertThat(extent.range().length()).isEqualTo(expectedData.length); + assertThat(extent.data()).isEqualTo(expectedData); + } + + @Test + public void testCreateFileExtentWithNonZeroPosition() { + // ByteBuffer with non-zero position should only return remaining bytes + byte[] backingArray = {0, 0, 1, 2, 3, 4, 5}; + ByteBuffer buffer = ByteBuffer.wrap(backingArray); + buffer.position(2); // Skip first 2 bytes + + byte[] expectedData = {1, 2, 3, 4, 5}; + ByteRange range = new ByteRange(0, expectedData.length); + FileExtent extent = FileFetchJob.createFileExtent(objectA, range, buffer); + + assertThat(extent.range().length()).isEqualTo(expectedData.length); + assertThat(extent.data()).isEqualTo(expectedData); + } + + @Test + public void testCreateFileExtentWithSlicedBuffer() { + // Sliced buffers have non-zero arrayOffset - this tests the arrayOffset handling + byte[] backingArray = {0, 0, 10, 20, 30, 0, 0}; + ByteBuffer original = ByteBuffer.wrap(backingArray); + original.position(2); + original.limit(5); + ByteBuffer sliced = original.slice(); // Creates buffer with arrayOffset=2 + + byte[] expectedData = {10, 20, 30}; + ByteRange range = new ByteRange(0, expectedData.length); + FileExtent extent = FileFetchJob.createFileExtent(objectA, range, sliced); + + assertThat(extent.range().length()).isEqualTo(expectedData.length); + assertThat(extent.data()).isEqualTo(expectedData); + } + + @Test + public void testCreateFileExtentWithHeapBufferSpanningEntireArray() { + // When buffer spans entire backing array, we can use array() directly (zero-copy) + byte[] expectedData = {1, 2, 3, 4, 5}; + ByteBuffer buffer = ByteBuffer.wrap(expectedData); + + ByteRange range = new ByteRange(0, expectedData.length); + FileExtent extent = FileFetchJob.createFileExtent(objectA, range, buffer); + + assertThat(extent.range().length()).isEqualTo(expectedData.length); + assertThat(extent.data()).isSameAs(expectedData); // Same reference - no copy + } + } diff --git a/storage/inkless/src/test/java/io/aiven/inkless/consume/ReaderTest.java b/storage/inkless/src/test/java/io/aiven/inkless/consume/ReaderTest.java index cfbcc6d80ee..85edef63594 100644 --- a/storage/inkless/src/test/java/io/aiven/inkless/consume/ReaderTest.java +++ b/storage/inkless/src/test/java/io/aiven/inkless/consume/ReaderTest.java @@ -44,7 +44,6 @@ import java.io.IOException; import java.nio.ByteBuffer; -import java.nio.channels.ReadableByteChannel; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Collections; @@ -82,7 +81,6 @@ import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.atLeastOnce; -import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoInteractions; @@ -575,7 +573,7 @@ public void testFileFetchException() throws Exception { .thenReturn(List.of(singleResponse)); // Simulate fetcher failing and throwing an exception - when(objectFetcher.fetch(any(ObjectKey.class), any(ByteRange.class))) + when(objectFetcher.fetchToByteBuffer(any(ObjectKey.class), any(ByteRange.class))) .thenThrow(new StorageBackendException("Storage backend error")); try (final var reader = getReader()) { @@ -616,11 +614,9 @@ public void testFetchException() throws Exception { .thenReturn(List.of(singleResponse)); // Simulate fetcher returning invalid/corrupted data that doesn't match expected size - final ReadableByteChannel file1Channel = mock(ReadableByteChannel.class); - when(objectFetcher.fetch(any(), any())).thenReturn(file1Channel); // Corrupted data with size that doesn't match expected batch size final ByteBuffer corruptedRecords = ByteBuffer.wrap("invalid-batch-data".getBytes(StandardCharsets.UTF_8)); - when(objectFetcher.readToByteBuffer(file1Channel)).thenReturn(corruptedRecords); + when(objectFetcher.fetchToByteBuffer(any(), any())).thenReturn(corruptedRecords); try (final var reader = getReader()) { final CompletableFuture> fetch = reader.fetch(fetchParams, fetchInfos); @@ -653,9 +649,7 @@ public void testSuccessfulFetchMetrics() throws Exception { .thenReturn(List.of(singleResponse)); // Simulate fetcher returning valid data - final ReadableByteChannel file1Channel = mock(ReadableByteChannel.class); - when(objectFetcher.fetch(any(), any())).thenReturn(file1Channel); - when(objectFetcher.readToByteBuffer(file1Channel)).thenReturn(records.buffer()); + when(objectFetcher.fetchToByteBuffer(any(), any())).thenReturn(records.buffer()); try (final var reader = getReader()) { final CompletableFuture> fetch = reader.fetch(fetchParams, fetchInfos); @@ -765,9 +759,7 @@ public void testRateLimitingWithLoad() throws Exception { when(controlPlane.findBatches(any(), anyInt(), anyInt())) .thenReturn(List.of(oldResponse)); - final ReadableByteChannel channel = mock(ReadableByteChannel.class); - when(objectFetcher.fetch(any(), any())).thenReturn(channel); - when(objectFetcher.readToByteBuffer(channel)).thenReturn(records.buffer()); + when(objectFetcher.fetchToByteBuffer(any(), any())).thenReturn(records.buffer()); try (final var reader = new Reader( time, @@ -844,9 +836,7 @@ public void testRateLimitingDisabled() throws Exception { when(controlPlane.findBatches(any(), anyInt(), anyInt())) .thenReturn(List.of(oldResponse)); - final ReadableByteChannel channel = mock(ReadableByteChannel.class); - when(objectFetcher.fetch(any(), any())).thenReturn(channel); - when(objectFetcher.readToByteBuffer(channel)).thenReturn(records.buffer()); + when(objectFetcher.fetchToByteBuffer(any(), any())).thenReturn(records.buffer()); try (final var reader = new Reader( time, @@ -970,10 +960,9 @@ public void testMixedLaggingAndRecentPartitions() throws Exception { }); // Setup object fetcher to succeed for all requests (hot path will use this) - final ReadableByteChannel channel = mock(ReadableByteChannel.class); - when(objectFetcher.fetch(any(ObjectKey.class), any(ByteRange.class))).thenReturn(channel); // Return a fresh buffer each time to avoid buffer exhaustion issues - when(objectFetcher.readToByteBuffer(channel)).thenAnswer(invocation -> records.buffer().duplicate()); + when(objectFetcher.fetchToByteBuffer(any(ObjectKey.class), any(ByteRange.class))) + .thenAnswer(invocation -> records.buffer().duplicate()); // Create a lagging executor and immediately shut it down - will reject all tasks final ExecutorService saturatedLaggingExecutor = Executors.newSingleThreadExecutor(); @@ -1074,9 +1063,7 @@ public void testRecentDataBypassesRateLimiting() throws Exception { when(controlPlane.findBatches(any(), anyInt(), anyInt())) .thenReturn(List.of(recentResponse)); - final ReadableByteChannel channel = mock(ReadableByteChannel.class); - when(objectFetcher.fetch(any(), any())).thenReturn(channel); - when(objectFetcher.readToByteBuffer(channel)).thenReturn(records.buffer()); + when(objectFetcher.fetchToByteBuffer(any(), any())).thenReturn(records.buffer()); try (final var reader = new Reader( time, diff --git a/storage/inkless/src/test/java/io/aiven/inkless/produce/FileCommitterTest.java b/storage/inkless/src/test/java/io/aiven/inkless/produce/FileCommitterTest.java index e89b18c3c0c..0b77ae5587e 100644 --- a/storage/inkless/src/test/java/io/aiven/inkless/produce/FileCommitterTest.java +++ b/storage/inkless/src/test/java/io/aiven/inkless/produce/FileCommitterTest.java @@ -33,7 +33,8 @@ import org.mockito.quality.Strictness; import java.io.IOException; -import java.io.InputStream; +import java.nio.ByteBuffer; +import java.nio.ReadOnlyBufferException; import java.time.Duration; import java.time.Instant; import java.util.List; @@ -116,12 +117,14 @@ public ObjectKey create(String value) { ArgumentCaptor> uploadCallableCaptor; @Captor ArgumentCaptor commitRunnableCaptor; + @Captor + ArgumentCaptor byteBufferCaptor; @Test @SuppressWarnings("unchecked") void success() throws Exception { doNothing() - .when(storage).upload(eq(OBJECT_KEY), any(InputStream.class), eq((long) FILE.data().length)); + .when(storage).upload(eq(OBJECT_KEY), any(ByteBuffer.class)); when(time.nanoseconds()).thenReturn(10_000_000L); @@ -170,7 +173,7 @@ void success() throws Exception { @SuppressWarnings("unchecked") void commitFailed() throws Exception { doNothing() - .when(storage).upload(eq(OBJECT_KEY), any(InputStream.class), eq((long) FILE.data().length)); + .when(storage).upload(eq(OBJECT_KEY), any(ByteBuffer.class)); when(time.nanoseconds()).thenReturn(10_000_000L); @@ -221,7 +224,7 @@ void commitFailed() throws Exception { @SuppressWarnings("unchecked") void uploadFailed() throws Exception { doNothing() - .when(storage).upload(eq(OBJECT_KEY), any(InputStream.class), eq((long) FILE.data().length)); + .when(storage).upload(eq(OBJECT_KEY), any(ByteBuffer.class)); when(time.nanoseconds()).thenReturn(10_000_000L); @@ -396,4 +399,69 @@ void commitNull() { .isInstanceOf(NullPointerException.class) .hasMessage("file cannot be null"); } + + @Test + @SuppressWarnings("unchecked") + void usesByteBufferUpload() throws Exception { + doNothing() + .when(storage).upload(eq(OBJECT_KEY), any(ByteBuffer.class)); + + when(time.nanoseconds()).thenReturn(10_000_000L); + + final CompletableFuture uploadFuture = CompletableFuture.completedFuture(OBJECT_KEY); + when(executorServiceUpload.submit(any(Callable.class))) + .thenReturn(uploadFuture); + + final FileCommitter committer = new FileCommitter( + BROKER_ID, controlPlane, OBJECT_KEY_CREATOR, storage, + KEY_ALIGNMENT_STRATEGY, OBJECT_CACHE, BATCH_COORDINATE_CACHE, time, + 3, Duration.ofMillis(100), + executorServiceUpload, executorServiceCommit, executorServiceCacheStore, + metrics); + + committer.commit(FILE); + + verify(executorServiceUpload).submit(uploadCallableCaptor.capture()); + final Callable uploadCallable = uploadCallableCaptor.getValue(); + + uploadCallable.call(); + + // Verify ByteBuffer-based upload was used + verify(storage).upload(eq(OBJECT_KEY), any(ByteBuffer.class)); + } + + @Test + @SuppressWarnings("unchecked") + void passesReadOnlyBufferToStorage() throws Exception { + doNothing() + .when(storage).upload(eq(OBJECT_KEY), byteBufferCaptor.capture()); + + when(time.nanoseconds()).thenReturn(10_000_000L); + + final CompletableFuture uploadFuture = CompletableFuture.completedFuture(OBJECT_KEY); + when(executorServiceUpload.submit(any(Callable.class))) + .thenReturn(uploadFuture); + + final FileCommitter committer = new FileCommitter( + BROKER_ID, controlPlane, OBJECT_KEY_CREATOR, storage, + KEY_ALIGNMENT_STRATEGY, OBJECT_CACHE, BATCH_COORDINATE_CACHE, time, + 3, Duration.ofMillis(100), + executorServiceUpload, executorServiceCommit, executorServiceCacheStore, + metrics); + + committer.commit(FILE); + + verify(executorServiceUpload).submit(uploadCallableCaptor.capture()); + final Callable uploadCallable = uploadCallableCaptor.getValue(); + + uploadCallable.call(); + + // Verify the ByteBuffer passed to storage is read-only + final ByteBuffer capturedBuffer = byteBufferCaptor.getValue(); + assertThat(capturedBuffer.isReadOnly()).isTrue(); + + // Verify that attempting to write throws ReadOnlyBufferException + assertThatThrownBy(() -> capturedBuffer.put((byte) 0)) + .isInstanceOf(ReadOnlyBufferException.class); + } } diff --git a/storage/inkless/src/test/java/io/aiven/inkless/produce/FileUploadJobTest.java b/storage/inkless/src/test/java/io/aiven/inkless/produce/FileUploadJobTest.java index 4f9b8567bba..25ad68314c0 100644 --- a/storage/inkless/src/test/java/io/aiven/inkless/produce/FileUploadJobTest.java +++ b/storage/inkless/src/test/java/io/aiven/inkless/produce/FileUploadJobTest.java @@ -29,6 +29,7 @@ import org.mockito.quality.Strictness; import java.io.InputStream; +import java.nio.ByteBuffer; import java.time.Duration; import java.util.HashSet; import java.util.function.Consumer; @@ -184,4 +185,139 @@ void constructorInvalidArguments() { .isInstanceOf(NullPointerException.class) .hasMessage("durationCallback cannot be null"); } + + // ByteBuffer upload tests + + @Test + void byteBufferSuccessAtFirstAttempt() throws Exception { + final ByteBuffer data = ByteBuffer.wrap(new byte[] {1, 2, 3}); + final int originalPosition = data.position(); + + doNothing().when(objectUploader).upload(eq(OBJECT_KEY), any(ByteBuffer.class)); + when(time.nanoseconds()).thenReturn(10_000_000L, 20_000_000L); + + final FileUploadJob fileUploadJob = FileUploadJob.createFromByteBuffer( + OBJECT_KEY_CREATOR, objectUploader, time, 1, Duration.ofMillis(100), data, uploadTimeDurationCallback); + + final ObjectKey objectKey = fileUploadJob.call(); + + assertThat(objectKey).isEqualTo(OBJECT_KEY); + verify(objectUploader).upload(eq(OBJECT_KEY), any(ByteBuffer.class)); + verify(time, never()).sleep(anyLong()); + verify(uploadTimeDurationCallback).accept(eq(10L)); + // Buffer position should not be modified + assertThat(data.position()).isEqualTo(originalPosition); + } + + @Test + void byteBufferSuccessAfterRetry() throws Exception { + final ByteBuffer data = ByteBuffer.wrap(new byte[] {1, 2, 3}); + + doThrow(new StorageBackendException("Test")) + .doThrow(new StorageBackendException("Test")) + .doNothing() + .when(objectUploader).upload(eq(OBJECT_KEY), any(ByteBuffer.class)); + when(time.nanoseconds()).thenReturn(10_000_000L, 20_000_000L); + + final FileUploadJob fileUploadJob = FileUploadJob.createFromByteBuffer( + OBJECT_KEY_CREATOR, objectUploader, time, 3, Duration.ofMillis(100), data, uploadTimeDurationCallback); + final ObjectKey objectKey = fileUploadJob.call(); + + assertThat(objectKey).isEqualTo(OBJECT_KEY); + verify(objectUploader, times(3)).upload(eq(OBJECT_KEY), any(ByteBuffer.class)); + // We don't sleep at the last attempt. + verify(time, times(2)).sleep(eq(100L)); + verify(uploadTimeDurationCallback).accept(eq(10L)); + } + + @Test + void byteBufferUploadStorageFailure() throws Exception { + final ByteBuffer data = ByteBuffer.wrap(new byte[] {1, 2, 3}); + final StorageBackendException exception = new StorageBackendException("Test"); + + doThrow(exception).when(objectUploader).upload(any(), any(ByteBuffer.class)); + when(time.nanoseconds()).thenReturn(10_000_000L, 20_000_000L); + + final FileUploadJob fileUploadJob = FileUploadJob.createFromByteBuffer( + OBJECT_KEY_CREATOR, objectUploader, time, 2, Duration.ofMillis(100), data, uploadTimeDurationCallback); + + assertThatThrownBy(fileUploadJob::call).isSameAs(exception); + verify(objectUploader, times(2)).upload(eq(OBJECT_KEY), any(ByteBuffer.class)); + // We don't sleep at the last attempt. + verify(time, times(1)).sleep(eq(100L)); + verify(uploadTimeDurationCallback).accept(eq(10L)); + } + + @Test + void byteBufferConstructorInvalidArguments() { + final ByteBuffer validBuffer = ByteBuffer.wrap(new byte[] {1, 2, 3}); + + assertThatThrownBy(() -> FileUploadJob.createFromByteBuffer( + null, objectUploader, time, 2, Duration.ofMillis(100), validBuffer, uploadTimeDurationCallback)) + .isInstanceOf(NullPointerException.class) + .hasMessage("objectKeyCreator cannot be null"); + assertThatThrownBy(() -> FileUploadJob.createFromByteBuffer( + OBJECT_KEY_CREATOR, null, time, 2, Duration.ofMillis(100), validBuffer, uploadTimeDurationCallback)) + .isInstanceOf(NullPointerException.class) + .hasMessage("objectUploader cannot be null"); + assertThatThrownBy(() -> FileUploadJob.createFromByteBuffer( + OBJECT_KEY_CREATOR, objectUploader, null, 2, Duration.ofMillis(100), validBuffer, uploadTimeDurationCallback)) + .isInstanceOf(NullPointerException.class) + .hasMessage("time cannot be null"); + assertThatThrownBy(() -> FileUploadJob.createFromByteBuffer( + OBJECT_KEY_CREATOR, objectUploader, time, 0, Duration.ofMillis(100), validBuffer, uploadTimeDurationCallback)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("attempts must be positive"); + assertThatThrownBy(() -> FileUploadJob.createFromByteBuffer( + OBJECT_KEY_CREATOR, objectUploader, time, 2, null, validBuffer, uploadTimeDurationCallback)) + .isInstanceOf(NullPointerException.class) + .hasMessage("retryBackoff cannot be null"); + assertThatThrownBy(() -> FileUploadJob.createFromByteBuffer( + OBJECT_KEY_CREATOR, objectUploader, time, 2, Duration.ofMillis(100), null, uploadTimeDurationCallback)) + .isInstanceOf(NullPointerException.class) + .hasMessage("data cannot be null"); + assertThatThrownBy(() -> FileUploadJob.createFromByteBuffer( + OBJECT_KEY_CREATOR, objectUploader, time, 2, Duration.ofMillis(100), validBuffer, null)) + .isInstanceOf(NullPointerException.class) + .hasMessage("durationCallback cannot be null"); + + // Empty buffer should be rejected + final ByteBuffer emptyBuffer = ByteBuffer.allocate(0); + assertThatThrownBy(() -> FileUploadJob.createFromByteBuffer( + OBJECT_KEY_CREATOR, objectUploader, time, 2, Duration.ofMillis(100), emptyBuffer, uploadTimeDurationCallback)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("data must have remaining bytes"); + + // Buffer with no remaining bytes should be rejected + final ByteBuffer exhaustedBuffer = ByteBuffer.wrap(new byte[] {1, 2, 3}); + exhaustedBuffer.position(exhaustedBuffer.limit()); + assertThatThrownBy(() -> FileUploadJob.createFromByteBuffer( + OBJECT_KEY_CREATOR, objectUploader, time, 2, Duration.ofMillis(100), exhaustedBuffer, uploadTimeDurationCallback)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("data must have remaining bytes"); + } + + @Test + void byteBufferWithOffsetPreservesPosition() throws Exception { + // Create a buffer with offset (simulates a sliced buffer) + final byte[] fullData = new byte[] {0, 0, 0, 1, 2, 3, 0, 0}; + final ByteBuffer data = ByteBuffer.wrap(fullData); + data.position(3); + data.limit(6); // Now buffer represents bytes [1, 2, 3] + + final int originalPosition = data.position(); + final int originalLimit = data.limit(); + + doNothing().when(objectUploader).upload(eq(OBJECT_KEY), any(ByteBuffer.class)); + when(time.nanoseconds()).thenReturn(10_000_000L, 20_000_000L); + + final FileUploadJob fileUploadJob = FileUploadJob.createFromByteBuffer( + OBJECT_KEY_CREATOR, objectUploader, time, 1, Duration.ofMillis(100), data, uploadTimeDurationCallback); + + fileUploadJob.call(); + + // Buffer position and limit should not be modified + assertThat(data.position()).isEqualTo(originalPosition); + assertThat(data.limit()).isEqualTo(originalLimit); + } } diff --git a/storage/inkless/src/test/java/io/aiven/inkless/produce/WriterPropertyTest.java b/storage/inkless/src/test/java/io/aiven/inkless/produce/WriterPropertyTest.java index e273fbbad8d..9c6575f485a 100644 --- a/storage/inkless/src/test/java/io/aiven/inkless/produce/WriterPropertyTest.java +++ b/storage/inkless/src/test/java/io/aiven/inkless/produce/WriterPropertyTest.java @@ -47,7 +47,7 @@ import org.mockito.invocation.Invocation; import org.testcontainers.junit.jupiter.Container; -import java.io.InputStream; +import java.nio.ByteBuffer; import java.time.Duration; import java.time.Instant; import java.util.ArrayList; @@ -83,7 +83,6 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.atLeast; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockingDetails; @@ -290,13 +289,16 @@ void test(final int requestCount, requester.checkResponses(); if (requestCount > 0) { - verify(storage, atLeast(1)).upload(any(ObjectKey.class), any(InputStream.class), anyLong()); + verify(storage, atLeast(1)).upload(any(ObjectKey.class), any(ByteBuffer.class)); } - final Collection uploadInvocations = mockingDetails(storage).getInvocations(); + // Filter to only upload(ObjectKey, ByteBuffer) invocations to avoid counting other mock interactions + final List uploadInvocations = mockingDetails(storage).getInvocations().stream() + .filter(inv -> inv.getMethod().getName().equals("upload") && inv.getArguments().length == 2) + .toList(); Statistics.label("files").collect(uploadInvocations.size()); for (final Invocation invocation : uploadInvocations) { - final long uploadedBytesLength = invocation.getArgument(2); - Statistics.label("file-size").collect(uploadedBytesLength); + final ByteBuffer uploadedBuffer = invocation.getArgument(1); + Statistics.label("file-size").collect(uploadedBuffer.remaining()); } } } diff --git a/storage/inkless/src/test/java/io/aiven/inkless/storage_backend/common/fixtures/BaseStorageTest.java b/storage/inkless/src/test/java/io/aiven/inkless/storage_backend/common/fixtures/BaseStorageTest.java index 71f03761686..0b2ba2b8dd8 100644 --- a/storage/inkless/src/test/java/io/aiven/inkless/storage_backend/common/fixtures/BaseStorageTest.java +++ b/storage/inkless/src/test/java/io/aiven/inkless/storage_backend/common/fixtures/BaseStorageTest.java @@ -139,6 +139,46 @@ void testRetryUploadKeepLatestVersion() throws Exception { assertThat(fetch.array()).isEqualTo(content2); } + @Test + void testUploadFromByteBuffer() throws Exception { + try (StorageBackend storage = storage()) { + final byte[] content = "ByteBuffer upload test content".getBytes(); + final ByteBuffer buffer = ByteBuffer.wrap(content); + + storage.upload(TOPIC_PARTITION_SEGMENT_KEY, buffer); + + // Verify buffer position was not modified (uses duplicate internally) + assertThat(buffer.position()).isEqualTo(0); + + // Verify data was uploaded correctly + final ByteBuffer fetch = storage.readToByteBuffer( + storage.fetch(TOPIC_PARTITION_SEGMENT_KEY, new ByteRange(0, content.length))); + assertThat(fetch.array()).isEqualTo(content); + } + } + + @Test + void testUploadFromByteBufferWithOffset() throws Exception { + try (StorageBackend storage = storage()) { + // Create a buffer with data starting at an offset to test sliced buffer upload + // Full string: "PREFIX_actual content to upload_SUFFIX" (38 bytes) + // We want to upload only: "actual content to upload" (24 bytes, positions 7-30) + final byte[] fullContent = "PREFIX_actual content to upload_SUFFIX".getBytes(); + final ByteBuffer buffer = ByteBuffer.wrap(fullContent); + buffer.position(7); // Skip "PREFIX_" (7 bytes) + buffer.limit(31); // End before "_SUFFIX" (positions 31-37) + + final byte[] expectedContent = "actual content to upload".getBytes(); + + storage.upload(TOPIC_PARTITION_SEGMENT_KEY, buffer); + + // Verify data was uploaded correctly (only the slice) + final ByteBuffer fetch = storage.readToByteBuffer( + storage.fetch(TOPIC_PARTITION_SEGMENT_KEY, new ByteRange(0, expectedContent.length))); + assertThat(fetch.array()).isEqualTo(expectedContent); + } + } + @Test void testFetchFailWhenNonExistingKey() throws Exception { try (StorageBackend storage = storage()) { @@ -299,4 +339,32 @@ protected void testDeletes() throws Exception { } } } + + @Test + void testFetchToByteBuffer() throws Exception { + try (StorageBackend storage = storage()) { + final byte[] data = "fetchToByteBuffer test content".getBytes(); + storage.upload(TOPIC_PARTITION_SEGMENT_KEY, new ByteArrayInputStream(data), data.length); + + // Fetch entire content + ByteBuffer result = storage.fetchToByteBuffer(TOPIC_PARTITION_SEGMENT_KEY, new ByteRange(0, data.length)); + byte[] resultBytes = new byte[result.remaining()]; + result.get(resultBytes); + assertThat(resultBytes).isEqualTo(data); + } + } + + @Test + void testFetchToByteBufferWithRange() throws Exception { + try (StorageBackend storage = storage()) { + final byte[] data = "AABBBBAA".getBytes(); + storage.upload(TOPIC_PARTITION_SEGMENT_KEY, new ByteArrayInputStream(data), data.length); + + // Fetch partial content + ByteBuffer result = storage.fetchToByteBuffer(TOPIC_PARTITION_SEGMENT_KEY, new ByteRange(2, 4)); + byte[] resultBytes = new byte[result.remaining()]; + result.get(resultBytes); + assertThat(new String(resultBytes)).isEqualTo("BBBB"); + } + } }