Uses content length to determine when to abort the stream. #14329

singhpk234 · 2025-10-14T16:12:20Z

Re: getLength()

While this does not appear to have any impact as such, it does add a lot of noise to the logs

if this is just for logs there can be cases where the getLength() could be a getObjectMetadata call to s3 is it a valid tradeoff ?

That read() causes the abort to fail, as it throws the exception.

The abort() code will swallow the exception right now, but this shows up as

software.amazon.awssdk.core.exception.RetryableException: Data read has a different checksum than expected. Was 0x4dd4fa955ccf4a27e2f635a22948298d, but expected 0x00000000000000000000000000000000. This commonly means that the data was corrupted between the client and service. WARN S3InputStream: An error occurred while aborting the stream

And this happens pretty frequently.

I'm still quite new to iceberg, but from my understanding, by this point we should already have the length and this shouldn't result in a new HEAD call, I expect:

Either its been passed down from the metadata from the avro metadata files

Or the parquet reader has already asked for it to figure out what bytes it needs for the footer.

I don't think this getLength() call will cause an extra HEAD, but could totally be wrong. what do you think?

by this point we should already have the length and this shouldn't result in a new HEAD call

Icebergs metadata.json and manifest list would not have content length but subsequent reads down the iceberg metadata tree i,e mainifest / data / delete files should have it

How about correcting the error message ? if this is too much of a noise

So you're saying removing the LOG.warn("An error occurred while aborting the stream", e);? Not sure if we want that abort() to fail silently.

I'm trying to see if I can figure out what it really means if an abort() fails on a stream that threw an exception. For eg, what's happening right now is:

* We opened a stream from 0-EoF, where EoF is 500MB * After reading 100MB, a retryable exception was thrown * In the resetForRetry(), abort() is called, but it fails because the `read()` fails.

My expectation is that since the underlying stream already failed, it's released that connection and the abort()/close() is a NOOP at this point. but i'm concerned that the abort() failing might mean that connections aren't getting released properly. Will try to confirm.

I am not saying removing, i am saying better rephrasing ?

that the abort() failing might mean that connections aren't getting released properly.Will try to confirm.

agree, it will be a good thing to confirm if failure when aborting the stream leads to connection not being released in the pool, then i think its a valid tradeoff vs introducing a new getLenght() for some objects

abort() was introduced here as an optimization in the first place to reuse connection, please ref: https://github.com/apache/iceberg/pull/7262/files#r1156363240

singhpk234 · 2025-10-14T16:05:01Z

can you make a new constructor for using these arguments ?

This class is protected/private so we don't necessarily need to create a new constructor, but I'm not convinced we actually should do this.

danielcweeks · 2025-10-14T18:16:07Z

Rather than adding the length check here, which adds extra requests when length isn't provided, why can't we just removed the stream.read() check here and quietly swallow errors on the abort path? I'm not sure we really need to validate the stream isn't consumed to abort. Even then, we can just ignore the exception since it's effectively closed at that point.

@danielcweeks if we remove the read(), and then also don't do the remainingInCurrentRequest(), then will abort() the stream, even when we're at EoF. From what I understand, calling abort() removes the connection, and close() allows for the connection reuse.

What is not clear to me is if the same behaviour holds if you're at EoF. What I would like to avoid is removing connections from the pool, when they could be re-used.

I am trying to confirm the above behaviour with the AWS SDK team, but do let me know if you have any advice here.

Since this is quite low level, and depends on the HTTP client implementation I haven't been able to find a clear answer yet. but agree on not making the getLength() call just for this, my understanding was that the length would already be available at this point.

To add some historical context on why the stream.read() != -1 was added: #7262 (comment)

I think we should keep the read there and just remove the noisy warning messages as @bryanck already called out in #7262 (comment). Otherwise the connection will be invalidated and removed from the pool and thus won't be reused

stubz151 · 2025-10-14T14:58:55Z

[nit] remainingInCurrentRequest should be bytesRemainingInCurrentRequest

-Original file line number
+Diff line change
@@ Expand Up / @@ -30,6 +30,7 @@ @@
     import java.util.concurrent.atomic.AtomicInteger;
     import java.util.stream.Stream;
     import javax.net.ssl.SSLException;
+    import org.apache.http.ConnectionClosedException;
     import org.junit.jupiter.api.BeforeEach;
     import org.junit.jupiter.params.ParameterizedTest;
     import org.junit.jupiter.params.provider.Arguments;
