Introduce UnrecoverableIOException for terminal cases

Convert terminal s3 filesystem exceptions to extend UnrecoverableIOException

Convert SDKException to UnrecoverableIOException
if they are not supposed to be retryable

lib/trino-filesystem-s3/src/main/java/io/trino/filesystem/s3/S3FileSystem.java

@@ -20,6 +20,7 @@
 import io.trino.filesystem.TrinoFileSystem;
 import io.trino.filesystem.TrinoInputFile;
 import io.trino.filesystem.TrinoOutputFile;
+import io.trino.filesystem.UnrecoverableIOException;
 import software.amazon.awssdk.core.exception.SdkException;
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.model.CommonPrefix;
@@ -100,7 +101,7 @@ public void deleteFile(Location location)
             client.deleteObject(request);
         }
         catch (SdkException e) {
-            throw new IOException("Failed to delete file: " + location, e);
+            throw asIOException("Failed to delete file: " + location, e);
         }
     }
 
@@ -158,7 +159,7 @@ private void deleteObjects(Collection<Location> locations)
                     }
                 }
                 catch (SdkException e) {
-                    throw new IOException("Error while batch deleting files", e);
+                    throw asIOException("Error while batch deleting files", e);
                 }
             }
         }
@@ -172,7 +173,7 @@ private void deleteObjects(Collection<Location> locations)
     public void renameFile(Location source, Location target)
             throws IOException
     {
-        throw new IOException("S3 does not support renames");
+        throw new UnrecoverableIOException("S3 does not support renames");
     }
 
     @Override
@@ -206,7 +207,7 @@ private FileIterator listObjects(Location location, boolean includeDirectoryObje
             return new S3FileIterator(s3Location, s3ObjectStream.iterator());
         }
         catch (SdkException e) {
-            throw new IOException("Failed to list location: " + location, e);
+            throw asIOException("Failed to list location: " + location, e);
         }
     }
 
@@ -232,7 +233,7 @@ public void createDirectory(Location location)
     public void renameDirectory(Location source, Location target)
             throws IOException
     {
-        throw new IOException("S3 does not support directory renames");
+        throw new UnrecoverableIOException("S3 does not support directory renames");
     }
 
     @Override
@@ -262,7 +263,7 @@ public Set<Location> listDirectories(Location location)
                     .collect(toImmutableSet());
         }
         catch (SdkException e) {
-            throw new IOException("Failed to list location: " + location, e);
+            throw asIOException("Failed to list location: " + location, e);
         }
     }
 
@@ -274,6 +275,13 @@ public Optional<Location> createTemporaryDirectory(Location targetPath, String t
         return Optional.empty();
     }
 
+    private IOException asIOException(String message, SdkException exception)
+    {
+        return exception.retryable() ?
+                new IOException(message, exception) :
+                new UnrecoverableIOException(message, exception);
+    }
+
     @SuppressWarnings("ResultOfObjectAllocationIgnored")
     private static void validateS3Location(Location location)
     {

lib/trino-filesystem/src/main/java/io/trino/filesystem/UnrecoverableIOException.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed 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 io.trino.filesystem;
+
+import java.io.IOException;
+
+/**
+ * This exception is for stopping retries for FileSystem calls that shouldn't be retried.
+ */
+public class UnrecoverableIOException
+        extends IOException
+{
+    public UnrecoverableIOException(String message, Throwable cause)
+    {
+        super(message, cause);
+    }
+
+    public UnrecoverableIOException(String message)
+    {
+        super(message);
+    }
+}

lib/trino-hdfs/src/main/java/io/trino/hdfs/s3/TrinoS3FileSystem.java

@@ -90,6 +90,7 @@
 import io.airlift.units.Duration;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.instrumentation.awssdk.v1_11.AwsSdkTelemetry;
+import io.trino.filesystem.UnrecoverableIOException;
 import io.trino.hdfs.FSDataInputStreamTail;
 import io.trino.hdfs.FileSystemWithBatchDelete;
 import io.trino.hdfs.MemoryAwareFileSystem;
@@ -934,17 +935,12 @@ private static boolean isHadoopFolderMarker(S3ObjectSummary object)
         return object.getKey().endsWith("_$folder$");
     }
 
-    /**
-     * This exception is for stopping retries for S3 calls that shouldn't be retried.
-     * For example, "Caused by: com.amazonaws.services.s3.model.AmazonS3Exception: Forbidden (Service: Amazon S3; Status Code: 403 ..."
-     */
-    public static class UnrecoverableS3OperationException
-            extends IOException
+    static class UnrecoverableS3OperationException
+            extends UnrecoverableIOException
     {
         public UnrecoverableS3OperationException(String bucket, String key, Throwable cause)
         {
-            // append bucket and key to the message
-            super(format("%s (Bucket: %s, Key: %s)", cause, bucket, key));
+            super(format("%s (Bucket: %s, Key: %s)", cause, bucket, key), cause);
         }
     }