Repository: apache/iceberg (language: python)

Issue:
javax.net.ssl.SSLException: Connection reset on S3 w/ S3FileIO and Apache HTTP client
### Apache Iceberg version

1.3.1

### Query engine

Spark

### Please describe the bug 🐞

```
24/05/15 15:10:31 ERROR [Executor task launch worker for task 34.0 in stage 14.0 (TID 406)] source.BaseReader: Error reading file(s): s3://bucket/.../file.parquete
org.apache.iceberg.exceptions.RuntimeIOException: javax.net.ssl.SSLException: Connection reset
	at org.apache.iceberg.parquet.ParquetReader$FileIterator.advance(ParquetReader.java:153)
	at org.apache.iceberg.parquet.ParquetReader$FileIterator.next(ParquetReader.java:130)
	at org.apache.iceberg.io.FilterIterator.advance(FilterIterator.java:65)
	at org.apache.iceberg.io.FilterIterator.hasNext(FilterIterator.java:49)
	at org.apache.iceberg.spark.source.BaseReader.next(BaseReader.java:129)
	at org.apache.spark.sql.execution.datasources.v2.PartitionIterator.hasNext(DataSourceRDD.scala:119)
	at org.apache.spark.sql.execution.datasources.v2.MetricsIterator.hasNext(DataSourceRDD.scala:156)
	at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1(DataSourceRDD.scala:63)
	at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1$adapted(DataSourceRDD.scala:63)
	at scala.Option.exists(Option.scala:376)
	at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63)
	at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
	at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage4.processNext(Unknown Source)
	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
	at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:759)
	at org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask$.$anonfun$run$1(WriteToDataSourceV2Exec.scala:412)
	at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1504)
	at org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask$.run(WriteToDataSourceV2Exec.scala:457)
	at org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec.$anonfun$writeWithV2$2(WriteToDataSourceV2Exec.scala:358)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
	at org.apache.spark.scheduler.Task.run(Task.scala:131)
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:506)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1470)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:509)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:750)
Caused by: javax.net.ssl.SSLException: Connection reset
	at sun.security.ssl.Alert.createSSLException(Alert.java:127)
	at sun.security.ssl.TransportContext.fatal(TransportContext.java:324)
	at sun.security.ssl.TransportContext.fatal(TransportContext.java:267)
	at sun.security.ssl.TransportContext.fatal(TransportContext.java:262)
	at sun.security.ssl.SSLTransport.decode(SSLTransport.java:138)
	at sun.security.ssl.SSLSocketImpl.decode(SSLSocketImpl.java:1400)
	at sun.security.ssl.SSLSocketImpl.readApplicationRecord(SSLSocketImpl.java:1368)
	at sun.security.ssl.SSLSocketImpl.access$300(SSLSocketImpl.java:73)
	at sun.security.ssl.SSLSocketImpl$AppInputStream.read(SSLSocketImpl.java:962)
	at software.amazon.awssdk.thirdparty.org.apache.http.impl.io.SessionInputBufferImpl.streamRead(SessionInputBufferImpl.java:137)
	at software.amazon.awssdk.thirdparty.org.apache.http.impl.io.SessionInputBufferImpl.read(SessionInputBufferImpl.java:197)
	at software.amazon.awssdk.thirdparty.org.apache.http.impl.io.ContentLengthInputStream.read(ContentLengthInputStream.java:176)
	at software.amazon.awssdk.thirdparty.org.apache.http.conn.EofSensorInputStream.read(EofSensorInputStream.java:135)
	at java.io.FilterInputStream.read(FilterInputStream.java:133)
	at software.amazon.awssdk.services.s3.checksums.ChecksumValidatingInputStream.read(ChecksumValidatingInputStream.java:112)
	at java.io.FilterInputStream.read(FilterInputStream.java:133)
	at software.amazon.awssdk.core.io.SdkFilterInputStream.read(SdkFilterInputStream.java:66)
	at org.apache.iceberg.aws.s3.S3InputStream.read(S3InputStream.java:109)
	at org.apache.iceberg.shaded.org.apache.parquet.io.DelegatingSeekableInputStream.readFully(DelegatingSeekableInputStream.java:102)
	at org.apache.iceberg.shaded.org.apache.parquet.io.DelegatingSeekableInputStream.readFullyHeapBuffer(DelegatingSeekableInputStream.java:127)
	at org.apache.iceberg.shaded.org.apache.parquet.io.DelegatingSeekableInputStream.readFully(DelegatingSeekableInputStream.java:91)
	at org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader$ConsecutivePartList.readAll(ParquetFileReader.java:1850)
	at org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader.internalReadRowGroup(ParquetFileReader.java:990)
	at org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader.readNextRowGroup(ParquetFileReader.java:940)
	at org.apache.iceberg.parquet.ParquetReader$FileIterator.advance(ParquetReader.java:151)
	... 27 more
	Suppressed: java.net.SocketException: Broken pipe (Write failed)
		at java.net.SocketOutputStream.socketWrite0(Native Method)
		at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:111)
		at java.net.SocketOutputStream.write(SocketOutputStream.java:155)
		at sun.security.ssl.SSLSocketOutputRecord.encodeAlert(SSLSocketOutputRecord.java:81)
		at sun.security.ssl.TransportContext.fatal(TransportContext.java:355)
		... 50 more
Caused by: java.net.SocketException: Connection reset
	at java.net.SocketInputStream.read(SocketInputStream.java:210)
	at java.net.SocketInputStream.read(SocketInputStream.java:141)
	at sun.security.ssl.SSLSocketInputRecord.read(SSLSocketInputRecord.java:464)
	at sun.security.ssl.SSLSocketInputRecord.decodeInputRecord(SSLSocketInputRecord.java:237)
	at sun.security.ssl.SSLSocketInputRecord.decode(SSLSocketInputRecord.java:190)
	at sun.security.ssl.SSLTransport.decode(SSLTransport.java:109)
	... 47 more
```