@@ Expand All / @@ -52,6 +53,7 @@ @@
     public class TestFlakyS3InputStream extends TestS3InputStream {
       private AtomicInteger resetForRetryCounter;
+      private static final long CONTENT_LENGTH = 1024 * 1024; // An arbitrary content length.
       @BeforeEach
       public void setupTest() {
@@ Expand All / @@ -60,7 +62,7 @@ public void setupTest() { @@
       @Override
       S3InputStream newInputStream(S3Client s3Client, S3URI uri) {
-        return new S3InputStream(s3Client, uri) {
+        return new S3InputStream(s3Client, uri, CONTENT_LENGTH) {
           @Override
           void resetForRetry() throws IOException {
             resetForRetryCounter.incrementAndGet();
@@ Expand Down Expand Up @@
       private static Stream<Arguments> retryableExceptions() {
         return Stream.of(
             Arguments.of(
+                new ConnectionClosedException("connection closed exception"),
                 new SocketTimeoutException("socket timeout exception"),
                 new SSLException("some ssl exception")));
       }
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up / @@ -46,6 +46,8 @@ public class TestS3InputStream { @@
       private final S3Client s3 = MinioUtil.createS3Client(MINIO);
       private final Random random = new Random(1);
+      private static final int CONTENT_LENGTH = 1024 * 1024 * 10; // 10MB
       @BeforeEach
       public void before() {
         createBucket("bucket");
@@ Expand All / @@ -57,13 +59,13 @@ public void testRead() throws Exception { @@
       }
       S3InputStream newInputStream(S3Client s3Client, S3URI uri) {
-        return new S3InputStream(s3Client, uri);
+        return new S3InputStream(s3Client, uri, CONTENT_LENGTH);
       }
       protected void testRead(S3Client s3Client) throws Exception {
         S3URI uri = new S3URI("s3://bucket/path/to/read.dat");
-        int dataSize = 1024 * 1024 * 10;
-        byte[] data = randomData(dataSize);
+        byte[] data = randomData(CONTENT_LENGTH);
         writeS3Data(uri, data);
@@ Expand Down Expand Up / @@ -121,9 +123,9 @@ public void testRangeRead() throws Exception { @@
       protected void testRangeRead(S3Client s3Client) throws Exception {
         S3URI uri = new S3URI("s3://bucket/path/to/range-read.dat");
-        int dataSize = 1024 * 1024 * 10;
-        byte[] expected = randomData(dataSize);
-        byte[] actual = new byte[dataSize];
+        byte[] expected = randomData(CONTENT_LENGTH);
+        byte[] actual = new byte[CONTENT_LENGTH];
         long position;
         int offset;
@@ Expand All @@
           readAndCheckRanges(in, expected, position, actual, offset, length);
           // last 1k
-          position = dataSize - 1024;
-          offset = dataSize - 1024;
+          position = CONTENT_LENGTH - 1024;
+          offset = CONTENT_LENGTH - 1024;
           readAndCheckRanges(in, expected, position, actual, offset, length);
           // middle 2k
-          position = dataSize / 2 - 1024;
-          offset = dataSize / 2 - 1024;
+          position = CONTENT_LENGTH / 2 - 1024;
+          offset = CONTENT_LENGTH / 2 - 1024;
           length = 1024 * 2;
           readAndCheckRanges(in, expected, position, actual, offset, length);
         }
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up / @@ -76,7 +76,7 @@ public SeekableInputStream newStream() { @@
         if (s3FileIOProperties().isS3AnalyticsAcceleratorEnabled()) {
           return AnalyticsAcceleratorUtil.newStream(this);
         }
-        return new S3InputStream(client(), uri(), s3FileIOProperties(), metrics());
+        return new S3InputStream(client(), uri(), s3FileIOProperties(), metrics(), getLength());
       }
       @Override
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up / @@ -28,6 +28,7 @@ @@
     import java.util.Arrays;
     import java.util.List;
     import javax.net.ssl.SSLException;
+    import org.apache.http.ConnectionClosedException;
     import org.apache.iceberg.exceptions.NotFoundException;
     import org.apache.iceberg.io.FileIOMetricsContext;
     import org.apache.iceberg.io.IOUtil;
@@ Expand All @@
       private static final Logger LOG = LoggerFactory.getLogger(S3InputStream.class);
       private static final List<Class<? extends Throwable>> RETRYABLE_EXCEPTIONS =
-          ImmutableList.of(SSLException.class, SocketTimeoutException.class, SocketException.class);
+          ImmutableList.of(
+              SSLException.class,
+              SocketTimeoutException.class,
+              SocketException.class,
+              ConnectionClosedException.class);
       private final StackTraceElement[] createStack;
       private final S3Client s3;
@@ Expand All @@
       private InputStream stream;
       private long pos = 0;
       private long next = 0;
+      private long contentLength = 0;
       private boolean closed = false;
       private final Counter readBytes;
@@ Expand All @@
               .withMaxRetries(3)
               .build();
-      S3InputStream(S3Client s3, S3URI location) {
-        this(s3, location, new S3FileIOProperties(), MetricsContext.nullMetrics());
+      S3InputStream(S3Client s3, S3URI location, long contentLength) {
+        this(s3, location, new S3FileIOProperties(), MetricsContext.nullMetrics(), contentLength);
       }
       S3InputStream(
-          S3Client s3, S3URI location, S3FileIOProperties s3FileIOProperties, MetricsContext metrics) {
+          S3Client s3,
+          S3URI location,
+          S3FileIOProperties s3FileIOProperties,
+          MetricsContext metrics,
+          long contentLength) {
         this.s3 = s3;
         this.location = location;
         this.s3FileIOProperties = s3FileIOProperties;
+        this.contentLength = contentLength;
         this.readBytes = metrics.counter(FileIOMetricsContext.READ_BYTES, Unit.BYTES);
         this.readOperations = metrics.counter(FileIOMetricsContext.READ_OPERATIONS);
@@ Expand Down Expand Up @@
       private void abortStream() {
         try {
-          if (stream instanceof Abortable && stream.read() != -1) {
+          if (stream instanceof Abortable && remainingInCurrentRequest() > 0) {
             ((Abortable) stream).abort();
           }
         } catch (Exception e) {
           LOG.warn("An error occurred while aborting the stream", e);
         }
       }
+      private long remainingInCurrentRequest() {
+        return this.contentLength - this.pos;
+      }
       public void setSkipSize(int skipSize) {
         this.skipSize = skipSize;
       }
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up / @@ -20,33 +20,44 @@ @@
     import static org.mockito.Mockito.any;
     import static org.mockito.Mockito.mock;
+    import static org.mockito.Mockito.never;
+    import static org.mockito.Mockito.spy;
     import static org.mockito.Mockito.verify;
     import static org.mockito.Mockito.when;
+    import java.io.ByteArrayInputStream;
     import java.io.IOException;
-    import java.io.InputStream;
+    import java.util.Arrays;
     import org.junit.jupiter.api.BeforeEach;
     import org.junit.jupiter.api.Test;
     import org.junit.jupiter.api.extension.ExtendWith;
     import org.mockito.Mock;
     import org.mockito.junit.jupiter.MockitoExtension;
     import software.amazon.awssdk.core.sync.ResponseTransformer;
+    import software.amazon.awssdk.http.AbortableInputStream;
     import software.amazon.awssdk.services.s3.S3Client;
     import software.amazon.awssdk.services.s3.model.GetObjectRequest;
     @ExtendWith(MockitoExtension.class)
     public final class TestS3InputStream {
       @Mock private S3Client s3Client;
-      @Mock private InputStream inputStream;
+      private AbortableInputStream inputStream;
       private S3InputStream s3InputStream;
+      private static final int CONTENT_LENGTH = 1024;
       @BeforeEach
       void before() {
+        byte[] writeValue = new byte[CONTENT_LENGTH];
+        Arrays.fill(writeValue, (byte) 1);
+        inputStream =
+            spy(AbortableInputStream.create(new ByteArrayInputStream(new byte[CONTENT_LENGTH])));
         when(s3Client.getObject(any(GetObjectRequest.class), any(ResponseTransformer.class)))
             .thenReturn(inputStream);
-        s3InputStream = new S3InputStream(s3Client, mock());
+        s3InputStream = new S3InputStream(s3Client, mock(), CONTENT_LENGTH);
       }
       @Test
@@ Expand All / @@ -62,4 +73,24 @@ void testReadTailClosesTheStream() throws IOException { @@
         verify(inputStream).close();
       }
+      @Test
+      void testAbortIsCalledAfterPartialRead() throws IOException {
+        byte[] buff = new byte[500];
+        s3InputStream.read(buff);
+        // close after reading partial object, should call abort
+        s3InputStream.close();
+        verify(inputStream).abort();
+      }
+      @Test
+      void testAbortIsCalledAfterFullRead() throws IOException {
+        byte[] buff = new byte[CONTENT_LENGTH];
+        s3InputStream.read(buff);
+        // If we're at EoF, this should not call abort.
+        s3InputStream.close();
+        verify(inputStream, never()).abort();
+      }
     }

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uses content length to determine when to abort the stream. #14329

Uh oh!

Diff view

Diff view

There are no files selected for viewing

singhpk234 Oct 14, 2025

Uh oh!

ahmarsuhail Oct 14, 2025

Uh oh!

singhpk234 Oct 14, 2025

Uh oh!

ahmarsuhail Oct 14, 2025

Uh oh!

singhpk234 Oct 14, 2025 •

edited

Loading

Uh oh!

singhpk234 Oct 14, 2025

Uh oh!

danielcweeks Oct 14, 2025 •

edited

Loading

Uh oh!

danielcweeks Oct 14, 2025

Uh oh!

ahmarsuhail Oct 14, 2025

Uh oh!

ahmarsuhail Oct 14, 2025

Uh oh!

nastra Oct 15, 2025 •

edited

Loading

Uh oh!

stubz151 Oct 14, 2025

Uh oh!

Uh oh!

Uses content length to determine when to abort the stream. #14329

Uh oh!

Uses content length to determine when to abort the stream. #14329

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

singhpk234 Oct 14, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

danielcweeks Oct 14, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nastra Oct 15, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

singhpk234 Oct 14, 2025 •

edited

Loading

danielcweeks Oct 14, 2025 •

edited

Loading

nastra Oct 15, 2025 •

edited

Loading