From cc857d39c1199493fef9ec07ad84bb4c61d71fd8 Mon Sep 17 00:00:00 2001 From: Han You Date: Tue, 28 Jul 2026 15:32:48 -0500 Subject: [PATCH] Core, AWS: Fix REST catalog hanging on permission errors Refreshing table or view metadata retries any read failure up to 20 times with exponential backoff (about 90 seconds) unless it is a NotFoundException. A permission error is permanent, so retrying can never succeed and only delays surfacing the real failure. Behind a REST catalog, the server holds the loadTable request open for the whole retry loop, so clients hit their socket read timeout first and report SocketTimeoutException while the actual cause (for example an HDFS "Permission denied" on the metadata file) is only visible in server logs. Stuck requests also pin server worker threads, causing timeouts for unrelated tables. Translate access-denied errors into ForbiddenException at the read boundary - HadoopFileIO on AccessControlException and S3FileIO on S3Exception with status 403 - and stop metadata refresh retries on ForbiddenException. RESTCatalogAdapter already maps ForbiddenException to HTTP 403 and REST clients map 403 back to ForbiddenException, so callers now receive the underlying permission error immediately. --- .../apache/iceberg/aws/s3/S3InputStream.java | 26 ++++- .../iceberg/aws/s3/TestS3InputStream.java | 28 ++++- .../iceberg/BaseMetastoreTableOperations.java | 5 +- .../iceberg/hadoop/HadoopInputFile.java | 5 + .../iceberg/view/BaseViewOperations.java | 5 +- ...stBaseMetastoreTableOperationsRefresh.java | 106 ++++++++++++++++++ 6 files changed, 165 insertions(+), 10 deletions(-) create mode 100644 core/src/test/java/org/apache/iceberg/TestBaseMetastoreTableOperationsRefresh.java 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); + } +}