Core, AWS: Fix REST catalog hanging on permission errors

aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java

@@ -28,6 +28,7 @@
 import java.util.Arrays;
 import java.util.List;
 import javax.net.ssl.SSLException;
+import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.io.FileIOMetricsContext;
 import org.apache.iceberg.io.IOUtil;
@@ -45,9 +46,11 @@
 import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.core.sync.ResponseTransformer;
 import software.amazon.awssdk.http.Abortable;
+import software.amazon.awssdk.http.HttpStatusCode;
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
 import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
+import software.amazon.awssdk.services.s3.model.S3Exception;
 
 class S3InputStream extends SeekableInputStream implements RangeReadable {
   private static final Logger LOG = LoggerFactory.getLogger(S3InputStream.class);
@@ -195,7 +198,22 @@ private InputStream readRange(String range) {
 
     S3RequestUtil.configureEncryption(s3FileIOProperties, requestBuilder);
 
-    return s3.getObject(requestBuilder.build(), ResponseTransformer.toInputStream());
+    return openObject(requestBuilder.build());
+  }
+
+  private InputStream openObject(GetObjectRequest request) {
+    try {
+      return s3.getObject(request, ResponseTransformer.toInputStream());
+    } catch (NoSuchKeyException e) {
+      throw new NotFoundException(e, "Location does not exist: %s", location);
+    } catch (S3Exception e) {
+      if (e.statusCode() == HttpStatusCode.FORBIDDEN) {
+        throw new ForbiddenException(
+            e, "Failed to open input stream for location: %s: %s", location, e.getMessage());
+      }
+
+      throw e;
+    }
   }
 
   @Override
@@ -248,11 +266,7 @@ private void openStream(boolean closeQuietly) throws IOException {
 
     closeStream(closeQuietly);
 
-    try {
-      stream = s3.getObject(requestBuilder.build(), ResponseTransformer.toInputStream());
-    } catch (NoSuchKeyException e) {
-      throw new NotFoundException(e, "Location does not exist: %s", location);
-    }
+    stream = openObject(requestBuilder.build());
   }
 
   @VisibleForTesting

aws/src/test/java/org/apache/iceberg/aws/s3/TestS3InputStream.java

@@ -18,48 +18,72 @@
  */
 package org.apache.iceberg.aws.s3;
 
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.io.IOException;
 import java.io.InputStream;
+import org.apache.iceberg.exceptions.ForbiddenException;
 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.HttpStatusCode;
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.S3Exception;
 
 @ExtendWith(MockitoExtension.class)
 public final class TestS3InputStream {
 
+  private static final String SERVER_ACCESS_DENIED_MESSAGE = "server-side access denied detail";
+
   @Mock private S3Client s3Client;
   @Mock private InputStream inputStream;
 
   private S3InputStream s3InputStream;
 
   @BeforeEach
   void before() {
-    when(s3Client.getObject(any(GetObjectRequest.class), any(ResponseTransformer.class)))
-        .thenReturn(inputStream);
     s3InputStream = new S3InputStream(s3Client, mock());
   }
 
   @Test
   void testReadFullyClosesTheStream() throws IOException {
+    when(s3Client.getObject(any(GetObjectRequest.class), any(ResponseTransformer.class)))
+        .thenReturn(inputStream);
+
     s3InputStream.readFully(0, new byte[0]);
 
     verify(inputStream).close();
   }
 
   @Test
   void testReadTailClosesTheStream() throws IOException {
+    when(s3Client.getObject(any(GetObjectRequest.class), any(ResponseTransformer.class)))
+        .thenReturn(inputStream);
+
     s3InputStream.readTail(new byte[0], 0, 0);
 
     verify(inputStream).close();
   }
+
+  @Test
+  void testAccessDeniedTranslatedToForbiddenException() {
+    when(s3Client.getObject(any(GetObjectRequest.class), any(ResponseTransformer.class)))
+        .thenThrow(
+            S3Exception.builder()
+                .statusCode(HttpStatusCode.FORBIDDEN)
+                .message(SERVER_ACCESS_DENIED_MESSAGE)
+                .build());
+
+    assertThatThrownBy(() -> s3InputStream.readFully(0, new byte[0]))
+        .isInstanceOf(ForbiddenException.class)
+        .hasMessageContaining(SERVER_ACCESS_DENIED_MESSAGE);
+  }
 }

core/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java

@@ -27,6 +27,7 @@
 import org.apache.iceberg.encryption.EncryptionManager;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.io.FileIO;
@@ -195,7 +196,9 @@ protected void refreshFromMetadataLocation(
           .retry(numRetries)
           .exponentialBackoff(100, 5000, 600000, 4.0 /* 100, 400, 1600, ... */)
           .throwFailureWhenFinished()
-          .stopRetryOn(NotFoundException.class) // overridden if shouldRetry is non-null
+          .stopRetryOn(
+              NotFoundException.class,
+              ForbiddenException.class) // overridden if shouldRetry is non-null
           .shouldRetryTest(shouldRetry)
           .run(metadataLocation -> newMetadata.set(metadataLoader.apply(metadataLocation)));
 

core/src/main/java/org/apache/iceberg/hadoop/HadoopInputFile.java

@@ -27,8 +27,10 @@
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.security.AccessControlException;
 import org.apache.iceberg.encryption.NativeFileCryptoParameters;
 import org.apache.iceberg.encryption.NativelyEncryptedFile;
+import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.exceptions.RuntimeIOException;
 import org.apache.iceberg.io.InputFile;
@@ -183,6 +185,9 @@ public SeekableInputStream newStream() {
       return HadoopStreams.wrap(fs.open(path));
     } catch (FileNotFoundException e) {
       throw new NotFoundException(e, "Failed to open input stream for file: %s", path);
+    } catch (AccessControlException e) {
+      throw new ForbiddenException(
+          e, "Failed to open input stream for file: %s: %s", path, e.getMessage());
     } catch (IOException e) {
       throw new RuntimeIOException(e, "Failed to open input stream for file: %s", path);
     }

core/src/main/java/org/apache/iceberg/view/BaseViewOperations.java

@@ -27,6 +27,7 @@
 import org.apache.iceberg.TableMetadataParser;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.exceptions.NoSuchViewException;
 import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.io.FileIO;
@@ -200,7 +201,9 @@ protected void refreshFromMetadataLocation(
           .retry(numRetries)
           .exponentialBackoff(100, 5000, 600000, 4.0 /* 100, 400, 1600, ... */)
           .throwFailureWhenFinished()
-          .stopRetryOn(NotFoundException.class) // overridden if shouldRetry is non-null
+          .stopRetryOn(
+              NotFoundException.class,
+              ForbiddenException.class) // overridden if shouldRetry is non-null
           .shouldRetryTest(shouldRetry)
           .run(metadataLocation -> newMetadata.set(metadataLoader.apply(metadataLocation)));
 

core/src/test/java/org/apache/iceberg/TestBaseMetastoreTableOperationsRefresh.java

@@ -0,0 +1,106 @@
+/*
+ * 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;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.iceberg.exceptions.ForbiddenException;
+import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.io.FileIO;
+import org.junit.jupiter.api.Test;
+
+public class TestBaseMetastoreTableOperationsRefresh {
+
+  private static class NoopTableOperations extends BaseMetastoreTableOperations {
+    @Override
+    protected String tableName() {
+      return "test";
+    }
+
+    @Override
+    public FileIO io() {
+      return null;
+    }
+  }
+
+  @Test
+  public void refreshFailsFastOnForbidden() {
+    NoopTableOperations ops = new NoopTableOperations();
+    AtomicInteger attempts = new AtomicInteger();
+
+    assertThatThrownBy(
+            () ->
+                ops.refreshFromMetadataLocation(
+                    "file:/tmp/metadata.json",
+                    null,
+                    20,
+                    location -> {
+                      attempts.incrementAndGet();
+                      throw new ForbiddenException("Permission denied: %s", location);
+                    }))
+        .isInstanceOf(ForbiddenException.class)
+        .hasMessageContaining("Permission denied");
+
+    assertThat(attempts).hasValue(1);
+  }
+
+  @Test
+  public void refreshFailsFastOnNotFound() {
+    NoopTableOperations ops = new NoopTableOperations();
+    AtomicInteger attempts = new AtomicInteger();
+
+    assertThatThrownBy(
+            () ->
+                ops.refreshFromMetadataLocation(
+                    "file:/tmp/metadata.json",
+                    null,
+                    20,
+                    location -> {
+                      attempts.incrementAndGet();
+                      throw new NotFoundException("File does not exist: %s", location);
+                    }))
+        .isInstanceOf(NotFoundException.class)
+        .hasMessageContaining("File does not exist");
+
+    assertThat(attempts).hasValue(1);
+  }
+
+  @Test
+  public void refreshRetriesTransientFailures() {
+    NoopTableOperations ops = new NoopTableOperations();
+    AtomicInteger attempts = new AtomicInteger();
+
+    assertThatThrownBy(
+            () ->
+                ops.refreshFromMetadataLocation(
+                    "file:/tmp/metadata.json",
+                    null,
+                    2,
+                    location -> {
+                      attempts.incrementAndGet();
+                      throw new RuntimeException("transient failure");
+                    }))
+        .isInstanceOf(RuntimeException.class)
+        .hasMessageContaining("transient failure");
+
+    assertThat(attempts).hasValue(3);
+  }
+}