Discussion / hints:
Also seeing SSLException when accessing pre-signed urls.
```
Caused by: javax.net.ssl.SSLException: Connection reset
	at sun.security.ssl.Alert.createSSLException(Alert.java:127)
	at sun.security.ssl.TransportContext.fatal(TransportContext.java:355)
	at sun.security.ssl.TransportContext.fatal(TransportContext.java:298)
	at sun.security.ssl.TransportContext.fatal(TransportContext.java:293)
	at sun.security.ssl.SSLTransport.decode(SSLTransport.java:142)
	at sun.security.ssl.SSLSocketImpl.decode(SSLSocketImpl.java:1430)
	at sun.security.ssl.SSLSocketImpl.readApplicationRecord(SSLSocketImpl.java:1395)
	at sun.security.ssl.SSLSocketImpl.access$300(SSLSocketImpl.java:73)
	at sun.security.ssl.SSLSocketImpl$AppInputStream.read(SSLSocketImpl.java:982)
	at org.apache.http.impl.io.SessionInputBufferImpl.streamRead(SessionInputBufferImpl.java:137)
	at org.apache.http.impl.io.SessionInputBufferImpl.read(SessionInputBufferImpl.java:197)
	at org.apache.http.impl.io.ContentLengthInputStream.read(ContentLengthInputStream.java:176)
	at org.apache.http.conn.EofSensorInputStream.read(EofSensorInputStream.java:135)
	at io.delta.sharing.client.RandomAccessHttpInputStream.read(RandomAccessHttpInputStream.scala:128)
	at java.io.DataInputStream.read(DataInputStream.java:149)
	at org.apache.parquet.io.DelegatingSeekableInputStream.readFully(DelegatingSeekableInputStream.java:102)
	at org.apache.parquet.io.DelegatingSeekableInputStream.readFullyHeapBuffer(DelegatingSeekableInputStream.java:127)
	at org.apache.parquet.io.DelegatingSeekableInputStream.readFully(DelegatingSeekableInputStream.java:91)
	at org.apache.parquet.hadoop.ParquetFileReader$ConsecutivePartList.readAll(ParquetFileReader.java:1872)
	at org.apache.parquet.hadoop.ParquetFileReader.internalReadRowGroup(ParquetFileReader.java:1020)
	at org.apache.parquet.hadoop.ParquetFileReader.readNextRowGroup(ParquetFileReader.java:969)
	at org.apache.parquet.hadoop.ParquetFileReader.readNextFilteredRowGroup(ParquetFileReader.java:1083)
	at org.apache.parquet.hadoop.InternalParquetRecordReader.checkRead(InternalParquetRecordReader.java:134)
	at org.apache.parquet.hadoop.InternalParquetRecordReader.nextKeyValue(InternalParquetRecordReader.java:235)
	at org.apache.parquet.hadoop.ParquetRecordReader.nextKeyValue(ParquetRecordReader.java:207)
	at org.apache.spark.sql.execution.datasources.RecordReaderIterator.hasNext(RecordReaderIterator.scala:41)
	at org.apache.spark.sql.execution.datasources.RecordReaderIterator$$anon$1.hasNext(RecordReaderIterator.scala:83)
	at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1$$anon$2.getNext(FileScanRDD.scala:609)
	... 40 more
	Suppressed: java.net.SocketException: Broken pipe (Write failed)
		at java.net.SocketOutputStream.socketWrite0(Native Method)
		at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:111)
		at java.net.SocketOutputStream.write(SocketOutputStream.java:155)
		at sun.security.ssl.SSLSocketOutputRecord.encodeAlert(SSLSocketOutputRecord.java:81)
		at sun.security.ssl.TransportContext.fatal(TransportContext.java:386)
		... 66 more
Caused by: java.net.SocketException: Connection reset
	at java.net.SocketInputStream.read(SocketInputStream.java:210)
	at java.net.SocketInputStream.read(SocketInputStream.java:141)
	at sun.security.ssl.SSLSocketInputRecord.read(SSLSocketInputRecord.java:476)
	at sun.security.ssl.SSLSocketInputRecord.readFully(SSLSocketInputRecord.java:459)
	at sun.security.ssl.SSLSocketInputRecord.decodeInputRecord(SSLSocketInputRecord.java:243)
	at sun.security.ssl.SSLSocketInputRecord.decode(SSLSocketInputRecord.java:181)
	at sun.security.ssl.SSLTransport.decode(SSLTransport.java:110)
	... 63 more

Driver stacktrace:
```
Do we have any solution to this issue? I'm getting this issue while reading iceberg tables in glue.
Hi,
This issue/bug has been open for a while now. Do we know when can we expect a fix? Or is there any workaround?

