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 ef889d23639b..099f04c8391d 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 @@ -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 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 d8f415cfd221..1ca7a257ac2b 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 @@ -18,6 +18,7 @@ */ 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; @@ -25,18 +26,23 @@ 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; @@ -44,13 +50,14 @@ public final class TestS3InputStream { @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(); @@ -58,8 +65,25 @@ void testReadFullyClosesTheStream() throws IOException { @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); + } } diff --git a/core/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java b/core/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java index f1223705c11d..c97a9c6b13bc 100644 --- a/core/src/main/java/org/apache/iceberg/BaseMetastoreTableOperations.java +++ b/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))); diff --git a/core/src/main/java/org/apache/iceberg/hadoop/HadoopInputFile.java b/core/src/main/java/org/apache/iceberg/hadoop/HadoopInputFile.java index 558cac13a0e1..6c1275732ddb 100644 --- a/core/src/main/java/org/apache/iceberg/hadoop/HadoopInputFile.java +++ b/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); } diff --git a/core/src/main/java/org/apache/iceberg/view/BaseViewOperations.java b/core/src/main/java/org/apache/iceberg/view/BaseViewOperations.java index 0bab053bcb99..05674dd5d235 100644 --- a/core/src/main/java/org/apache/iceberg/view/BaseViewOperations.java +++ b/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))); diff --git a/core/src/test/java/org/apache/iceberg/TestBaseMetastoreTableOperationsRefresh.java b/core/src/test/java/org/apache/iceberg/TestBaseMetastoreTableOperationsRefresh.java new file mode 100644 index 000000000000..c176afc02a84 --- /dev/null +++ b/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); + } +}