From fd844e1161a6ca4b877351e656521180077bfe5f Mon Sep 17 00:00:00 2001 From: Fuat Basik Date: Fri, 1 Aug 2025 13:24:44 +0100 Subject: [PATCH 1/8] Add retries to new PhysicalIO --- .../util/retry/DefaultRetryStrategyImpl.java | 53 ++- .../util/retry/IOSupplier.java | 2 +- .../util/retry/RetryStrategy.java | 18 +- .../retry/DefaultRetryStrategyImplTest.java | 58 ++- .../io/physical/data/Block.java | 81 +--- .../io/physical/data/BlockManager.java | 9 +- .../io/physical/impl/PhysicalIOImpl.java | 7 - .../io/physical/reader/StreamReader.java | 113 +++--- .../io/physical/data/BlockStoreTest.java | 135 +------ .../io/physical/data/BlockTest.java | 345 +++--------------- 10 files changed, 217 insertions(+), 604 deletions(-) diff --git a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java index 40ecd72b..b36211a8 100644 --- a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java +++ b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java @@ -18,24 +18,29 @@ import dev.failsafe.Failsafe; import dev.failsafe.FailsafeException; import dev.failsafe.FailsafeExecutor; -import java.io.IOException; +import dev.failsafe.Timeout; +import dev.failsafe.TimeoutExceededException; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Optional; import java.util.stream.Collectors; +import lombok.SneakyThrows; import software.amazon.s3.analyticsaccelerator.common.Preconditions; /** * Retry strategy implementation for seekable input stream operations. Uses Failsafe library to * execute operations with configurable retry policies. * - *

This strategy will be additive to readTimeout and readRetryCount set on PhysicalIO - * configuration. + *

If provided with a timeout this strategy will overwrite readTimeout and readRetryCount set on + * PhysicalIOConfiguration. If not, values from PhysicalIOConfiguration will be used to manage + * storage read timeouts. */ public class DefaultRetryStrategyImpl implements RetryStrategy { private final List retryPolicies; FailsafeExecutor failsafeExecutor; + private boolean timeoutSet; /** Creates a retry strategy with no retry policies (no retries). */ public DefaultRetryStrategyImpl() { @@ -76,10 +81,10 @@ public DefaultRetryStrategyImpl(List policies) { * Executes a runnable operation with retry logic. * * @param runnable the operation to execute - * @throws IOException if the operation fails after all retries */ @Override - public void execute(IORunnable runnable) throws IOException { + @SneakyThrows + public void execute(IORunnable runnable) { try { this.failsafeExecutor.run(runnable::apply); } catch (Exception ex) { @@ -93,10 +98,10 @@ public void execute(IORunnable runnable) throws IOException { * @param return type of the supplier * @param supplier the operation that returns a byte array * @return the result of the supplier operation - * @throws IOException if the operation fails after all retries */ @Override - public T get(IOSupplier supplier) throws IOException { + @SneakyThrows + public T get(IOSupplier supplier) { try { return this.failsafeExecutor.get(supplier::apply); } catch (Exception ex) { @@ -135,24 +140,38 @@ private List> getDelegates() { } /** - * Handles exceptions after retry attempts are exhausted. + * Handles exceptions after retry attempts are exhausted. This is needed to unwrap Failsafe + * exception * * @param e the exception that occurred * @return an IOException to throw */ - private IOException handleExceptionAfterRetry(Exception e) { - IOException toThrow = new IOException("Failed to execute operation with retries", e); - + private Exception handleExceptionAfterRetry(Exception e) { if (e instanceof FailsafeException) { Optional cause = Optional.ofNullable(e.getCause()); if (cause.isPresent()) { - if (cause.get() instanceof IOException) { - return (IOException) cause.get(); - } else { - toThrow = new IOException("Failed to execute operation with retries", cause.get()); - } + return (Exception) cause.get(); } } - return toThrow; + return e; + } + + /** + * @inheritDoc + * @param timeoutDurationMillis Timeout duration for reading from storage + * @param retryCount Number of times to retry if Timeout Exceeds + */ + public void timeout(long timeoutDurationMillis, int retryCount) { + if (!this.timeoutSet) { + Timeout timeout = + Timeout.builder(Duration.ofMillis(timeoutDurationMillis)).withInterrupt().build(); + dev.failsafe.RetryPolicy timeoutPolicy = + dev.failsafe.RetryPolicy.builder() + .handle(TimeoutExceededException.class) + .withMaxRetries(retryCount) + .build(); + this.failsafeExecutor = this.failsafeExecutor.compose(timeoutPolicy).compose(timeout); + this.timeoutSet = true; + } } } diff --git a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/IOSupplier.java b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/IOSupplier.java index c3b758c2..6ca33f2a 100644 --- a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/IOSupplier.java +++ b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/IOSupplier.java @@ -31,5 +31,5 @@ public interface IOSupplier { * @return a value of type {@link T}. * @throws IOException on error condition. */ - T apply() throws IOException; + T apply() throws IOException, InterruptedException; } diff --git a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.java b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.java index b61d9288..1fa0867f 100644 --- a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.java +++ b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.java @@ -15,7 +15,6 @@ */ package software.amazon.s3.analyticsaccelerator.util.retry; -import java.io.IOException; import java.util.List; /** Interface for executing operations with retry logic. */ @@ -24,9 +23,8 @@ public interface RetryStrategy { * Executes a runnable with retry logic. * * @param runnable the operation to execute - * @throws IOException if the operation fails after all retry attempts */ - void execute(IORunnable runnable) throws IOException; + void execute(IORunnable runnable); /** * Executes a supplier with retry logic. @@ -34,9 +32,8 @@ public interface RetryStrategy { * @param return type of the supplier * @param supplier the operation to execute * @return result of the supplier - * @throws IOException if the operation fails after all retry attempts */ - T get(IOSupplier supplier) throws IOException; + T get(IOSupplier supplier); /** * Adds a retry policy to the strategy. This will be policy first to execute as it is appended to @@ -62,4 +59,15 @@ public interface RetryStrategy { * @return list of {@link RetryPolicy} */ List getRetryPolicies(); + + /** + * Create a timeout for read from storage operations and with specified retry count. This will + * override settings in PhysicalIOConfiguration (blockreadtimeout and blockreadretrycount) if set. + * If user does not set a timeout in their retry strategy, a timeout will be set based on + * aforementioned configuration. set blockreadtimeout = 0 to disable timeouts. + * + * @param durationInMillis Timeout duration for reading from storage + * @param retryCount Number of times to retry if Timeout Exceeds + */ + void timeout(long durationInMillis, int retryCount); } diff --git a/common/src/test/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImplTest.java b/common/src/test/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImplTest.java index 0b7c3242..8b3e496f 100644 --- a/common/src/test/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImplTest.java +++ b/common/src/test/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImplTest.java @@ -181,20 +181,19 @@ void testExecuteSuccess() throws IOException { } @Test - void testExecuteWrapsUncheckedException() { + void testExecuteDoesNotWrapUncheckedException() { DefaultRetryStrategyImpl executor = new DefaultRetryStrategyImpl(); - IOException exception = + RuntimeException exception = assertThrows( - IOException.class, + RuntimeException.class, () -> executor.execute( () -> { throw new RuntimeException("Test exception"); })); - assertEquals("Failed to execute operation with retries", exception.getMessage()); - assertNotNull(exception.getCause()); + assertEquals("Test exception", exception.getMessage()); } @Test @@ -225,20 +224,19 @@ void testGetSuccess() throws IOException { } @Test - void testGetWrapsException() { + void testGetDoesNotWrapException() { DefaultRetryStrategyImpl executor = new DefaultRetryStrategyImpl(); - IOException exception = + RuntimeException exception = assertThrows( - IOException.class, + RuntimeException.class, () -> executor.get( () -> { throw new RuntimeException("Test exception"); })); - assertEquals("Failed to execute operation with retries", exception.getMessage()); - assertNotNull(exception.getCause()); + assertEquals("Test exception", exception.getMessage()); } @Test @@ -288,6 +286,46 @@ void testOnRetryCallback() throws IOException { assertEquals(2, retryCounter.get()); } + @Test + void testTimeoutThrows() { + DefaultRetryStrategyImpl executor = new DefaultRetryStrategyImpl(); + executor.timeout(1000, 0); + AtomicInteger attempt = new AtomicInteger(0); + + assertThrows( + RuntimeException.class, + () -> + executor.get( + () -> { + attempt.incrementAndGet(); + Thread.sleep(10000); + return null; + })); + + assertEquals(1, attempt.get()); // 1 initial + } + + @Test + void testTimeoutWithSuccessAfterRetry() throws IOException { + DefaultRetryStrategyImpl executor = new DefaultRetryStrategyImpl(); + executor.timeout(1000, 3); + AtomicInteger attempt = new AtomicInteger(0); + String expected = "success"; + + byte[] result = + executor.get( + () -> { + int currentAttempt = attempt.incrementAndGet(); + if (currentAttempt <= 2) { + Thread.sleep(10000); + } + return expected.getBytes(StandardCharsets.UTF_8); + }); + + assertEquals(expected.getBytes(StandardCharsets.UTF_8).length, result.length); + assertEquals(3, attempt.get()); + } + private byte[] failXTimesThenSucceed(AtomicInteger counter, int failCount, String toByteArray) throws IOException { int attempt = counter.incrementAndGet(); diff --git a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/Block.java b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/Block.java index 8a88ebc8..d1a90af0 100644 --- a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/Block.java +++ b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/Block.java @@ -18,8 +18,6 @@ import java.io.Closeable; import java.io.IOException; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import javax.annotation.Nullable; import lombok.Getter; import lombok.NonNull; @@ -27,10 +25,6 @@ import software.amazon.s3.analyticsaccelerator.common.Preconditions; import software.amazon.s3.analyticsaccelerator.util.BlockKey; import software.amazon.s3.analyticsaccelerator.util.MetricKey; -import software.amazon.s3.analyticsaccelerator.util.OpenStreamInformation; -import software.amazon.s3.analyticsaccelerator.util.retry.DefaultRetryStrategyImpl; -import software.amazon.s3.analyticsaccelerator.util.retry.RetryPolicy; -import software.amazon.s3.analyticsaccelerator.util.retry.RetryStrategy; /** * Represents a block of data from an object stream, identified by a {@link BlockKey} and a @@ -55,10 +49,6 @@ public class Block implements Closeable { private final BlobStoreIndexCache indexCache; private final Metrics aggregatingMetrics; - private final long readTimeout; - private final int retryCount; - private final RetryStrategy retryStrategy; - private final OpenStreamInformation openStreamInformation; /** * A synchronization aid that allows threads to wait until the block's data is available. @@ -85,18 +75,12 @@ public class Block implements Closeable { * @param generation the generation number of this block in a sequential read pattern * @param indexCache blobstore index cache * @param aggregatingMetrics blobstore metrics - * @param readTimeout read timeout in milliseconds - * @param retryCount number of retries - * @param openStreamInformation contains stream information */ public Block( @NonNull BlockKey blockKey, long generation, @NonNull BlobStoreIndexCache indexCache, - @NonNull Metrics aggregatingMetrics, - long readTimeout, - int retryCount, - @NonNull OpenStreamInformation openStreamInformation) { + @NonNull Metrics aggregatingMetrics) { Preconditions.checkArgument( 0 <= generation, "`generation` must be non-negative; was: %s", generation); @@ -104,36 +88,6 @@ public Block( this.generation = generation; this.indexCache = indexCache; this.aggregatingMetrics = aggregatingMetrics; - this.readTimeout = readTimeout; - this.retryCount = retryCount; - this.openStreamInformation = openStreamInformation; - this.retryStrategy = createRetryStrategy(); - } - - /** - * Helper to construct retryStrategy - * - * @return a {@link RetryStrategy} to retry when timeouts are set - * @throws RuntimeException if all retries fails and an error occurs - */ - @SuppressWarnings("unchecked") - private RetryStrategy createRetryStrategy() { - RetryStrategy base = new DefaultRetryStrategyImpl(); - RetryStrategy provided = this.openStreamInformation.getRetryStrategy(); - - if (provided != null) { - base = base.merge(provided); - } - - if (this.readTimeout > 0) { - RetryPolicy timeoutPolicy = - RetryPolicy.builder() - .handle(InterruptedException.class, TimeoutException.class, IOException.class) - .withMaxRetries(this.retryCount) - .build(); - base = base.amend(timeoutPolicy); - } - return base; } /** @@ -145,8 +99,7 @@ private RetryStrategy createRetryStrategy() { */ public int read(long pos) throws IOException { Preconditions.checkArgument(0 <= pos, "`pos` must not be negative"); - awaitDataWithRetry(); - + awaitData(); indexCache.recordAccess(this.blockKey); int contentOffset = posToOffset(pos); return Byte.toUnsignedInt(this.data[contentOffset]); @@ -169,7 +122,7 @@ public int read(byte @NonNull [] buf, int off, int len, long pos) throws IOExcep Preconditions.checkArgument(0 <= len, "`len` must not be negative"); Preconditions.checkArgument(off < buf.length, "`off` must be less than size of buffer"); - awaitDataWithRetry(); + awaitData(); indexCache.recordAccess(this.blockKey); int contentOffset = posToOffset(pos); @@ -223,21 +176,6 @@ public void setError(@NonNull IOException error) { dataReadyLatch.countDown(); } - private void awaitDataWithRetry() throws IOException { - this.retryStrategy.get( - () -> { - awaitData(); - return null; - }); - - if (this.error != null) { - throw error; - } - if (this.data == null) { - throw new IOException("Error while reading data. Block data is null after successful await"); - } - } - /** * Waits for the block's data to become available. This method blocks until {@link * #setData(byte[])} is called. @@ -246,15 +184,16 @@ private void awaitDataWithRetry() throws IOException { */ private void awaitData() throws IOException { try { - if (!dataReadyLatch.await(readTimeout, TimeUnit.MILLISECONDS)) { - throw new IOException( - "Error while reading data. Request timed out after " - + readTimeout - + "ms while waiting for block data"); - } + dataReadyLatch.await(); } catch (InterruptedException e) { throw new IOException("Error while reading data. Read interrupted while waiting for data", e); } + if (this.error != null) { + throw error; + } + if (this.data == null) { + throw new IOException("Error while reading data. Block data is null after successful await"); + } } /** diff --git a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockManager.java b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockManager.java index d071d95a..7091814c 100644 --- a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockManager.java +++ b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockManager.java @@ -211,14 +211,7 @@ public synchronized void makeRangeAvailable(long pos, long len, ReadMode readMod for (int blockIndex : group) { BlockKey blockKey = new BlockKey(objectKey, getBlockIndexRange(blockIndex)); Block block = - new Block( - blockKey, - generation, - this.indexCache, - this.aggregatingMetrics, - this.configuration.getBlockReadTimeout(), - this.configuration.getBlockReadRetryCount(), - this.openStreamInformation); + new Block(blockKey, generation, this.indexCache, this.aggregatingMetrics); // Add block to the store for future reference blockStore.add(block); blocksToFill.add(block); diff --git a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/impl/PhysicalIOImpl.java b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/impl/PhysicalIOImpl.java index bef3d9ac..bb332184 100644 --- a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/impl/PhysicalIOImpl.java +++ b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/impl/PhysicalIOImpl.java @@ -327,13 +327,6 @@ private void makeReadVectoredRangesAvailable(List objectRanges) { private void handleOperationExceptions(Exception e) { boolean shouldEvict = false; - // Check for IO errors while reading data - if (e instanceof IOException - && e.getMessage() != null - && e.getMessage().contains("Error while reading data.")) { - shouldEvict = true; - } - // Check for precondition failed errors (412) if (e.getCause() != null && e.getCause().getMessage() != null diff --git a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java index 6978ab0c..454b0145 100644 --- a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java +++ b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java @@ -21,9 +21,7 @@ import java.io.IOException; import java.io.InputStream; import java.util.List; -import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; -import java.util.concurrent.TimeoutException; import java.util.function.Consumer; import java.util.stream.Collectors; import lombok.NonNull; @@ -46,7 +44,6 @@ import software.amazon.s3.analyticsaccelerator.util.OpenStreamInformation; import software.amazon.s3.analyticsaccelerator.util.StreamAttributes; import software.amazon.s3.analyticsaccelerator.util.retry.DefaultRetryStrategyImpl; -import software.amazon.s3.analyticsaccelerator.util.retry.RetryPolicy; import software.amazon.s3.analyticsaccelerator.util.retry.RetryStrategy; /** @@ -112,28 +109,19 @@ public StreamReader( * @return a {@link RetryStrategy} to retry when timeouts are set * @throws RuntimeException if all retries fails and an error occurs */ - @SuppressWarnings("unchecked") private RetryStrategy createRetryStrategy() { - RetryStrategy base = new DefaultRetryStrategyImpl(); RetryStrategy provided = this.openStreamInformation.getRetryStrategy(); - if (provided != null) { - base = base.merge(provided); + if (provided == null) { + provided = new DefaultRetryStrategyImpl(); } if (this.physicalIOConfiguration.getBlockReadTimeout() > 0) { - RetryPolicy timeoutPolicy = - RetryPolicy.builder() - .handle( - IOException.class, - InterruptedException.class, - TimeoutException.class, - ExecutionException.class) - .withMaxRetries(this.physicalIOConfiguration.getBlockReadRetryCount()) - .build(); - base = base.amend(timeoutPolicy); + provided.timeout( + physicalIOConfiguration.getBlockReadTimeout(), + physicalIOConfiguration.getBlockReadRetryCount()); } - return base; + return provided; } /** @@ -178,55 +166,56 @@ private Runnable processReadTask(final List blocks, ReadMode readMode) { blocks.get(blocks.size() - 1).getBlockKey().getRange().getEnd())) .build(), () -> { - // Calculate the byte range needed to cover all blocks - Range requestRange = computeRange(blocks); + try { + retryStrategy.execute( + () -> { + // Calculate the byte range needed to cover all blocks + List nonFilledBlocks = + blocks.stream() + .filter(block -> !block.isDataReady()) + .collect(Collectors.toList()); - // Build S3 GET request with range, ETag validation, and referrer info - GetRequest getRequest = - GetRequest.builder() - .s3Uri(objectKey.getS3URI()) - .range(requestRange) - .etag(objectKey.getEtag()) - .referrer(new Referrer(requestRange.toHttpString(), readMode)) - .build(); + Range requestRange = computeRange(nonFilledBlocks); - // Fetch the object content from S3 - ObjectContent objectContent; - try { - objectContent = this.retryStrategy.get(() -> fetchObjectContent(getRequest)); - } catch (IOException e) { - LOG.error("IOException while fetching object content", e); - setErrorOnBlocksAndRemove(blocks, e); - return; - } + // Build S3 GET request with range, ETag validation, and referrer info + GetRequest getRequest = + GetRequest.builder() + .s3Uri(objectKey.getS3URI()) + .range(requestRange) + .etag(objectKey.getEtag()) + .referrer(new Referrer(requestRange.toHttpString(), readMode)) + .build(); - openStreamInformation.getRequestCallback().onGetRequest(); + // Fetch the object content from S3 + ObjectContent objectContent; + objectContent = fetchObjectContent(getRequest); - if (objectContent == null) { - // Couldn't successfully get the response from S3. - // Remove blocks from store and complete async operation - removeNonFilledBlocksFromStore(blocks); - return; - } + openStreamInformation.getRequestCallback().onGetRequest(); - // Process the input stream and populate data blocks - try (InputStream inputStream = objectContent.getStream()) { - boolean success = - readBlocksFromStream(inputStream, blocks, requestRange.getStart()); - if (!success) { - removeNonFilledBlocksFromStore(blocks); - } - } catch (EOFException e) { - LOG.error("EOFException while reading blocks", e); - setErrorOnBlocksAndRemove(blocks, e); - } catch (IOException e) { - LOG.error("IOException while reading blocks", e); - setErrorOnBlocksAndRemove(blocks, e); - } catch (Exception e) { + if (objectContent == null) { + // Couldn't successfully get the response from S3. + // Remove blocks from store and complete async operation + removeNonFilledBlocksFromStore(nonFilledBlocks); + return; + } + InputStream inputStream = objectContent.getStream(); + boolean success = + readBlocksFromStream( + inputStream, nonFilledBlocks, requestRange.getStart()); + if (!success) { + removeNonFilledBlocksFromStore(nonFilledBlocks); + } + }); + } // Process the input stream and populate data blocks + catch (Exception e) { LOG.error("Unexpected exception while reading blocks", e); - IOException ioException = - new IOException("Unexpected error during block reading", e); - setErrorOnBlocksAndRemove(blocks, ioException); + if (e instanceof IOException) { + setErrorOnBlocksAndRemove(blocks, (IOException) e); + } else { + IOException ioException = + new IOException("Unexpected error during block reading", e); + setErrorOnBlocksAndRemove(blocks, ioException); + } } }); } @@ -287,7 +276,7 @@ private ObjectContent fetchObjectContent(GetRequest getRequest) throws IOExcepti .attribute(StreamAttributes.range(getRequest.getRange())) .build(), this.objectClient.getObject(getRequest, this.openStreamInformation), - this.physicalIOConfiguration.getBlockReadTimeout()); + 100000000); } /** diff --git a/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockStoreTest.java b/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockStoreTest.java index 805beffd..af1f61f3 100644 --- a/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockStoreTest.java +++ b/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockStoreTest.java @@ -40,9 +40,6 @@ public class BlockStoreTest { private static final S3URI TEST_URI = S3URI.of("foo", "bar"); private static final String ETAG = "RANDOM"; private static final ObjectKey objectKey = ObjectKey.builder().s3URI(TEST_URI).etag(ETAG).build(); - private static final long DEFAULT_READ_TIMEOUT = 120_000; - private static final int DEFAULT_RETRY_COUNT = 20; - private BlobStoreIndexCache mockIndexCache; private Metrics mockMetrics; private PhysicalIOConfiguration configuration; @@ -87,15 +84,7 @@ public void test__blockStore__getBlockAfterAddBlock() { BlockKey blockKey = new BlockKey(objectKey, new Range(3, 5)); // When: a new block is added - blockStore.add( - new Block( - blockKey, - 0, - mock(BlobStoreIndexCache.class), - mock(Metrics.class), - DEFAULT_READ_TIMEOUT, - DEFAULT_RETRY_COUNT, - OpenStreamInformation.DEFAULT)); + blockStore.add(new Block(blockKey, 0, mock(BlobStoreIndexCache.class), mock(Metrics.class))); // Then: getBlock can retrieve the same block Optional b = blockStore.getBlock(4); @@ -149,15 +138,7 @@ public void test__blockStore__getBlockByIndex() { // Given: BlockStore with a block at a specific index BlockKey blockKey = new BlockKey(objectKey, new Range(8192, 16383)); // Assuming readBufferSize is 8KB - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - DEFAULT_READ_TIMEOUT, - DEFAULT_RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); blockStore.add(block); // When: getBlockByIndex is called with the correct index @@ -193,24 +174,8 @@ public void test__blockStore__getBlock_negativePosition() { public void test__blockStore__add_duplicateBlock() { // Given: A block already in the store BlockKey blockKey = new BlockKey(objectKey, new Range(0, 8191)); - Block block1 = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - DEFAULT_READ_TIMEOUT, - DEFAULT_RETRY_COUNT, - OpenStreamInformation.DEFAULT); - Block block2 = - new Block( - blockKey, - 1, - mockIndexCache, - mockMetrics, - DEFAULT_READ_TIMEOUT, - DEFAULT_RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block1 = new Block(blockKey, 0, mockIndexCache, mockMetrics); + Block block2 = new Block(blockKey, 1, mockIndexCache, mockMetrics); // When: The first block is added blockStore.add(block1); @@ -231,16 +196,7 @@ public void test__blockStore__add_duplicateBlock() { public void test__blockStore__remove() throws IOException { // Given: A block in the store BlockKey blockKey = new BlockKey(objectKey, new Range(0, 4)); - Block block = - spy( - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - DEFAULT_READ_TIMEOUT, - DEFAULT_RETRY_COUNT, - OpenStreamInformation.DEFAULT)); + Block block = spy(new Block(blockKey, 0, mockIndexCache, mockMetrics)); block.setData(new byte[] {1, 2, 3, 4, 5}); when(block.isDataReady()).thenReturn(true); blockStore.add(block); @@ -264,15 +220,7 @@ public void test__blockStore__remove() throws IOException { public void test__blockStore__remove_nonExistentBlock() { // Given: A block not in the store BlockKey blockKey = new BlockKey(objectKey, new Range(0, 8191)); - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - DEFAULT_READ_TIMEOUT, - DEFAULT_RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); // When: An attempt is made to remove the block blockStore.remove(block); @@ -285,16 +233,7 @@ public void test__blockStore__remove_nonExistentBlock() { public void test__blockStore__remove_dataNotReady() throws IOException { // Given: A block in the store with data not ready BlockKey blockKey = new BlockKey(objectKey, new Range(0, 4)); - Block block = - spy( - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - DEFAULT_READ_TIMEOUT, - DEFAULT_RETRY_COUNT, - OpenStreamInformation.DEFAULT)); + Block block = spy(new Block(blockKey, 0, mockIndexCache, mockMetrics)); when(block.isDataReady()).thenReturn(false); blockStore.add(block); @@ -318,24 +257,8 @@ public void test__blockStore__getMissingBlockIndexesInRange() { BlockKey blockKey1 = new BlockKey(objectKey, new Range(0, 8191)); // Index 0 BlockKey blockKey2 = new BlockKey(objectKey, new Range(16384, 24575)); // Index 2 - Block block1 = - new Block( - blockKey1, - 0, - mockIndexCache, - mockMetrics, - DEFAULT_READ_TIMEOUT, - DEFAULT_RETRY_COUNT, - OpenStreamInformation.DEFAULT); - Block block2 = - new Block( - blockKey2, - 0, - mockIndexCache, - mockMetrics, - DEFAULT_READ_TIMEOUT, - DEFAULT_RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block1 = new Block(blockKey1, 0, mockIndexCache, mockMetrics); + Block block2 = new Block(blockKey2, 0, mockIndexCache, mockMetrics); blockStore.add(block1); blockStore.add(block2); @@ -393,15 +316,7 @@ public void test__blockStore__isEmpty() { // When: A block is added BlockKey blockKey = new BlockKey(objectKey, new Range(0, 8191)); - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - DEFAULT_READ_TIMEOUT, - DEFAULT_RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); blockStore.add(block); // Then: isEmpty returns false @@ -426,15 +341,7 @@ public void test__blockStore__concurrentAddRemove() throws InterruptedException () -> { BlockKey blockKey = new BlockKey(objectKey, new Range(index * 8192L, (index + 1) * 8192L - 1)); - Block block = - new Block( - blockKey, - index, - mockIndexCache, - mockMetrics, - DEFAULT_READ_TIMEOUT, - DEFAULT_RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, index, mockIndexCache, mockMetrics); blockStore.add(block); blockStore.remove(block); latch.countDown(); @@ -451,15 +358,7 @@ public void test__blockStore__concurrentAddRemove() throws InterruptedException @Test public void test__blockStore__getBlock_atRangeBoundaries() { BlockKey blockKey = new BlockKey(objectKey, new Range(0, 8191)); - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - DEFAULT_READ_TIMEOUT, - DEFAULT_RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); blockStore.add(block); // At start of range @@ -489,15 +388,7 @@ public void test__blockStore__remove_nullBlock() { @Test public void test__blockStore__getBlock_positionOutsideAnyBlock() { BlockKey blockKey = new BlockKey(objectKey, new Range(0, 8191)); - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - DEFAULT_READ_TIMEOUT, - DEFAULT_RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); blockStore.add(block); Optional outsideBlock = blockStore.getBlock(100_000); diff --git a/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockTest.java b/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockTest.java index 0e02bf79..c9d27aee 100644 --- a/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockTest.java +++ b/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockTest.java @@ -26,6 +26,7 @@ import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import software.amazon.s3.analyticsaccelerator.common.Metrics; import software.amazon.s3.analyticsaccelerator.request.Range; @@ -39,8 +40,6 @@ public class BlockTest { private static final String ETAG = "RandomString"; private static final String TEST_DATA = "test-data"; private static final byte[] TEST_DATA_BYTES = TEST_DATA.getBytes(StandardCharsets.UTF_8); - private static final long READ_TIMEOUT = 5_000; - private static final int RETRY_COUNT = 2; private ObjectKey objectKey; private BlockKey blockKey; @@ -57,15 +56,7 @@ void setUp() { @Test void testConstructorWithValidParameters() { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); assertNotNull(block); assertEquals(blockKey, block.getBlockKey()); @@ -75,62 +66,23 @@ void testConstructorWithValidParameters() { @Test void testConstructorWithNullBlockKey() { - assertThrows( - NullPointerException.class, - () -> - new Block( - null, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT)); + assertThrows(NullPointerException.class, () -> new Block(null, 0, mockIndexCache, mockMetrics)); } @Test void testConstructorWithNullIndexCache() { - assertThrows( - NullPointerException.class, - () -> - new Block( - blockKey, - 0, - null, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT)); + assertThrows(NullPointerException.class, () -> new Block(blockKey, 0, null, mockMetrics)); } @Test void testConstructorWithNullMetrics() { - assertThrows( - NullPointerException.class, - () -> - new Block( - blockKey, - 0, - mockIndexCache, - null, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT)); + assertThrows(NullPointerException.class, () -> new Block(blockKey, 0, mockIndexCache, null)); } @Test void testConstructorWithNegativeGeneration() { assertThrows( - IllegalArgumentException.class, - () -> - new Block( - blockKey, - -1, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT)); + IllegalArgumentException.class, () -> new Block(blockKey, -1, mockIndexCache, mockMetrics)); } @Test @@ -139,14 +91,7 @@ void testConstructorWithNegativeRangeStart() { IllegalArgumentException.class, () -> { BlockKey invalidBlockKey = new BlockKey(objectKey, new Range(-1, TEST_DATA.length())); - new Block( - invalidBlockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + new Block(invalidBlockKey, 0, mockIndexCache, mockMetrics); }); } @@ -156,28 +101,13 @@ void testConstructorWithNegativeRangeEnd() { IllegalArgumentException.class, () -> { BlockKey blockKey = new BlockKey(objectKey, new Range(0, -1)); - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + new Block(blockKey, 0, mockIndexCache, mockMetrics); }); } @Test void testSetDataAndIsDataReady() { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); assertFalse(block.isDataReady()); @@ -190,15 +120,7 @@ void testSetDataAndIsDataReady() { @Test void testReadSingleByteAfterDataSet() throws IOException { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); block.setData(TEST_DATA_BYTES); int result = block.read(0); @@ -209,15 +131,7 @@ void testReadSingleByteAfterDataSet() throws IOException { @Test void testReadSingleByteAtDifferentPositions() throws IOException { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); block.setData(TEST_DATA_BYTES); assertEquals(116, block.read(0)); // 't' @@ -233,15 +147,7 @@ void testReadSingleByteAtDifferentPositions() throws IOException { @Test void testReadSingleByteWithNegativePosition() { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); block.setData(TEST_DATA_BYTES); assertThrows(IllegalArgumentException.class, () -> block.read(-1)); @@ -249,15 +155,7 @@ void testReadSingleByteWithNegativePosition() { @Test void testReadBufferAfterDataSet() throws IOException { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); block.setData(TEST_DATA_BYTES); byte[] buffer = new byte[4]; @@ -270,15 +168,7 @@ void testReadBufferAfterDataSet() throws IOException { @Test void testReadBufferAtDifferentPositions() throws IOException { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); block.setData(TEST_DATA_BYTES); byte[] buffer1 = new byte[4]; @@ -294,15 +184,7 @@ void testReadBufferAtDifferentPositions() throws IOException { @Test void testReadBufferPartialRead() throws IOException { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); block.setData(TEST_DATA_BYTES); byte[] buffer = new byte[10]; @@ -314,15 +196,7 @@ void testReadBufferPartialRead() throws IOException { @Test void testReadBufferWithInvalidParameters() { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); block.setData(TEST_DATA_BYTES); byte[] buffer = new byte[4]; @@ -334,47 +208,26 @@ void testReadBufferWithInvalidParameters() { } @Test + @Disabled void testReadBeforeDataSet() { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - 100, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); // Short timeout + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); // Short timeout assertThrows(IOException.class, () -> block.read(0)); } @Test + @Disabled void testReadBufferBeforeDataSet() { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - 100, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); // Short timeout + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); // Short timeout byte[] buffer = new byte[4]; assertThrows(IOException.class, () -> block.read(buffer, 0, 4, 0)); } @Test + @Disabled void testReadWithTimeout() throws InterruptedException { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - 100, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); // Short timeout + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); // Short timeout CountDownLatch latch = new CountDownLatch(1); CompletableFuture readTask = @@ -397,15 +250,7 @@ void testReadWithTimeout() throws InterruptedException { @Test void testConcurrentReadsAfterDataSet() throws InterruptedException, ExecutionException { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); block.setData(TEST_DATA_BYTES); int numThreads = 10; @@ -437,15 +282,7 @@ void testConcurrentReadsAfterDataSet() throws InterruptedException, ExecutionExc @Test void testCloseReleasesData() throws IOException { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); block.setData(TEST_DATA_BYTES); assertTrue(block.isDataReady()); @@ -458,15 +295,7 @@ void testCloseReleasesData() throws IOException { @Test void testMultipleSetDataCalls() { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); block.setData(TEST_DATA_BYTES); assertTrue(block.isDataReady()); @@ -479,33 +308,9 @@ void testMultipleSetDataCalls() { @Test void testGenerationProperty() { - Block block1 = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); - Block block2 = - new Block( - blockKey, - 5, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); - Block block3 = - new Block( - blockKey, - 100, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block1 = new Block(blockKey, 0, mockIndexCache, mockMetrics); + Block block2 = new Block(blockKey, 5, mockIndexCache, mockMetrics); + Block block3 = new Block(blockKey, 100, mockIndexCache, mockMetrics); assertEquals(0, block1.getGeneration()); assertEquals(5, block2.getGeneration()); @@ -514,15 +319,7 @@ void testGenerationProperty() { @Test void testReadIntoBuffer() throws IOException { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); block.setData(TEST_DATA_BYTES); byte[] buffer = new byte[20]; @@ -533,32 +330,18 @@ void testReadIntoBuffer() throws IOException { } @Test + @Disabled void testReadTimeoutIfDataNeverSet() { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - 100, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); // 100 ms + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); // 100 ms IOException ex = assertThrows(IOException.class, () -> block.read(0)); assertTrue(ex.getMessage().contains("Error while reading data.")); } @Test + @Disabled void testReadBlocksUntilDataIsReady() throws Exception { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - 1000, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); ExecutorService executor = Executors.newSingleThreadExecutor(); Future result = executor.submit(() -> block.read(0)); @@ -573,15 +356,7 @@ void testReadBlocksUntilDataIsReady() throws Exception { @Test void testReadHandlesInterruptedException() throws InterruptedException { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - 500, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); Thread testThread = new Thread( @@ -602,6 +377,7 @@ void testReadHandlesInterruptedException() throws InterruptedException { } @Test + @Disabled void testRetryStrategyWithTimeout() throws Exception { // Create a counter to track retry attempts AtomicInteger attempts = new AtomicInteger(0); @@ -611,8 +387,7 @@ void testRetryStrategyWithTimeout() throws Exception { CountDownLatch dataSetLatch = new CountDownLatch(1); // Create a block with a short timeout and 2 retries - Block block = - new Block(blockKey, 0, mockIndexCache, mockMetrics, 50, 2, OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); // Start a thread that will try to read from the block CompletableFuture future = @@ -645,10 +420,10 @@ void testRetryStrategyWithTimeout() throws Exception { } @Test + @Disabled void testRetryStrategyExhaustsRetries() { // Create a block with a short timeout and only 1 retry - Block block = - new Block(blockKey, 0, mockIndexCache, mockMetrics, 100, 1, OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); // Don't set data, so all retries will fail with timeout @@ -658,12 +433,12 @@ void testRetryStrategyExhaustsRetries() { } @Test + @Disabled void testRetryStrategyWithMultipleRetries() throws InterruptedException { // Create a block with a short timeout and multiple retries CountDownLatch readStarted = new CountDownLatch(1); - Block block = - new Block(blockKey, 0, mockIndexCache, mockMetrics, 100, 3, OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); // Start a thread that will try to read from the block CompletableFuture future = @@ -692,15 +467,7 @@ void testRetryStrategyWithMultipleRetries() throws InterruptedException { @Test void testSetErrorAndReadSingleByte() { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); IOException testError = new IOException("Test error message"); block.setError(testError); @@ -712,15 +479,7 @@ void testSetErrorAndReadSingleByte() { @Test void testSetErrorAndReadBuffer() { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); IOException testError = new IOException("Test error message"); block.setError(testError); @@ -732,30 +491,14 @@ void testSetErrorAndReadBuffer() { @Test void testSetErrorWithNullThrowsException() { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); assertThrows(NullPointerException.class, () -> block.setError(null)); } @Test void testConcurrentReadsWithError() throws InterruptedException { - Block block = - new Block( - blockKey, - 0, - mockIndexCache, - mockMetrics, - READ_TIMEOUT, - RETRY_COUNT, - OpenStreamInformation.DEFAULT); + Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); IOException testError = new IOException("Concurrent test error"); From 1736b0a775e7410ca828f71cafd2e8e7e3b169e1 Mon Sep 17 00:00:00 2001 From: Fuat Basik Date: Mon, 4 Aug 2025 10:40:37 +0100 Subject: [PATCH 2/8] Remove timeouts from telemetry --- .../common/telemetry/DefaultTelemetry.java | 21 ++++--------- .../common/telemetry/Telemetry.java | 31 +++++-------------- .../util/retry/DefaultRetryStrategyImpl.java | 6 +++- .../util/retry/IORunnable.java | 6 ++-- .../util/retry/IOSupplier.java | 6 ++-- .../common/telemetry/TelemetryTest.java | 24 ++++++-------- .../io/physical/data/MetadataStore.java | 5 +-- .../io/physical/reader/StreamReader.java | 6 ++-- 8 files changed, 34 insertions(+), 71 deletions(-) diff --git a/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/DefaultTelemetry.java b/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/DefaultTelemetry.java index 003457a7..2205c4ae 100644 --- a/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/DefaultTelemetry.java +++ b/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/DefaultTelemetry.java @@ -20,8 +20,6 @@ import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; import java.util.function.Predicate; import java.util.function.Supplier; import lombok.*; @@ -280,24 +278,19 @@ public void measure(@NonNull Metric metric, double value) { * @param level telemetry level. * @param operationSupplier operation to record this execution as. * @param operationCode the future to measure the execution of. - * @param operationTimeout Timeout duration (in milliseconds) for operation * @return an instance of {@link T} that returns the same result as the one passed in. - * @throws IOException if the underlying operation threw an IOException */ @Override public T measureJoin( @NonNull TelemetryLevel level, @NonNull OperationSupplier operationSupplier, - @NonNull CompletableFuture operationCode, - long operationTimeout) + @NonNull CompletableFuture operationCode) throws IOException { if (operationCode.isDone()) { - return handleCompletableFutureJoin(operationCode, operationTimeout); + return handleCompletableFutureJoin(operationCode); } else { return this.measure( - level, - operationSupplier, - () -> handleCompletableFutureJoin(operationCode, operationTimeout)); + level, operationSupplier, () -> handleCompletableFutureJoin(operationCode)); } } @@ -306,15 +299,13 @@ public T measureJoin( * * @param - return type of the CompletableFuture * @param future the CompletableFuture to join - * @param operationTimeout Timeout duration (in milliseconds) for operation * @return the result of the CompletableFuture * @throws IOException if the underlying future threw an IOException */ - private T handleCompletableFutureJoin(CompletableFuture future, long operationTimeout) - throws IOException { + private T handleCompletableFutureJoin(CompletableFuture future) throws IOException { try { - return future.get(operationTimeout, TimeUnit.MILLISECONDS); - } catch (ExecutionException | InterruptedException | TimeoutException e) { + return future.get(); + } catch (ExecutionException | InterruptedException e) { Throwable cause = e.getCause(); if (cause instanceof UncheckedIOException) { throw ((UncheckedIOException) cause).getCause(); diff --git a/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/Telemetry.java b/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/Telemetry.java index 8738b700..6bd2491a 100644 --- a/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/Telemetry.java +++ b/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/Telemetry.java @@ -81,15 +81,12 @@ CompletableFuture measure( * @param level telemetry level. * @param operationSupplier operation to record this execution as. * @param operationCode the future to measure the execution of. - * @param operationTimeout Timeout duration (in milliseconds) for operation * @return an instance of {@link T} that returns the same result as the one passed in. - * @throws IOException if the underlying operation threw an IOException */ default T measureJoin( @NonNull TelemetryLevel level, @NonNull OperationSupplier operationSupplier, - @NonNull CompletableFuture operationCode, - long operationTimeout) + @NonNull CompletableFuture operationCode) throws IOException { if (operationCode.isDone()) { return operationCode.join(); @@ -151,16 +148,12 @@ default CompletableFuture measureCritical( * @param - return type of the {@link CompletableFuture}. * @param operationSupplier operation to record this execution as. * @param operationCode the future to measure the execution of. - * @param operationTimeout Timeout duration (in milliseconds) for operation * @return an instance of {@link T} that returns the same result as the one passed in. * @throws IOException if the underlying operation threw an IOException */ default T measureJoinCritical( - OperationSupplier operationSupplier, - CompletableFuture operationCode, - long operationTimeout) - throws IOException { - return measureJoin(TelemetryLevel.CRITICAL, operationSupplier, operationCode, operationTimeout); + OperationSupplier operationSupplier, CompletableFuture operationCode) throws IOException { + return measureJoin(TelemetryLevel.CRITICAL, operationSupplier, operationCode); } /** @@ -216,16 +209,11 @@ default CompletableFuture measureStandard( * @param - return type of the {@link CompletableFuture}. * @param operationSupplier operation to record this execution as. * @param operationCode the future to measure the execution of. - * @param operationTimeout Timeout duration (in milliseconds) for operation * @return an instance of {@link T} that returns the same result as the one passed in. - * @throws IOException if the underlying operation threw an IOException */ default T measureJoinStandard( - OperationSupplier operationSupplier, - CompletableFuture operationCode, - long operationTimeout) - throws IOException { - return measureJoin(TelemetryLevel.STANDARD, operationSupplier, operationCode, operationTimeout); + OperationSupplier operationSupplier, CompletableFuture operationCode) throws IOException { + return measureJoin(TelemetryLevel.STANDARD, operationSupplier, operationCode); } /** @@ -281,16 +269,11 @@ default CompletableFuture measureVerbose( * @param - return type of the {@link CompletableFuture}. * @param operationSupplier operation to record this execution as. * @param operationCode the future to measure the execution of. - * @param operationTimeout Timeout duration (in milliseconds) for operation * @return an instance of {@link T} that returns the same result as the one passed in. - * @throws IOException if the underlying operation threw an IOException */ default T measureJoinVerbose( - OperationSupplier operationSupplier, - CompletableFuture operationCode, - long operationTimeout) - throws IOException { - return measureJoin(TelemetryLevel.VERBOSE, operationSupplier, operationCode, operationTimeout); + OperationSupplier operationSupplier, CompletableFuture operationCode) throws IOException { + return measureJoin(TelemetryLevel.VERBOSE, operationSupplier, operationCode); } /** diff --git a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java index b36211a8..240066e4 100644 --- a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java +++ b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java @@ -157,7 +157,11 @@ private Exception handleExceptionAfterRetry(Exception e) { } /** - * @inheritDoc + * Create a timeout for read from storage operations and with specified retry count. This will + * override settings in PhysicalIOConfiguration (blockreadtimeout and blockreadretrycount) if set. + * If user does not set a timeout in their retry strategy, a timeout will be set based on + * aforementioned configuration. set blockreadtimeout = 0 to disable timeouts + * * @param timeoutDurationMillis Timeout duration for reading from storage * @param retryCount Number of times to retry if Timeout Exceeds */ diff --git a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/IORunnable.java b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/IORunnable.java index 5043251e..19b9b962 100644 --- a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/IORunnable.java +++ b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/IORunnable.java @@ -15,15 +15,13 @@ */ package software.amazon.s3.analyticsaccelerator.util.retry; -import java.io.IOException; - /** A functional interface that mimics {@link Runnable}, but allows IOException to be thrown. */ public interface IORunnable { /** * Functional representation of the code that takes no parameters and returns no value. The code * is allowed to throw any exception. * - * @throws IOException on error condition. + * @throws Exception on error condition. */ - void apply() throws IOException; + void apply() throws Exception; } diff --git a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/IOSupplier.java b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/IOSupplier.java index 6ca33f2a..7e188337 100644 --- a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/IOSupplier.java +++ b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/IOSupplier.java @@ -15,8 +15,6 @@ */ package software.amazon.s3.analyticsaccelerator.util.retry; -import java.io.IOException; - /** * A function that mimics {@link java.util.function.Supplier}, but allows IOException to be thrown * and returns T. @@ -29,7 +27,7 @@ public interface IOSupplier { * {@link T}. The code is allowed to throw any exception. * * @return a value of type {@link T}. - * @throws IOException on error condition. + * @throws Exception on error condition. */ - T apply() throws IOException, InterruptedException; + T apply() throws Exception; } diff --git a/common/src/test/java/software/amazon/s3/analyticsaccelerator/common/telemetry/TelemetryTest.java b/common/src/test/java/software/amazon/s3/analyticsaccelerator/common/telemetry/TelemetryTest.java index 35d934cc..b060b1d3 100644 --- a/common/src/test/java/software/amazon/s3/analyticsaccelerator/common/telemetry/TelemetryTest.java +++ b/common/src/test/java/software/amazon/s3/analyticsaccelerator/common/telemetry/TelemetryTest.java @@ -27,7 +27,6 @@ value = "NP_NONNULL_PARAM_VIOLATION", justification = "We mean to pass nulls to checks") public class TelemetryTest { - private static final long DEFAULT_TIMEOUT = 120_000; @Test void testCreateTelemetry() { @@ -80,7 +79,7 @@ void testMeasureJoin() throws Exception { // This will complete the future completionThread.start(); - defaultTelemetry.measureJoinCritical(() -> operation, completableFuture, DEFAULT_TIMEOUT); + defaultTelemetry.measureJoinCritical(() -> operation, completableFuture); assertTrue(completableFuture.isDone()); assertFalse(completableFuture.isCompletedExceptionally()); assertEquals(42, completableFuture.get()); @@ -97,8 +96,7 @@ void testMeasureJoin() throws Exception { assertEquals(Optional.empty(), operationMeasurement.getError()); // Try again - nothing should be recorded - long result = - defaultTelemetry.measureJoinStandard(() -> operation, completableFuture, DEFAULT_TIMEOUT); + long result = defaultTelemetry.measureJoinStandard(() -> operation, completableFuture); assertEquals(1, reporter.getOperationCompletions().size()); assertEquals(42, result); } @@ -132,7 +130,7 @@ void testMeasureJoinStandard() throws Exception { // This will complete the future completionThread.start(); - defaultTelemetry.measureJoinStandard(() -> operation, completableFuture, DEFAULT_TIMEOUT); + defaultTelemetry.measureJoinStandard(() -> operation, completableFuture); assertTrue(completableFuture.isDone()); assertFalse(completableFuture.isCompletedExceptionally()); assertEquals(42, completableFuture.get()); @@ -149,8 +147,7 @@ void testMeasureJoinStandard() throws Exception { assertEquals(Optional.empty(), operationMeasurement.getError()); // Try again - nothing should be recorded - long result = - defaultTelemetry.measureJoinStandard(() -> operation, completableFuture, DEFAULT_TIMEOUT); + long result = defaultTelemetry.measureJoinStandard(() -> operation, completableFuture); assertEquals(1, reporter.getOperationCompletions().size()); assertEquals(42, result); } @@ -184,8 +181,7 @@ void testMeasureJoinVerbose() throws Exception { // This will complete the future completionThread.start(); - Long result = - defaultTelemetry.measureJoinVerbose(() -> operation, completableFuture, DEFAULT_TIMEOUT); + Long result = defaultTelemetry.measureJoinVerbose(() -> operation, completableFuture); assertEquals(42L, result); assertTrue(completableFuture.isDone()); assertFalse(completableFuture.isCompletedExceptionally()); @@ -203,8 +199,7 @@ void testMeasureJoinVerbose() throws Exception { assertEquals(Optional.empty(), operationMeasurement.getError()); // Try again - nothing should be recorded - result = - defaultTelemetry.measureJoinStandard(() -> operation, completableFuture, DEFAULT_TIMEOUT); + result = defaultTelemetry.measureJoinStandard(() -> operation, completableFuture); assertEquals(1, reporter.getOperationCompletions().size()); assertEquals(42, result); } @@ -227,15 +222,14 @@ void testMeasureJoinCheckNulls() throws Exception { assertThrows( NullPointerException.class, - () -> defaultTelemetry.measureJoinStandard(null, completableFuture, DEFAULT_TIMEOUT)); + () -> defaultTelemetry.measureJoinStandard(null, completableFuture)); assertThrows( NullPointerException.class, - () -> - defaultTelemetry.measureJoinStandard(() -> null, completableFuture, DEFAULT_TIMEOUT)); + () -> defaultTelemetry.measureJoinStandard(() -> null, completableFuture)); assertThrows( NullPointerException.class, - () -> defaultTelemetry.measureJoinStandard(() -> operation, null, DEFAULT_TIMEOUT)); + () -> defaultTelemetry.measureJoinStandard(() -> operation, null)); } } diff --git a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/MetadataStore.java b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/MetadataStore.java index 31e83f3d..4dfe5ad2 100644 --- a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/MetadataStore.java +++ b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/MetadataStore.java @@ -46,7 +46,6 @@ public class MetadataStore implements Closeable { private final ObjectClient objectClient; private final Telemetry telemetry; private final Map> cache; - private final PhysicalIOConfiguration configuration; private final Metrics aggregatingMetrics; private static final Logger LOG = LoggerFactory.getLogger(MetadataStore.class); @@ -78,7 +77,6 @@ protected boolean removeEldestEntry( return this.size() > configuration.getMetadataStoreCapacity(); } }); - this.configuration = configuration; } /** @@ -98,8 +96,7 @@ public ObjectMetadata get(S3URI s3URI, OpenStreamInformation openStreamInformati .name(OPERATION_METADATA_HEAD_JOIN) .attribute(StreamAttributes.uri(s3URI)) .build(), - this.asyncGet(s3URI, openStreamInformation), - this.configuration.getBlockReadTimeout()); + this.asyncGet(s3URI, openStreamInformation)); } /** diff --git a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java index 454b0145..e2ec6c07 100644 --- a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java +++ b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java @@ -206,8 +206,7 @@ private Runnable processReadTask(final List blocks, ReadMode readMode) { removeNonFilledBlocksFromStore(nonFilledBlocks); } }); - } // Process the input stream and populate data blocks - catch (Exception e) { + } catch (Exception e) { LOG.error("Unexpected exception while reading blocks", e); if (e instanceof IOException) { setErrorOnBlocksAndRemove(blocks, (IOException) e); @@ -275,8 +274,7 @@ private ObjectContent fetchObjectContent(GetRequest getRequest) throws IOExcepti .attribute(StreamAttributes.rangeLength(getRequest.getRange().getLength())) .attribute(StreamAttributes.range(getRequest.getRange())) .build(), - this.objectClient.getObject(getRequest, this.openStreamInformation), - 100000000); + this.objectClient.getObject(getRequest, this.openStreamInformation)); } /** From 47c7ca248b8351e3eedec82441c5282f62a7667f Mon Sep 17 00:00:00 2001 From: Fuat Basik Date: Mon, 4 Aug 2025 10:48:12 +0100 Subject: [PATCH 3/8] Remove disabled tests --- .../io/physical/data/BlockTest.java | 143 +----------------- 1 file changed, 1 insertion(+), 142 deletions(-) diff --git a/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockTest.java b/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockTest.java index c9d27aee..1a0ae63a 100644 --- a/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockTest.java +++ b/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockTest.java @@ -24,9 +24,7 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.*; -import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import software.amazon.s3.analyticsaccelerator.common.Metrics; import software.amazon.s3.analyticsaccelerator.request.Range; @@ -207,47 +205,6 @@ void testReadBufferWithInvalidParameters() { assertThrows(IllegalArgumentException.class, () -> block.read(buffer, 4, 1, 0)); } - @Test - @Disabled - void testReadBeforeDataSet() { - Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); // Short timeout - - assertThrows(IOException.class, () -> block.read(0)); - } - - @Test - @Disabled - void testReadBufferBeforeDataSet() { - Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); // Short timeout - byte[] buffer = new byte[4]; - - assertThrows(IOException.class, () -> block.read(buffer, 0, 4, 0)); - } - - @Test - @Disabled - void testReadWithTimeout() throws InterruptedException { - Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); // Short timeout - - CountDownLatch latch = new CountDownLatch(1); - CompletableFuture readTask = - CompletableFuture.runAsync( - () -> { - try { - latch.countDown(); - block.read(0); - fail("Expected IOException due to timeout"); - } catch (IOException e) { - // Expected - } - }); - - latch.await(); // Wait for read to start - Thread.sleep(200); // Wait longer than timeout - - assertDoesNotThrow(() -> readTask.get(1, TimeUnit.SECONDS)); - } - @Test void testConcurrentReadsAfterDataSet() throws InterruptedException, ExecutionException { Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); @@ -330,16 +287,6 @@ void testReadIntoBuffer() throws IOException { } @Test - @Disabled - void testReadTimeoutIfDataNeverSet() { - Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); // 100 ms - - IOException ex = assertThrows(IOException.class, () -> block.read(0)); - assertTrue(ex.getMessage().contains("Error while reading data.")); - } - - @Test - @Disabled void testReadBlocksUntilDataIsReady() throws Exception { Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); @@ -352,6 +299,7 @@ void testReadBlocksUntilDataIsReady() throws Exception { assertEquals(Byte.toUnsignedInt(TEST_DATA_BYTES[0]), result.get(1, TimeUnit.SECONDS)); executor.shutdown(); + block.close(); } @Test @@ -376,95 +324,6 @@ void testReadHandlesInterruptedException() throws InterruptedException { testThread.join(); } - @Test - @Disabled - void testRetryStrategyWithTimeout() throws Exception { - // Create a counter to track retry attempts - AtomicInteger attempts = new AtomicInteger(0); - - // Create a custom CountDownLatch to coordinate the test - CountDownLatch attemptLatch = new CountDownLatch(1); - CountDownLatch dataSetLatch = new CountDownLatch(1); - - // Create a block with a short timeout and 2 retries - Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); - - // Start a thread that will try to read from the block - CompletableFuture future = - CompletableFuture.supplyAsync( - () -> { - try { - // Record the attempt and notify the test thread - attempts.incrementAndGet(); - attemptLatch.countDown(); - - // Try to read - this will trigger retries internally - return block.read(0); - } catch (IOException e) { - throw new CompletionException(e); - } - }); - - // Wait for the read attempt to start - assertTrue(attemptLatch.await(500, TimeUnit.MILLISECONDS)); - - // Wait a bit to ensure at least one retry happens (50ms timeout * 1 retry) - Thread.sleep(75); - - // Now set the data so the next retry will succeed - block.setData(TEST_DATA_BYTES); - dataSetLatch.countDown(); - - // The read should eventually succeed - assertEquals(Byte.toUnsignedInt(TEST_DATA_BYTES[0]), future.join()); - } - - @Test - @Disabled - void testRetryStrategyExhaustsRetries() { - // Create a block with a short timeout and only 1 retry - Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); - - // Don't set data, so all retries will fail with timeout - - // This should fail after max retries - IOException exception = assertThrows(IOException.class, () -> block.read(0)); - assertTrue(exception.getMessage().contains("timed out")); - } - - @Test - @Disabled - void testRetryStrategyWithMultipleRetries() throws InterruptedException { - // Create a block with a short timeout and multiple retries - CountDownLatch readStarted = new CountDownLatch(1); - - Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); - - // Start a thread that will try to read from the block - CompletableFuture future = - CompletableFuture.supplyAsync( - () -> { - try { - readStarted.countDown(); - return block.read(0); - } catch (IOException e) { - throw new CompletionException(e); - } - }); - - // Wait for read to start - assertTrue(readStarted.await(500, TimeUnit.MILLISECONDS)); - - // Wait a bit to ensure at least one retry happens - Thread.sleep(250); - - // Now set the data - block.setData(TEST_DATA_BYTES); - - // The read should eventually succeed - assertEquals(Byte.toUnsignedInt(TEST_DATA_BYTES[0]), future.join()); - } - @Test void testSetErrorAndReadSingleByte() { Block block = new Block(blockKey, 0, mockIndexCache, mockMetrics); From 7fb58df1fdba53759ee7bb7b3d3085d0d89428c4 Mon Sep 17 00:00:00 2001 From: Fuat Basik Date: Mon, 4 Aug 2025 10:55:51 +0100 Subject: [PATCH 4/8] Update javadocs --- docs/index-all.html | 8 ++++ .../util/package-tree.html | 2 +- .../util/retry/DefaultRetryStrategyImpl.html | 47 ++++++++++++++----- .../util/retry/RetryStrategy.html | 40 ++++++++++++---- 4 files changed, 74 insertions(+), 23 deletions(-) diff --git a/docs/index-all.html b/docs/index-all.html index 7e61c9e7..b8280a53 100644 --- a/docs/index-all.html +++ b/docs/index-all.html @@ -562,6 +562,14 @@

S

T

+
timeout(long, int) - Method in class software.amazon.s3.analyticsaccelerator.util.retry.DefaultRetryStrategyImpl
+
+
Create a timeout for read from storage operations and with specified retry count.
+
+
timeout(long, int) - Method in interface software.amazon.s3.analyticsaccelerator.util.retry.RetryStrategy
+
+
Create a timeout for read from storage operations and with specified retry count.
+
toString() - Method in class software.amazon.s3.analyticsaccelerator.request.ObjectMetadata.ObjectMetadataBuilder
 
toString() - Method in class software.amazon.s3.analyticsaccelerator.request.ObjectMetadata
diff --git a/docs/software/amazon/s3/analyticsaccelerator/util/package-tree.html b/docs/software/amazon/s3/analyticsaccelerator/util/package-tree.html index ef8c48f9..2d97f172 100644 --- a/docs/software/amazon/s3/analyticsaccelerator/util/package-tree.html +++ b/docs/software/amazon/s3/analyticsaccelerator/util/package-tree.html @@ -91,8 +91,8 @@

Enum Hierarchy

  • java.lang.Enum<E> (implements java.lang.Comparable<T>, java.io.Serializable)
      -
    • software.amazon.s3.analyticsaccelerator.util.PrefetchMode
    • software.amazon.s3.analyticsaccelerator.util.InputPolicy
      • +
    • software.amazon.s3.analyticsaccelerator.util.PrefetchMode
diff --git a/docs/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.html b/docs/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.html index a9ea1d72..28171e71 100644 --- a/docs/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.html +++ b/docs/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.html @@ -17,7 +17,7 @@ catch(err) { } //--> -var methods = {"i0":10,"i1":10,"i2":10,"i3":10,"i4":10}; +var methods = {"i0":10,"i1":10,"i2":10,"i3":10,"i4":10,"i5":10}; var tabs = {65535:["t0","All Methods"],2:["t2","Instance Methods"],8:["t4","Concrete Methods"]}; var altColor = "altColor"; var rowColor = "rowColor"; @@ -117,8 +117,9 @@

Class DefaultRetryStrat
Retry strategy implementation for seekable input stream operations. Uses Failsafe library to execute operations with configurable retry policies. -

This strategy will be additive to readTimeout and readRetryCount set on PhysicalIO - configuration.

+

If provided with a timeout this strategy will overwrite readTimeout and readRetryCount set on + PhysicalIOConfiguration. If not, values from PhysicalIOConfiguration will be used to manage + storage read timeouts. @@ -197,6 +198,13 @@

Method Summary

Merge two retry strategies and return a new RetryStrategy.
+ +void +timeout(long timeoutDurationMillis, + int retryCount) +
Create a timeout for read from storage operations and with specified retry count.
+ +
+ + + +
    +
  • +

    timeout

    +
    public void timeout(long timeoutDurationMillis,
+                    int retryCount)
    +
    Create a timeout for read from storage operations and with specified retry count. This will + override settings in PhysicalIOConfiguration (blockreadtimeout and blockreadretrycount) if set. + If user does not set a timeout in their retry strategy, a timeout will be set based on + aforementioned configuration. set blockreadtimeout = 0 to disable timeouts
    +
    +
    Specified by:
    +
    timeout in interface RetryStrategy
    +
    Parameters:
    +
    timeoutDurationMillis - Timeout duration for reading from storage
    +
    retryCount - Number of times to retry if Timeout Exceeds
    +
    +
    • +
diff --git a/docs/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.html b/docs/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.html index 353aae3d..2090322b 100644 --- a/docs/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.html +++ b/docs/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.html @@ -17,7 +17,7 @@ catch(err) { } //--> -var methods = {"i0":6,"i1":6,"i2":6,"i3":6,"i4":6}; +var methods = {"i0":6,"i1":6,"i2":6,"i3":6,"i4":6,"i5":6}; var tabs = {65535:["t0","All Methods"],2:["t2","Instance Methods"],4:["t3","Abstract Methods"]}; var altColor = "altColor"; var rowColor = "rowColor"; @@ -153,6 +153,13 @@

Method Summary

Merge two retry strategies and return a new RetryStrategy.
+ +void +timeout(long durationInMillis, + int retryCount) +
Create a timeout for read from storage operations and with specified retry count.
+ + @@ -174,14 +181,11 @@

Method Detail

  • execute

    -
    void execute(software.amazon.s3.analyticsaccelerator.util.retry.IORunnable runnable)
-      throws java.io.IOException
    +
    void execute(software.amazon.s3.analyticsaccelerator.util.retry.IORunnable runnable)
    Executes a runnable with retry logic.
    Parameters:
    runnable - the operation to execute
    -
    Throws:
    -
    java.io.IOException - if the operation fails after all retry attempts
@@ -191,8 +195,7 @@

execute

  • get

    -
    <T> T get(software.amazon.s3.analyticsaccelerator.util.retry.IOSupplier<T> supplier)
-   throws java.io.IOException
    +
    <T> T get(software.amazon.s3.analyticsaccelerator.util.retry.IOSupplier<T> supplier)
    Executes a supplier with retry logic.
    Type Parameters:
    @@ -201,8 +204,6 @@

    get

    supplier - the operation to execute
    Returns:
    result of the supplier
    -
    Throws:
    -
    java.io.IOException - if the operation fails after all retry attempts
@@ -243,7 +244,7 @@

merge

-
    +

    • getRetryPolicies

      java.util.List<RetryPolicy> getRetryPolicies()
      @@ -254,6 +255,25 @@

      getRetryPolicies

    + + + +
      +
    • +

      timeout

      +
      void timeout(long durationInMillis,
+             int retryCount)
      +
      Create a timeout for read from storage operations and with specified retry count. This will + override settings in PhysicalIOConfiguration (blockreadtimeout and blockreadretrycount) if set. + If user does not set a timeout in their retry strategy, a timeout will be set based on + aforementioned configuration. set blockreadtimeout = 0 to disable timeouts.
      +
      +
      Parameters:
      +
      durationInMillis - Timeout duration for reading from storage
      +
      retryCount - Number of times to retry if Timeout Exceeds
      +
      +
      • +
From a0adea0b9038b74c316d3c6cf14ee5968d70cc77 Mon Sep 17 00:00:00 2001 From: Fuat Basik Date: Mon, 4 Aug 2025 14:16:50 +0100 Subject: [PATCH 5/8] Fix flaky read vectored test --- .../access/ReadVectoredTest.java | 16 ++++++++++------ .../io/physical/data/BlockManagerTest.java | 14 +++++++++++--- 2 files changed, 21 insertions(+), 9 deletions(-) diff --git a/input-stream/src/integrationTest/java/software/amazon/s3/analyticsaccelerator/access/ReadVectoredTest.java b/input-stream/src/integrationTest/java/software/amazon/s3/analyticsaccelerator/access/ReadVectoredTest.java index 7fcd97d0..0826225c 100644 --- a/input-stream/src/integrationTest/java/software/amazon/s3/analyticsaccelerator/access/ReadVectoredTest.java +++ b/input-stream/src/integrationTest/java/software/amazon/s3/analyticsaccelerator/access/ReadVectoredTest.java @@ -234,17 +234,21 @@ void testSomeRangesFail() throws IOException { objectRanges.add(new ObjectRange(new CompletableFuture<>(), 500 * ONE_MB, 500)); s3SeekableInputStream.readVectored(objectRanges, allocate, LOG_BYTE_BUFFER_RELEASED); - - assertThrows(CompletionException.class, () -> objectRanges.get(0).getByteBuffer().join()); - assertDoesNotThrow(() -> objectRanges.get(1).getByteBuffer().join()); - assertDoesNotThrow(() -> objectRanges.get(2).getByteBuffer().join()); + try { + // One of the joins must throw but we dont know which one due to asynchrony + objectRanges.get(0).getByteBuffer().join(); + objectRanges.get(1).getByteBuffer().join(); + objectRanges.get(2).getByteBuffer().join(); + } catch (Exception e) { + assertInstanceOf(CompletionException.class, e); + } assertEquals( + 3, s3AALClientStreamReader .getS3SeekableInputStreamFactory() .getMetrics() - .get(MetricKey.GET_REQUEST_COUNT), - 3); + .get(MetricKey.GET_REQUEST_COUNT)); } } diff --git a/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockManagerTest.java b/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockManagerTest.java index 34f99b36..1326bcd0 100644 --- a/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockManagerTest.java +++ b/input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/io/physical/data/BlockManagerTest.java @@ -708,11 +708,19 @@ void testMakeRangeAvailableAsync(ReadMode readMode) { verify(objectClient, timeout(1_000).times(3)).getObject(requestCaptor.capture(), any()); List getRequestList = requestCaptor.getAllValues(); + int count5MBRequests = 0; + int count3MBRequests = 0; + for (GetRequest request : getRequestList) { + if (request.getRange().getLength() == 5 * ONE_MB) { + count5MBRequests++; + } else if (request.getRange().getLength() == 3 * ONE_MB) { + count3MBRequests++; + } + } // Verify that prefetch modes don't trigger sequential prefetching - assertEquals(getRequestList.get(0).getRange().getLength(), 5 * ONE_MB); - assertEquals(getRequestList.get(1).getRange().getLength(), 3 * ONE_MB); - assertEquals(getRequestList.get(2).getRange().getLength(), 5 * ONE_MB); + assertEquals(2, count5MBRequests); + assertEquals(1, count3MBRequests); } private static List readModes() { From 37d318afd342f7e3ed05cf865f6caf15fb38bebd Mon Sep 17 00:00:00 2001 From: Fuat Basik Date: Mon, 4 Aug 2025 17:17:30 +0100 Subject: [PATCH 6/8] Fix static OpenStreamInformation --- .../util/OpenStreamInformation.java | 10 +++- .../util/retry/DefaultRetryStrategyImpl.java | 49 ++++++++++--------- .../util/retry/RetryStrategy.java | 7 +++ .../retry/DefaultRetryStrategyImplTest.java | 4 -- .../access/ReadVectoredTest.java | 21 ++++---- .../io/physical/reader/StreamReader.java | 2 +- 6 files changed, 54 insertions(+), 39 deletions(-) diff --git a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/OpenStreamInformation.java b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/OpenStreamInformation.java index 73bfde15..e33b9001 100644 --- a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/OpenStreamInformation.java +++ b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/OpenStreamInformation.java @@ -46,9 +46,17 @@ public class OpenStreamInformation { private final InputPolicy inputPolicy; @Builder.Default private RequestCallback requestCallback = new DefaultRequestCallbackImpl(); private final EncryptionSecrets encryptionSecrets; - @Builder.Default private final RetryStrategy retryStrategy = new DefaultRetryStrategyImpl(); /** Default set of settings for {@link OpenStreamInformation} */ public static final OpenStreamInformation DEFAULT = OpenStreamInformation.builder().build(); + + /** + * Default set of settings for {@link OpenStreamInformation} + * + * @return new OpenStreamInformation instance + */ + public static OpenStreamInformation ofDefaults() { + return OpenStreamInformation.builder().build(); + } } diff --git a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java index 240066e4..75cefeb8 100644 --- a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java +++ b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java @@ -26,6 +26,7 @@ import java.util.List; import java.util.Optional; import java.util.stream.Collectors; +import lombok.Getter; import lombok.SneakyThrows; import software.amazon.s3.analyticsaccelerator.common.Preconditions; @@ -39,13 +40,12 @@ */ public class DefaultRetryStrategyImpl implements RetryStrategy { private final List retryPolicies; - FailsafeExecutor failsafeExecutor; - private boolean timeoutSet; + private Timeout timeoutPolicy; + @Getter private boolean timeoutSet; /** Creates a retry strategy with no retry policies (no retries). */ public DefaultRetryStrategyImpl() { this.retryPolicies = new ArrayList<>(); - this.failsafeExecutor = Failsafe.none(); } /** @@ -62,7 +62,6 @@ public DefaultRetryStrategyImpl(RetryPolicy outerPolicy, RetryPolicy... policies if (policies != null && policies.length > 0) { this.retryPolicies.addAll(Arrays.asList(policies)); } - this.failsafeExecutor = Failsafe.with(getDelegates()); } /** @@ -74,7 +73,6 @@ public DefaultRetryStrategyImpl(List policies) { Preconditions.checkNotNull(policies); this.retryPolicies = new ArrayList<>(); this.retryPolicies.addAll(policies); - this.failsafeExecutor = Failsafe.with(getDelegates()); } /** @@ -86,7 +84,7 @@ public DefaultRetryStrategyImpl(List policies) { @SneakyThrows public void execute(IORunnable runnable) { try { - this.failsafeExecutor.run(runnable::apply); + executor().run(runnable::apply); } catch (Exception ex) { throw handleExceptionAfterRetry(ex); } @@ -103,7 +101,7 @@ public void execute(IORunnable runnable) { @SneakyThrows public T get(IOSupplier supplier) { try { - return this.failsafeExecutor.get(supplier::apply); + return executor().get(supplier::apply); } catch (Exception ex) { throw handleExceptionAfterRetry(ex); } @@ -112,16 +110,14 @@ public T get(IOSupplier supplier) { @Override public RetryStrategy amend(RetryPolicy policy) { Preconditions.checkNotNull(policy); - this.failsafeExecutor = this.failsafeExecutor.compose(policy.getDelegate()); + this.retryPolicies.add(policy); return this; } @Override public RetryStrategy merge(RetryStrategy strategy) { Preconditions.checkNotNull(strategy); - for (RetryPolicy policy : strategy.getRetryPolicies()) { - this.failsafeExecutor = this.failsafeExecutor.compose(policy.getDelegate()); - } + this.retryPolicies.addAll(strategy.getRetryPolicies()); return this; } @@ -156,6 +152,17 @@ private Exception handleExceptionAfterRetry(Exception e) { return e; } + private FailsafeExecutor executor() { + FailsafeExecutor executor; + if (retryPolicies.isEmpty()) { + executor = Failsafe.none(); + } else { + executor = Failsafe.with(getDelegates()); + } + if (this.timeoutSet) executor = executor.compose(timeoutPolicy); + return executor; + } + /** * Create a timeout for read from storage operations and with specified retry count. This will * override settings in PhysicalIOConfiguration (blockreadtimeout and blockreadretrycount) if set. @@ -166,16 +173,14 @@ private Exception handleExceptionAfterRetry(Exception e) { * @param retryCount Number of times to retry if Timeout Exceeds */ public void timeout(long timeoutDurationMillis, int retryCount) { - if (!this.timeoutSet) { - Timeout timeout = - Timeout.builder(Duration.ofMillis(timeoutDurationMillis)).withInterrupt().build(); - dev.failsafe.RetryPolicy timeoutPolicy = - dev.failsafe.RetryPolicy.builder() - .handle(TimeoutExceededException.class) - .withMaxRetries(retryCount) - .build(); - this.failsafeExecutor = this.failsafeExecutor.compose(timeoutPolicy).compose(timeout); - this.timeoutSet = true; - } + this.timeoutPolicy = + Timeout.builder(Duration.ofMillis(timeoutDurationMillis)).withInterrupt().build(); + RetryPolicy timeoutRetries = + RetryPolicy.builder() + .handle(TimeoutExceededException.class) + .withMaxRetries(retryCount) + .build(); + this.retryPolicies.add(timeoutRetries); + this.timeoutSet = true; } } diff --git a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.java b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.java index 1fa0867f..142e46fd 100644 --- a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.java +++ b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.java @@ -70,4 +70,11 @@ public interface RetryStrategy { * @param retryCount Number of times to retry if Timeout Exceeds */ void timeout(long durationInMillis, int retryCount); + + /** + * Method to check if timeout of the strategy already set. + * + * @return timeoutSet + */ + boolean isTimeoutSet(); } diff --git a/common/src/test/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImplTest.java b/common/src/test/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImplTest.java index 8b3e496f..4f463d7e 100644 --- a/common/src/test/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImplTest.java +++ b/common/src/test/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImplTest.java @@ -19,7 +19,6 @@ import java.io.IOException; import java.nio.charset.StandardCharsets; -import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.concurrent.TimeoutException; @@ -50,9 +49,6 @@ void testPolicyConstructor() { assertNotNull(executor); assertThrows(NullPointerException.class, () -> new DefaultRetryStrategyImpl(null)); - - ArrayList emptyList = new ArrayList(); - assertThrows(IllegalArgumentException.class, () -> new DefaultRetryStrategyImpl(emptyList)); } @Test diff --git a/input-stream/src/integrationTest/java/software/amazon/s3/analyticsaccelerator/access/ReadVectoredTest.java b/input-stream/src/integrationTest/java/software/amazon/s3/analyticsaccelerator/access/ReadVectoredTest.java index 0826225c..cde69c57 100644 --- a/input-stream/src/integrationTest/java/software/amazon/s3/analyticsaccelerator/access/ReadVectoredTest.java +++ b/input-stream/src/integrationTest/java/software/amazon/s3/analyticsaccelerator/access/ReadVectoredTest.java @@ -106,7 +106,7 @@ void testEmptyRanges() throws IOException { S3SeekableInputStream s3SeekableInputStream = s3AALClientStreamReader.createReadStream( - S3Object.RANDOM_1GB, OpenStreamInformation.DEFAULT); + S3Object.RANDOM_1GB, OpenStreamInformation.ofDefaults()); List objectRanges = new ArrayList<>(); @@ -133,7 +133,7 @@ void testEoFRanges() throws IOException { S3SeekableInputStream s3SeekableInputStream = s3AALClientStreamReader.createReadStream( - S3Object.RANDOM_1GB, OpenStreamInformation.DEFAULT); + S3Object.RANDOM_1GB, OpenStreamInformation.ofDefaults()); List objectRanges = new ArrayList<>(); @@ -164,7 +164,7 @@ void testNullRange() throws IOException { S3SeekableInputStream s3SeekableInputStream = s3AALClientStreamReader.createReadStream( - S3Object.RANDOM_1GB, OpenStreamInformation.DEFAULT); + S3Object.RANDOM_1GB, OpenStreamInformation.ofDefaults()); List objectRanges = new ArrayList<>(); objectRanges.add(new ObjectRange(new CompletableFuture<>(), 700, 500)); @@ -194,7 +194,7 @@ void testOverlappingRanges() throws IOException { S3SeekableInputStream s3SeekableInputStream = s3AALClientStreamReader.createReadStream( - S3Object.RANDOM_1GB, OpenStreamInformation.DEFAULT); + S3Object.RANDOM_1GB, OpenStreamInformation.ofDefaults()); List objectRanges = new ArrayList<>(); objectRanges.add(new ObjectRange(new CompletableFuture<>(), 700, 500)); @@ -226,7 +226,7 @@ void testSomeRangesFail() throws IOException { S3SeekableInputStream s3SeekableInputStream = s3AALClientStreamReader.createReadStream( - S3Object.RANDOM_1GB, OpenStreamInformation.DEFAULT); + S3Object.RANDOM_1GB, OpenStreamInformation.ofDefaults()); List objectRanges = new ArrayList<>(); objectRanges.add(new ObjectRange(new CompletableFuture<>(), 700, 500)); @@ -242,7 +242,6 @@ void testSomeRangesFail() throws IOException { } catch (Exception e) { assertInstanceOf(CompletionException.class, e); } - assertEquals( 3, s3AALClientStreamReader @@ -283,7 +282,7 @@ private void performReadVectored(S3AALClientStreamReader s3AALClientStreamReader try { S3SeekableInputStream s3SeekableInputStream = s3AALClientStreamReader.createReadStream( - S3Object.RANDOM_1GB, OpenStreamInformation.DEFAULT); + S3Object.RANDOM_1GB, OpenStreamInformation.ofDefaults()); List objectRanges = new ArrayList<>(); objectRanges.add(new ObjectRange(new CompletableFuture<>(), 700, 500)); @@ -338,7 +337,7 @@ protected void testReadVectored( this.createS3AALClientStreamReader(s3ClientKind, AALInputStreamConfigurationKind)) { S3SeekableInputStream s3SeekableInputStream = - s3AALClientStreamReader.createReadStream(s3Object, OpenStreamInformation.DEFAULT); + s3AALClientStreamReader.createReadStream(s3Object, OpenStreamInformation.ofDefaults()); List objectRanges = new ArrayList<>(); objectRanges.add(new ObjectRange(new CompletableFuture<>(), 50, ONE_MB)); @@ -389,7 +388,7 @@ protected void testReadVectoredInSingleBlock( this.createS3AALClientStreamReader(s3ClientKind, AALInputStreamConfigurationKind)) { S3SeekableInputStream s3SeekableInputStream = - s3AALClientStreamReader.createReadStream(s3Object, OpenStreamInformation.DEFAULT); + s3AALClientStreamReader.createReadStream(s3Object, OpenStreamInformation.ofDefaults()); List objectRanges = new ArrayList<>(); objectRanges.add(new ObjectRange(new CompletableFuture<>(), 500, 800)); @@ -424,7 +423,7 @@ protected void testReadVectoredForSequentialRanges( this.createS3AALClientStreamReader(s3ClientKind, AALInputStreamConfigurationKind)) { S3SeekableInputStream s3SeekableInputStream = - s3AALClientStreamReader.createReadStream(s3Object, OpenStreamInformation.DEFAULT); + s3AALClientStreamReader.createReadStream(s3Object, OpenStreamInformation.ofDefaults()); List objectRanges = new ArrayList<>(); objectRanges.add(new ObjectRange(new CompletableFuture<>(), 2 * ONE_MB, 8 * ONE_MB)); @@ -459,7 +458,7 @@ private void verifyStreamContents( ByteBuffer byteBuffer = objectRange.getByteBuffer().join(); S3SeekableInputStream verificationStream = - s3AALClientStreamReader.createReadStream(s3Object, OpenStreamInformation.DEFAULT); + s3AALClientStreamReader.createReadStream(s3Object, OpenStreamInformation.ofDefaults()); verificationStream.seek(objectRange.getOffset()); byte[] buffer = new byte[objectRange.getLength()]; int readBytes = verificationStream.read(buffer, 0, buffer.length); diff --git a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java index e2ec6c07..50a9daf2 100644 --- a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java +++ b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java @@ -116,7 +116,7 @@ private RetryStrategy createRetryStrategy() { provided = new DefaultRetryStrategyImpl(); } - if (this.physicalIOConfiguration.getBlockReadTimeout() > 0) { + if (this.physicalIOConfiguration.getBlockReadTimeout() > 0 && !provided.isTimeoutSet()) { provided.timeout( physicalIOConfiguration.getBlockReadTimeout(), physicalIOConfiguration.getBlockReadRetryCount()); From 5a192e5011a8220530f6759711bd64044cc06226 Mon Sep 17 00:00:00 2001 From: Fuat Basik Date: Wed, 6 Aug 2025 13:35:25 +0100 Subject: [PATCH 7/8] Address feedback on javadoc and method name --- .../common/telemetry/DefaultTelemetry.java | 1 + .../s3/analyticsaccelerator/common/telemetry/Telemetry.java | 2 ++ .../util/retry/DefaultRetryStrategyImpl.java | 2 +- .../s3/analyticsaccelerator/util/retry/RetryStrategy.java | 2 +- .../util/retry/DefaultRetryStrategyImplTest.java | 4 ++-- .../analyticsaccelerator/io/physical/reader/StreamReader.java | 2 +- 6 files changed, 8 insertions(+), 5 deletions(-) diff --git a/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/DefaultTelemetry.java b/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/DefaultTelemetry.java index 2205c4ae..199d0380 100644 --- a/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/DefaultTelemetry.java +++ b/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/DefaultTelemetry.java @@ -279,6 +279,7 @@ public void measure(@NonNull Metric metric, double value) { * @param operationSupplier operation to record this execution as. * @param operationCode the future to measure the execution of. * @return an instance of {@link T} that returns the same result as the one passed in. + * @throws IOException if the underlying operation threw an IOException */ @Override public T measureJoin( diff --git a/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/Telemetry.java b/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/Telemetry.java index 6bd2491a..3063a949 100644 --- a/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/Telemetry.java +++ b/common/src/main/java/software/amazon/s3/analyticsaccelerator/common/telemetry/Telemetry.java @@ -82,6 +82,7 @@ CompletableFuture measure( * @param operationSupplier operation to record this execution as. * @param operationCode the future to measure the execution of. * @return an instance of {@link T} that returns the same result as the one passed in. + * @throws IOException if the underlying operation threw an IOException */ default T measureJoin( @NonNull TelemetryLevel level, @@ -210,6 +211,7 @@ default CompletableFuture measureStandard( * @param operationSupplier operation to record this execution as. * @param operationCode the future to measure the execution of. * @return an instance of {@link T} that returns the same result as the one passed in. + * @throws IOException if the underlying operation threw an IOException */ default T measureJoinStandard( OperationSupplier operationSupplier, CompletableFuture operationCode) throws IOException { diff --git a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java index 75cefeb8..8c4bc599 100644 --- a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java +++ b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java @@ -172,7 +172,7 @@ private FailsafeExecutor executor() { * @param timeoutDurationMillis Timeout duration for reading from storage * @param retryCount Number of times to retry if Timeout Exceeds */ - public void timeout(long timeoutDurationMillis, int retryCount) { + public void setTimeoutPolicy(long timeoutDurationMillis, int retryCount) { this.timeoutPolicy = Timeout.builder(Duration.ofMillis(timeoutDurationMillis)).withInterrupt().build(); RetryPolicy timeoutRetries = diff --git a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.java b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.java index 142e46fd..2e7ee6b6 100644 --- a/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.java +++ b/common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/RetryStrategy.java @@ -69,7 +69,7 @@ public interface RetryStrategy { * @param durationInMillis Timeout duration for reading from storage * @param retryCount Number of times to retry if Timeout Exceeds */ - void timeout(long durationInMillis, int retryCount); + void setTimeoutPolicy(long durationInMillis, int retryCount); /** * Method to check if timeout of the strategy already set. diff --git a/common/src/test/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImplTest.java b/common/src/test/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImplTest.java index 4f463d7e..b3c80f20 100644 --- a/common/src/test/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImplTest.java +++ b/common/src/test/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImplTest.java @@ -285,7 +285,7 @@ void testOnRetryCallback() throws IOException { @Test void testTimeoutThrows() { DefaultRetryStrategyImpl executor = new DefaultRetryStrategyImpl(); - executor.timeout(1000, 0); + executor.setTimeoutPolicy(1000, 0); AtomicInteger attempt = new AtomicInteger(0); assertThrows( @@ -304,7 +304,7 @@ void testTimeoutThrows() { @Test void testTimeoutWithSuccessAfterRetry() throws IOException { DefaultRetryStrategyImpl executor = new DefaultRetryStrategyImpl(); - executor.timeout(1000, 3); + executor.setTimeoutPolicy(1000, 3); AtomicInteger attempt = new AtomicInteger(0); String expected = "success"; diff --git a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java index 50a9daf2..de82d919 100644 --- a/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java +++ b/input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/reader/StreamReader.java @@ -117,7 +117,7 @@ private RetryStrategy createRetryStrategy() { } if (this.physicalIOConfiguration.getBlockReadTimeout() > 0 && !provided.isTimeoutSet()) { - provided.timeout( + provided.setTimeoutPolicy( physicalIOConfiguration.getBlockReadTimeout(), physicalIOConfiguration.getBlockReadRetryCount()); } From f08c5ea27223a4f416fc9881627848a32071616e Mon Sep 17 00:00:00 2001 From: Fuat Basik Date: Thu, 7 Aug 2025 11:36:03 +0100 Subject: [PATCH 8/8] Add new integrationTest to check override --- .../access/GrayFailureTest.java | 58 ++++++++++++++++++- 1 file changed, 57 insertions(+), 1 deletion(-) diff --git a/input-stream/src/integrationTest/java/software/amazon/s3/analyticsaccelerator/access/GrayFailureTest.java b/input-stream/src/integrationTest/java/software/amazon/s3/analyticsaccelerator/access/GrayFailureTest.java index 2acbf9bc..34d7e65f 100644 --- a/input-stream/src/integrationTest/java/software/amazon/s3/analyticsaccelerator/access/GrayFailureTest.java +++ b/input-stream/src/integrationTest/java/software/amazon/s3/analyticsaccelerator/access/GrayFailureTest.java @@ -15,14 +15,20 @@ */ package software.amazon.s3.analyticsaccelerator.access; -import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.*; import static software.amazon.s3.analyticsaccelerator.util.Constants.ONE_MB; import java.io.IOException; import java.util.ArrayList; import java.util.List; import org.junit.jupiter.api.Test; +import software.amazon.s3.analyticsaccelerator.S3SeekableInputStream; +import software.amazon.s3.analyticsaccelerator.S3SeekableInputStreamFactory; +import software.amazon.s3.analyticsaccelerator.SeekableInputStream; import software.amazon.s3.analyticsaccelerator.util.MetricKey; +import software.amazon.s3.analyticsaccelerator.util.OpenStreamInformation; +import software.amazon.s3.analyticsaccelerator.util.retry.DefaultRetryStrategyImpl; +import software.amazon.s3.analyticsaccelerator.util.retry.RetryStrategy; /** Tests read stream behaviour with untrusted S3ClientKinds on multiple sizes and read patterns */ public class GrayFailureTest extends IntegrationTestBase { @@ -61,4 +67,54 @@ void testFailedReadRecovers() throws IOException { .get(MetricKey.GET_REQUEST_COUNT)); } } + + @Test + void testRetryStrategyOverridesPhysicalIOConfiguration() throws IOException { + RetryStrategy customStrategy = new DefaultRetryStrategyImpl(); + customStrategy.setTimeoutPolicy(100, 0); + OpenStreamInformation openStreamInfo = + software.amazon.s3.analyticsaccelerator.util.OpenStreamInformation.builder() + .retryStrategy(customStrategy) + .build(); + + // PhysicalIOConfiguration on GrayFailure type has 2 retries, we are passing 0 + try (S3AALClientStreamReader s3AALClientStreamReader = + this.createS3AALClientStreamReader( + S3ClientKind.FAULTY_S3_CLIENT, AALInputStreamConfigurationKind.GRAY_FAILURE)) { + + S3SeekableInputStreamFactory factory = + s3AALClientStreamReader.getS3SeekableInputStreamFactory(); + SeekableInputStream defaultStream = + factory.createStream( + S3Object.RANDOM_128MB.getObjectUri( + this.getS3ExecutionContext().getConfiguration().getBaseUri()), + OpenStreamInformation.DEFAULT); + assertEquals(10, defaultStream.read(new byte[10], 0, 10)); + // Assert 2 request are made 1 failure, 1 retry. + assertEquals( + 2, + s3AALClientStreamReader + .getS3SeekableInputStreamFactory() + .getMetrics() + .get(MetricKey.GET_REQUEST_COUNT)); + + // Create another stream to a new object + S3SeekableInputStream overrideStream = + factory.createStream( + S3Object.RANDOM_16MB.getObjectUri( + this.getS3ExecutionContext().getConfiguration().getBaseUri()), + openStreamInfo); + + // Asserting there is timeout + assertThrows(IOException.class, overrideStream::read); + // Assert there are no retries and only 1 additional request is made + // 2 requests should be made by the previous stream and only 1 from this stream. + assertEquals( + 3, + s3AALClientStreamReader + .getS3SeekableInputStreamFactory() + .getMetrics() + .get(MetricKey.GET_REQUEST_COUNT)); + } + } }