Background:  I'm joining multiple iceberg tables in glue that have 3 merges applied on them. Whenever I do any transform joining these table and write it to non-iceberg glue table, I'm getting SSL connection reset exception. On further checking exception in the executor logs I see  Base Reader exception in reading delete files or data files. 

Error:


```
24/08/12 04:07:15 ERROR BaseReader: Error reading file(s): s3://some-bucket/iceberg_catalog/iceberg_db.db/d_table/data/0yWGCw/region_id=1/marketplace_id=7/asin_bucket=7044/00598-112719-90dfe711-47dc-43e7-af6c-3c5395c527b6-00024.parquet, s3:// some-bucket/iceberg_catalog/iceberg_db.db/d_table/data/0yWGCw/region_id=1/marketplace_id=7/asin_bucket=7044/01086-113207-90dfe711-47dc-43e7-af6c-3c5395c527b6-00025-deletes.parquet, s3:// some-bucket/iceberg_catalog/iceberg_db.db/d_table/data/0yWGCw/region_id=1/marketplace_id=7/asin_bucket=7044/01086-113214-45a89e31-efe0-4110-bdb3-e467a520b1b3-00025-deletes.parquet
org.apache.iceberg.exceptions.RuntimeIOException: javax.net.ssl.SSLException: Connection reset
 at org.apache.iceberg.parquet.VectorizedParquetReader$FileIterator.advance(VectorizedParquetReader.java:165) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.parquet.VectorizedParquetReader$FileIterator.next(VectorizedParquetReader.java:141) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.spark.source.BaseReader.next(BaseReader.java:136) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.spark.sql.execution.datasources.v2.PartitionIterator.hasNext(DataSourceRDD.scala:119) ~[spark-sql_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.sql.execution.datasources.v2.MetricsIterator.hasNext(DataSourceRDD.scala:156) ~[spark-sql_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1(DataSourceRDD.scala:63) ~[spark-sql_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1$adapted(DataSourceRDD.scala:63) ~[spark-sql_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at scala.Option.exists(Option.scala:376) ~[scala-library-2.12.15.jar:?]
 at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63) ~[spark-sql_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37) ~[spark-core_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) ~[scala-library-2.12.15.jar:?]
 at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.columnartorow_nextBatch_0$(Unknown Source) ~[?:?]
 at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source) ~[?:?]
 at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:35) ~[spark-sql_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.hasNext(Unknown Source) ~[?:?]
 at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:968) ~[spark-sql_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) ~[scala-library-2.12.15.jar:?]
 at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:183) ~[spark-core_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) ~[spark-core_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) ~[spark-core_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52) ~[spark-core_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.scheduler.Task.run(Task.scala:138) ~[spark-core_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548) ~[spark-core_2.12-3.3.0-amzn-1.jar:?]
 at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1516) ~[spark-core_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551) ~[spark-core_2.12-3.3.0-amzn-1.jar:?]
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_412]
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_412]
 at java.lang.Thread.run(Thread.java:750) ~[?:1.8.0_412]
Caused by: javax.net.ssl.SSLException: Connection reset
 at sun.security.ssl.Alert.createSSLException(Alert.java:127) ~[?:1.8.0_412]
 at sun.security.ssl.TransportContext.fatal(TransportContext.java:331) ~[?:1.8.0_412]
 at sun.security.ssl.TransportContext.fatal(TransportContext.java:274) ~[?:1.8.0_412]
 at sun.security.ssl.TransportContext.fatal(TransportContext.java:269) ~[?:1.8.0_412]
 at sun.security.ssl.SSLTransport.decode(SSLTransport.java:138) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketImpl.decode(SSLSocketImpl.java:1404) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketImpl.readApplicationRecord(SSLSocketImpl.java:1372) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketImpl.access$300(SSLSocketImpl.java:73) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketImpl$AppInputStream.read(SSLSocketImpl.java:966) ~[?:1.8.0_412]
 at org.apache.iceberg.aws.shaded.org.apache.http.impl.io.SessionInputBufferImpl.streamRead(SessionInputBufferImpl.java:137) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at org.apache.iceberg.aws.shaded.org.apache.http.impl.io.SessionInputBufferImpl.read(SessionInputBufferImpl.java:197) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at org.apache.iceberg.aws.shaded.org.apache.http.impl.io.ContentLengthInputStream.read(ContentLengthInputStream.java:176) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at org.apache.iceberg.aws.shaded.org.apache.http.conn.EofSensorInputStream.read(EofSensorInputStream.java:135) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at java.io.FilterInputStream.read(FilterInputStream.java:133) ~[?:1.8.0_412]
 at software.amazon.awssdk.services.s3.internal.checksums.S3ChecksumValidatingInputStream.read(S3ChecksumValidatingInputStream.java:112) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at java.io.FilterInputStream.read(FilterInputStream.java:133) ~[?:1.8.0_412]
 at software.amazon.awssdk.core.io.SdkFilterInputStream.read(SdkFilterInputStream.java:66) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at software.amazon.awssdk.core.internal.metrics.BytesReadTrackingInputStream.read(BytesReadTrackingInputStream.java:49) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at java.io.FilterInputStream.read(FilterInputStream.java:133) ~[?:1.8.0_412]
 at software.amazon.awssdk.core.io.SdkFilterInputStream.read(SdkFilterInputStream.java:66) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at org.apache.iceberg.aws.s3.S3InputStream.read(S3InputStream.java:109) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.io.DelegatingSeekableInputStream.readFully(DelegatingSeekableInputStream.java:102) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.io.DelegatingSeekableInputStream.readFullyHeapBuffer(DelegatingSeekableInputStream.java:127) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.io.DelegatingSeekableInputStream.readFully(DelegatingSeekableInputStream.java:91) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader$ConsecutivePartList.readAll(ParquetFileReader.java:1850) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader.internalReadRowGroup(ParquetFileReader.java:990) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader.readNextRowGroup(ParquetFileReader.java:940) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.parquet.VectorizedParquetReader$FileIterator.advance(VectorizedParquetReader.java:163) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 ... 27 more
 Suppressed: java.net.SocketException: Broken pipe (Write failed)
 at java.net.SocketOutputStream.socketWrite0(Native Method) ~[?:1.8.0_412]
 at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:111) ~[?:1.8.0_412]
 at java.net.SocketOutputStream.write(SocketOutputStream.java:155) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketOutputRecord.encodeAlert(SSLSocketOutputRecord.java:81) ~[?:1.8.0_412]
 at sun.security.ssl.TransportContext.fatal(TransportContext.java:362) ~[?:1.8.0_412]
 at sun.security.ssl.TransportContext.fatal(TransportContext.java:274) ~[?:1.8.0_412]
 at sun.security.ssl.TransportContext.fatal(TransportContext.java:269) ~[?:1.8.0_412]
 at sun.security.ssl.SSLTransport.decode(SSLTransport.java:138) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketImpl.decode(SSLSocketImpl.java:1404) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketImpl.readApplicationRecord(SSLSocketImpl.java:1372) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketImpl.access$300(SSLSocketImpl.java:73) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketImpl$AppInputStream.read(SSLSocketImpl.java:966) ~[?:1.8.0_412]
 at org.apache.iceberg.aws.shaded.org.apache.http.impl.io.SessionInputBufferImpl.streamRead(SessionInputBufferImpl.java:137) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at org.apache.iceberg.aws.shaded.org.apache.http.impl.io.SessionInputBufferImpl.read(SessionInputBufferImpl.java:197) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at org.apache.iceberg.aws.shaded.org.apache.http.impl.io.ContentLengthInputStream.read(ContentLengthInputStream.java:176) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at org.apache.iceberg.aws.shaded.org.apache.http.conn.EofSensorInputStream.read(EofSensorInputStream.java:135) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at java.io.FilterInputStream.read(FilterInputStream.java:133) ~[?:1.8.0_412]
 at software.amazon.awssdk.services.s3.internal.checksums.S3ChecksumValidatingInputStream.read(S3ChecksumValidatingInputStream.java:112) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at java.io.FilterInputStream.read(FilterInputStream.java:133) ~[?:1.8.0_412]
 at software.amazon.awssdk.core.io.SdkFilterInputStream.read(SdkFilterInputStream.java:66) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at software.amazon.awssdk.core.internal.metrics.BytesReadTrackingInputStream.read(BytesReadTrackingInputStream.java:49) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at java.io.FilterInputStream.read(FilterInputStream.java:133) ~[?:1.8.0_412]
 at software.amazon.awssdk.core.io.SdkFilterInputStream.read(SdkFilterInputStream.java:66) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at org.apache.iceberg.aws.s3.S3InputStream.read(S3InputStream.java:109) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.io.DelegatingSeekableInputStream.readFully(DelegatingSeekableInputStream.java:102) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.io.DelegatingSeekableInputStream.readFullyHeapBuffer(DelegatingSeekableInputStream.java:127) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.io.DelegatingSeekableInputStream.readFully(DelegatingSeekableInputStream.java:91) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader$ConsecutivePartList.readAll(ParquetFileReader.java:1850) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader.internalReadRowGroup(ParquetFileReader.java:990) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader.readNextRowGroup(ParquetFileReader.java:940) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.parquet.VectorizedParquetReader$FileIterator.advance(VectorizedParquetReader.java:163) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.parquet.VectorizedParquetReader$FileIterator.next(VectorizedParquetReader.java:141) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.spark.source.BaseReader.next(BaseReader.java:136) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.spark.sql.execution.datasources.v2.PartitionIterator.hasNext(DataSourceRDD.scala:119) ~[spark-sql_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.sql.execution.datasources.v2.MetricsIterator.hasNext(DataSourceRDD.scala:156) ~[spark-sql_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1(DataSourceRDD.scala:63) ~[spark-sql_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.$anonfun$hasNext$1$adapted(DataSourceRDD.scala:63) ~[spark-sql_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at scala.Option.exists(Option.scala:376) ~[scala-library-2.12.15.jar:?]
 at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:63) ~[spark-sql_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37) ~[spark-core_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) ~[scala-library-2.12.15.jar:?]
 at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.columnartorow_nextBatch_0$(Unknown Source) ~[?:?]
 at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source) ~[?:?]
 at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:35) ~[spark-sql_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.hasNext(Unknown Source) ~[?:?]
 at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:968) ~[spark-sql_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:460) ~[scala-library-2.12.15.jar:?]
 at org.apache.spark.shuffle.sort.UnsafeShuffleWriter.write(UnsafeShuffleWriter.java:183) ~[spark-core_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59) ~[spark-core_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99) ~[spark-core_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52) ~[spark-core_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.scheduler.Task.run(Task.scala:138) ~[spark-core_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548) ~[spark-core_2.12-3.3.0-amzn-1.jar:?]
 at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1516) ~[spark-core_2.12-3.3.0-amzn-1.jar:3.3.0-amzn-1]
 at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551) ~[spark-core_2.12-3.3.0-amzn-1.jar:?]
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_412]
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_412]
 at java.lang.Thread.run(Thread.java:750) ~[?:1.8.0_412]
Caused by: java.net.SocketException: Connection reset
 at java.net.SocketInputStream.read(SocketInputStream.java:210) ~[?:1.8.0_412]
 at java.net.SocketInputStream.read(SocketInputStream.java:141) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketInputRecord.read(SSLSocketInputRecord.java:464) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketInputRecord.decodeInputRecord(SSLSocketInputRecord.java:237) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketInputRecord.decode(SSLSocketInputRecord.java:190) ~[?:1.8.0_412]
 at sun.security.ssl.SSLTransport.decode(SSLTransport.java:109) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketImpl.decode(SSLSocketImpl.java:1404) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketImpl.readApplicationRecord(SSLSocketImpl.java:1372) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketImpl.access$300(SSLSocketImpl.java:73) ~[?:1.8.0_412]
 at sun.security.ssl.SSLSocketImpl$AppInputStream.read(SSLSocketImpl.java:966) ~[?:1.8.0_412]
 at org.apache.iceberg.aws.shaded.org.apache.http.impl.io.SessionInputBufferImpl.streamRead(SessionInputBufferImpl.java:137) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at org.apache.iceberg.aws.shaded.org.apache.http.impl.io.SessionInputBufferImpl.read(SessionInputBufferImpl.java:197) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at org.apache.iceberg.aws.shaded.org.apache.http.impl.io.ContentLengthInputStream.read(ContentLengthInputStream.java:176) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at org.apache.iceberg.aws.shaded.org.apache.http.conn.EofSensorInputStream.read(EofSensorInputStream.java:135) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at java.io.FilterInputStream.read(FilterInputStream.java:133) ~[?:1.8.0_412]
 at software.amazon.awssdk.services.s3.internal.checksums.S3ChecksumValidatingInputStream.read(S3ChecksumValidatingInputStream.java:112) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at java.io.FilterInputStream.read(FilterInputStream.java:133) ~[?:1.8.0_412]
 at software.amazon.awssdk.core.io.SdkFilterInputStream.read(SdkFilterInputStream.java:66) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at software.amazon.awssdk.core.internal.metrics.BytesReadTrackingInputStream.read(BytesReadTrackingInputStream.java:49) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at java.io.FilterInputStream.read(FilterInputStream.java:133) ~[?:1.8.0_412]
 at software.amazon.awssdk.core.io.SdkFilterInputStream.read(SdkFilterInputStream.java:66) ~[iceberg-aws-bundle-1.5.0.jar:?]
 at org.apache.iceberg.aws.s3.S3InputStream.read(S3InputStream.java:109) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.io.DelegatingSeekableInputStream.readFully(DelegatingSeekableInputStream.java:102) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.io.DelegatingSeekableInputStream.readFullyHeapBuffer(DelegatingSeekableInputStream.java:127) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.io.DelegatingSeekableInputStream.readFully(DelegatingSeekableInputStream.java:91) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader$ConsecutivePartList.readAll(ParquetFileReader.java:1850) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader.internalReadRowGroup(ParquetFileReader.java:990) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetFileReader.readNextRowGroup(ParquetFileReader.java:940) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 at org.apache.iceberg.parquet.VectorizedParquetReader$FileIterator.advance(VectorizedParquetReader.java:163) ~[iceberg-spark-runtime-3.3_2.12-1.5.0.jar:?]
 ... 27 more
```

I have tried using updated version of iceberg i.e 1.6.0 as well, but getting same error.
one of those stack traces is from deltaio, so nothing to do with iceberg

both of them are caused by the AWS sdk itself not retrying, or retrying but not enough times for the problem to recover. There's also http connection pooling at play here too: there's no point the library trying to repeat the request if it keeps returning the failed stream to the pool for it to be picked up again. 

Some suggestions
* look up the sdk docs on system properties `aws.retryMode` and `aws.maxAttempts` and see if setting things there help
* if possible, switch to the Hadoop s3a connector which does a lot of work recovering from connectivity and other network errors
Thanks for the suggestion, I will try them out. However, there is a pull request open already. Also, @danielcweeks mentioned here -> https://github.com/apache/iceberg/pull/4912 about a neat implementation for this issue. Are there any plans on iceberg side to handle it? I'm asking because it's very common issue asked multiple times on various platforms without a solution.
I can't speak for the S3FileIO developers; S3AFS is where I code and while there's a lot of work there for recovery [here](https://github.com/apache/hadoop/blob/trunk/hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3ARetryPolicy.java) and elsewhere, we are all still  finding obscure recovery failures one by one, such as how the AWS SDK doesn't recovery properly [if a multipart part upload fails with a 500](https://github.com/apache/hadoop/pull/6938).

1. If you want to use the S3FileIO: try those options.
2. If you want an S3 client which has fixes for all the failures we've hit: S3A is your friend.
3. Or you take up the PR, do your own iceberg release with it and let everyone know if it does/doesn't work. Real world pre-release testing is the way to do this


Fix patch:
diff --git a/aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java b/aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java
index f442a0f04a1c..74e602a27378 100644
--- a/aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java
+++ b/aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java
@@ -18,9 +18,15 @@
  */
 package org.apache.iceberg.aws.s3;
 
+import dev.failsafe.Failsafe;
+import dev.failsafe.FailsafeException;
+import dev.failsafe.RetryPolicy;
 import java.io.IOException;
 import java.io.InputStream;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
 import java.util.Arrays;
+import javax.net.ssl.SSLException;
 import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.io.FileIOMetricsContext;
 import org.apache.iceberg.io.IOUtil;
@@ -31,6 +37,7 @@
 import org.apache.iceberg.metrics.MetricsContext.Unit;
 import org.apache.iceberg.relocated.com.google.common.base.Joiner;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.io.ByteStreams;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -57,6 +64,14 @@ class S3InputStream extends SeekableInputStream implements RangeReadable {
   private final Counter readOperations;
 
   private int skipSize = 1024 * 1024;
+  private RetryPolicy<Object> retryPolicy =
+      RetryPolicy.builder()
+          .handle(
+              ImmutableList.of(
+                  SSLException.class, SocketTimeoutException.class, SocketException.class))
+          .onFailure(failure -> openStream(true))
+          .withMaxRetries(3)
+          .build();
 
   S3InputStream(S3Client s3, S3URI location) {
     this(s3, location, new S3FileIOProperties(), MetricsContext.nullMetrics());
@@ -92,13 +107,21 @@ public void seek(long newPos) {
   public int read() throws IOException {
     Preconditions.checkState(!closed, "Cannot read: already closed");
     positionStream();
+    try {
+      int bytesRead = Failsafe.with(retryPolicy).get(() -> stream.read());
+      pos += 1;
+      next += 1;
+      readBytes.increment();
+      readOperations.increment();
+
+      return bytesRead;
+    } catch (FailsafeException ex) {
+      if (ex.getCause() instanceof IOException) {
+        throw (IOException) ex.getCause();
+      }
 
-    pos += 1;
-    next += 1;
-    readBytes.increment();
-    readOperations.increment();
-
-    return stream.read();
+      throw ex;
+    }
   }
 
   @Override
@@ -106,13 +129,21 @@ public int read(byte[] b, int off, int len) throws IOException {
     Preconditions.checkState(!closed, "Cannot read: already closed");
     positionStream();
 
-    int bytesRead = stream.read(b, off, len);
-    pos += bytesRead;
-    next += bytesRead;
-    readBytes.increment(bytesRead);
-    readOperations.increment();
+    try {
+      int bytesRead = Failsafe.with(retryPolicy).get(() -> stream.read(b, off, len));
+      pos += bytesRead;
+      next += bytesRead;
+      readBytes.increment(bytesRead);
+      readOperations.increment();
+
+      return bytesRead;
+    } catch (FailsafeException ex) {
+      if (ex.getCause() instanceof IOException) {
+        throw (IOException) ex.getCause();
+      }
 
-    return bytesRead;
+      throw ex;
+    }
   }
 
   @Override
@@ -146,7 +177,7 @@ private InputStream readRange(String range) {
   public void close() throws IOException {
     super.close();
     closed = true;
-    closeStream();
+    closeStream(false);
   }
 
   private void positionStream() throws IOException {
@@ -178,6 +209,10 @@ private void positionStream() throws IOException {
   }
 
   private void openStream() throws IOException {
+    openStream(false);
+  }
+
+  private void openStream(boolean closeQuietly) throws IOException {
     GetObjectRequest.Builder requestBuilder =
         GetObjectRequest.builder()
             .bucket(location.bucket())
@@ -186,7 +221,7 @@ private void openStream() throws IOException {
 
     S3RequestUtil.configureEncryption(s3FileIOProperties, requestBuilder);
 
-    closeStream();
+    closeStream(closeQuietly);
 
     try {
       stream = s3.getObject(requestBuilder.build(), ResponseTransformer.toInputStream());
@@ -195,7 +230,7 @@ private void openStream() throws IOException {
     }
   }
 
-  private void closeStream() throws IOException {
+  private void closeStream(boolean closeQuietly) throws IOException {
     if (stream != null) {
       // if we aren't at the end of the stream, and the stream is abortable, then
       // call abort() so we don't read the remaining data with the Apache HTTP client
@@ -203,6 +238,12 @@ private void closeStream() throws IOException {
       try {
         stream.close();
       } catch (IOException e) {
+        if (closeQuietly) {
+          stream = null;
+          LOG.warn("An error occurred while closing the stream", e);
+          return;
+        }
+
         // the Apache HTTP client will throw a ConnectionClosedException
         // when closing an aborted stream, which is expected
         if (!e.getClass().getSimpleName().equals("ConnectionClosedException")) {
diff --git a/build.gradle b/build.gradle
index fab949bf06a6..bb5515605152 100644
--- a/build.gradle
+++ b/build.gradle
@@ -347,6 +347,7 @@ project(':iceberg-core') {
     implementation libs.jackson.core
     implementation libs.jackson.databind
     implementation libs.caffeine
+    implementation libs.failsafe
     implementation libs.roaringbitmap
     compileOnly(libs.hadoop2.client) {
       exclude group: 'org.apache.avro', module: 'avro'
@@ -462,6 +463,7 @@ project(':iceberg-aws') {
     annotationProcessor libs.immutables.value
     compileOnly libs.immutables.value
     implementation libs.caffeine
+    implementation libs.failsafe
     implementation platform(libs.jackson.bom)
     implementation libs.jackson.core
     implementation libs.jackson.databind
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 55edaf281489..5fc549689078 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -39,6 +39,7 @@ delta-standalone = "3.2.0"
 delta-spark = "3.2.0"
 esotericsoftware-kryo = "4.0.3"
 errorprone-annotations = "2.31.0"
+failsafe = "3.3.2"
 findbugs-jsr305 = "3.0.2"
 flink118 =  { strictly = "1.18.1"}
 flink119 =  { strictly = "1.19.0"}
@@ -109,6 +110,7 @@ calcite-druid = { module = "org.apache.calcite:calcite-druid", version.ref = "ca
 datasketches = { module = "org.apache.datasketches:datasketches-java", version.ref = "datasketches" }
 delta-standalone = { module = "io.delta:delta-standalone_2.12", version.ref = "delta-standalone" }
 errorprone-annotations = { module = "com.google.errorprone:error_prone_annotations", version.ref = "errorprone-annotations" }
+failsafe = { module = "dev.failsafe:failsafe", version.ref = "failsafe"}
 findbugs-jsr305 = { module = "com.google.code.findbugs:jsr305", version.ref = "findbugs-jsr305" }
 flink118-avro = { module = "org.apache.flink:flink-avro", version.ref = "flink118" }
 flink118-connector-base = { module = "org.apache.flink:flink-connector-base", version.ref = "flink118" }


Test patch:
diff --git a/aws/src/test/java/org/apache/iceberg/aws/s3/TestFlakyS3InputStream.java b/aws/src/test/java/org/apache/iceberg/aws/s3/TestFlakyS3InputStream.java
new file mode 100644
index 000000000000..08d14512cdc7
--- /dev/null
+++ b/aws/src/test/java/org/apache/iceberg/aws/s3/TestFlakyS3InputStream.java
@@ -0,0 +1,206 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.aws.s3;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.SocketTimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
+import javax.net.ssl.SSLException;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import software.amazon.awssdk.awscore.exception.AwsServiceException;
+import software.amazon.awssdk.core.ResponseInputStream;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.core.sync.ResponseTransformer;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+import software.amazon.awssdk.services.s3.model.CreateBucketResponse;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
+import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import software.amazon.awssdk.services.s3.model.PutObjectResponse;
+
+public class TestFlakyS3InputStream extends TestS3InputStream {
+
+  @ParameterizedTest
+  @MethodSource("retryableExceptions")
+  public void testReadWithFlakyStreamRetrySucceed(IOException exception) throws Exception {
+    testRead(flakyStreamClient(new AtomicInteger(3), exception));
+  }
+
+  @ParameterizedTest
+  @MethodSource("retryableExceptions")
+  public void testReadWithFlakyStreamExhaustedRetries(IOException exception) {
+    assertThatThrownBy(() -> testRead(flakyStreamClient(new AtomicInteger(5), exception)))
+        .isInstanceOf(exception.getClass())
+        .hasMessage(exception.getMessage());
+  }
+
+  @ParameterizedTest
+  @MethodSource("nonRetryableExceptions")
+  public void testReadWithFlakyStreamNonRetryableException(IOException exception) {
+    assertThatThrownBy(() -> testRead(flakyStreamClient(new AtomicInteger(3), exception)))
+        .isInstanceOf(exception.getClass())
+        .hasMessage(exception.getMessage());
+  }
+
+  @ParameterizedTest
+  @MethodSource("retryableExceptions")
+  public void testSeekWithFlakyStreamRetrySucceed(IOException exception) throws Exception {
+    testSeek(flakyStreamClient(new AtomicInteger(3), exception));
+  }
+
+  @ParameterizedTest
+  @MethodSource("retryableExceptions")
+  public void testSeekWithFlakyStreamExhaustedRetries(IOException exception) {
+    assertThatThrownBy(() -> testSeek(flakyStreamClient(new AtomicInteger(5), exception)))
+        .isInstanceOf(exception.getClass())
+        .hasMessage(exception.getMessage());
+  }
+
+  @ParameterizedTest
+  @MethodSource("nonRetryableExceptions")
+  public void testSeekWithFlakyStreamNonRetryableException(IOException exception) {
+    assertThatThrownBy(() -> testSeek(flakyStreamClient(new AtomicInteger(3), exception)))
+        .isInstanceOf(exception.getClass())
+        .hasMessage(exception.getMessage());
+  }
+
+  private static Stream<Arguments> retryableExceptions() {
+    return Stream.of(
+        Arguments.of(
+            new SocketTimeoutException("socket timeout exception"),
+            new SSLException("some ssl exception")));
+  }
+
+  private static Stream<Arguments> nonRetryableExceptions() {
+    return Stream.of(Arguments.of(new IOException("some generic non-retryable IO exception")));
+  }
+
+  private S3ClientWrapper flakyStreamClient(AtomicInteger counter, IOException failure) {
+    S3ClientWrapper flakyClient = spy(new S3ClientWrapper(s3Client()));
+    doAnswer(invocation -> new FlakyInputStream(invocation.callRealMethod(), counter, failure))
+        .when(flakyClient)
+        .getObject(any(GetObjectRequest.class), any(ResponseTransformer.class));
+    return flakyClient;
+  }
+
+  /** Wrapper for S3 client, used to mock the final class DefaultS3Client */
+  public static class S3ClientWrapper implements S3Client {
+
+    private final S3Client delegate;
+
+    public S3ClientWrapper(S3Client delegate) {
+      this.delegate = delegate;
+    }
+
+    @Override
+    public String serviceName() {
+      return delegate.serviceName();
+    }
+
+    @Override
+    public void close() {
+      delegate.close();
+    }
+
+    @Override
+    public <ReturnT> ReturnT getObject(
+        GetObjectRequest getObjectRequest,
+        ResponseTransformer<GetObjectResponse, ReturnT> responseTransformer)
+        throws AwsServiceException, SdkClientException {
+      return delegate.getObject(getObjectRequest, responseTransformer);
+    }
+
+    @Override
+    public HeadObjectResponse headObject(HeadObjectRequest headObjectRequest)
+        throws AwsServiceException, SdkClientException {
+      return delegate.headObject(headObjectRequest);
+    }
+
+    @Override
+    public PutObjectResponse putObject(PutObjectRequest putObjectRequest, RequestBody requestBody)
+        throws AwsServiceException, SdkClientException {
+      return delegate.putObject(putObjectRequest, requestBody);
+    }
+
+    @Override
+    public CreateBucketResponse createBucket(CreateBucketRequest createBucketRequest)
+        throws AwsServiceException, SdkClientException {
+      return delegate.createBucket(createBucketRequest);
+    }
+  }
+
+  static class FlakyInputStream extends InputStream {
+    private final ResponseInputStream<GetObjectResponse> delegate;
+    private final AtomicInteger counter;
+    private final int round;
+    private final IOException exception;
+
+    FlakyInputStream(Object invocationResponse, AtomicInteger counter, IOException exception) {
+      this.delegate = (ResponseInputStream<GetObjectResponse>) invocationResponse;
+      this.counter = counter;
+      this.round = counter.get();
+      this.exception = exception;
+    }
+
+    private void checkCounter() throws IOException {
+      // for every round of n invocations, only the last call succeeds
+      if (counter.decrementAndGet() == 0) {
+        counter.set(round);
+      } else {
+        throw exception;
+      }
+    }
+
+    @Override
+    public int read() throws IOException {
+      checkCounter();
+      return delegate.read();
+    }
+
+    @Override
+    public int read(byte[] b) throws IOException {
+      checkCounter();
+      return delegate.read(b);
+    }
+
+    @Override
+    public int read(byte[] b, int off, int len) throws IOException {
+      checkCounter();
+      return delegate.read(b, off, len);
+    }
+
+    @Override
+    public void close() throws IOException {
+      delegate.close();
+    }
+  }
+}
diff --git a/aws/src/test/java/org/apache/iceberg/aws/s3/TestS3InputStream.java b/aws/src/test/java/org/apache/iceberg/aws/s3/TestS3InputStream.java
index ed71e259a26c..0e3f8b2136a6 100644
--- a/aws/src/test/java/org/apache/iceberg/aws/s3/TestS3InputStream.java
+++ b/aws/src/test/java/org/apache/iceberg/aws/s3/TestS3InputStream.java
@@ -54,16 +54,18 @@ public void before() {
 
   @Test
   public void testRead() throws Exception {
+    testRead(s3);
+  }
+
+  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);
 
     writeS3Data(uri, data);
 
-    try (SeekableInputStream in = new S3InputStream(s3, uri)) {
+    try (SeekableInputStream in = new S3InputStream(s3Client, uri)) {
       int readSize = 1024;
-      byte[] actual = new byte[readSize];
-
       readAndCheck(in, in.getPos(), readSize, data, false);
       readAndCheck(in, in.getPos(), readSize, data, true);
 
@@ -111,6 +113,10 @@ private void readAndCheck(
 
   @Test
   public void testRangeRead() throws Exception {
+    testRangeRead(s3);
+  }
+
+  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);
@@ -122,7 +128,7 @@ public void testRangeRead() throws Exception {
 
     writeS3Data(uri, expected);
 
-    try (RangeReadable in = new S3InputStream(s3, uri)) {
+    try (RangeReadable in = new S3InputStream(s3Client, uri)) {
       // first 1k
       position = 0;
       offset = 0;
@@ -163,12 +169,16 @@ public void testClose() throws Exception {
 
   @Test
   public void testSeek() throws Exception {
+    testSeek(s3);
+  }
+
+  protected void testSeek(S3Client s3Client) throws Exception {
     S3URI uri = new S3URI("s3://bucket/path/to/seek.dat");
     byte[] expected = randomData(1024 * 1024);
 
     writeS3Data(uri, expected);
 
-    try (SeekableInputStream in = new S3InputStream(s3, uri)) {
+    try (SeekableInputStream in = new S3InputStream(s3Client, uri)) {
       in.seek(expected.length / 2);
       byte[] actual = new byte[expected.length / 2];
       IOUtil.readFully(in, actual, 0, expected.length / 2);
@@ -200,4 +210,8 @@ private void createBucket(String bucketName) {
       // don't do anything
     }
   }
+
+  protected S3Client s3Client() {
+    return s3;
+  }
 }