diff --git a/plugin/trino-exchange-filesystem/src/main/java/io/trino/plugin/exchange/filesystem/FileSystemExchangeModule.java b/plugin/trino-exchange-filesystem/src/main/java/io/trino/plugin/exchange/filesystem/FileSystemExchangeModule.java index 36721bed7dc0..5a5d785b981d 100644 --- a/plugin/trino-exchange-filesystem/src/main/java/io/trino/plugin/exchange/filesystem/FileSystemExchangeModule.java +++ b/plugin/trino-exchange-filesystem/src/main/java/io/trino/plugin/exchange/filesystem/FileSystemExchangeModule.java @@ -49,7 +49,8 @@ protected void setup(Binder binder) List baseDirectories = buildConfigObject(FileSystemExchangeConfig.class).getBaseDirectories(); if (baseDirectories.stream().map(URI::getScheme).distinct().count() != 1) { - throw new TrinoException(CONFIGURATION_INVALID, "Multiple schemes in exchange base directories"); + binder.addError(new TrinoException(CONFIGURATION_INVALID, "Multiple schemes in exchange base directories")); + return; } String scheme = baseDirectories.get(0).getScheme(); if (scheme == null || scheme.equals("file")) { @@ -68,7 +69,7 @@ else if (ImmutableSet.of("abfs", "abfss").contains(scheme)) { configBinder(binder).bindConfig(ExchangeAzureConfig.class); } else { - throw new TrinoException(NOT_SUPPORTED, format("Scheme %s is not supported as exchange spooling storage", scheme)); + binder.addError(new TrinoException(NOT_SUPPORTED, format("Scheme %s is not supported as exchange spooling storage", scheme))); } } } diff --git a/plugin/trino-exchange-filesystem/src/main/java/io/trino/plugin/exchange/filesystem/s3/BufferWriteAsyncResponseTransformer.java b/plugin/trino-exchange-filesystem/src/main/java/io/trino/plugin/exchange/filesystem/s3/BufferWriteAsyncResponseTransformer.java index 46bf8e5517e3..871d038a30be 100644 --- a/plugin/trino-exchange-filesystem/src/main/java/io/trino/plugin/exchange/filesystem/s3/BufferWriteAsyncResponseTransformer.java +++ b/plugin/trino-exchange-filesystem/src/main/java/io/trino/plugin/exchange/filesystem/s3/BufferWriteAsyncResponseTransformer.java @@ -31,8 +31,6 @@ * An implementation of {@link AsyncResponseTransformer} that writes data to the specified range of the given buffer. * This class mimics the implementation of {@link ByteArrayAsyncResponseTransformer} but avoids memory copying. * - * {@link AsyncResponseTransformer} that writes the data to the specified range of the given buffer - * * @param Response POJO type. */ public final class BufferWriteAsyncResponseTransformer diff --git a/plugin/trino-exchange-filesystem/src/main/java/io/trino/plugin/exchange/filesystem/s3/S3FileSystemExchangeStorage.java b/plugin/trino-exchange-filesystem/src/main/java/io/trino/plugin/exchange/filesystem/s3/S3FileSystemExchangeStorage.java index 7c127fa78cab..0e977c7d2639 100644 --- a/plugin/trino-exchange-filesystem/src/main/java/io/trino/plugin/exchange/filesystem/s3/S3FileSystemExchangeStorage.java +++ b/plugin/trino-exchange-filesystem/src/main/java/io/trino/plugin/exchange/filesystem/s3/S3FileSystemExchangeStorage.java @@ -51,8 +51,6 @@ import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.s3.S3AsyncClient; import software.amazon.awssdk.services.s3.S3AsyncClientBuilder; -import software.amazon.awssdk.services.s3.S3Client; -import software.amazon.awssdk.services.s3.S3ClientBuilder; import software.amazon.awssdk.services.s3.S3Configuration; import software.amazon.awssdk.services.s3.model.AbortMultipartUploadRequest; import software.amazon.awssdk.services.s3.model.AbortMultipartUploadResponse; @@ -458,18 +456,6 @@ private static AwsCredentialsProvider createAwsCredentialsProvider(ExchangeS3Con return DefaultCredentialsProvider.create(); } - private S3Client createS3Client(AwsCredentialsProvider credentialsProvider, ClientOverrideConfiguration overrideConfig) - { - S3ClientBuilder clientBuilder = S3Client.builder() - .credentialsProvider(credentialsProvider) - .overrideConfiguration(overrideConfig); - - region.ifPresent(clientBuilder::region); - endpoint.ifPresent(s3Endpoint -> clientBuilder.endpointOverride(URI.create(s3Endpoint))); - - return clientBuilder.build(); - } - private S3AsyncClient createS3AsyncClient( AwsCredentialsProvider credentialsProvider, ClientOverrideConfiguration overrideConfig, diff --git a/plugin/trino-exchange-filesystem/src/test/java/io/trino/plugin/exchange/filesystem/TestFileSystemExchangeConfig.java b/plugin/trino-exchange-filesystem/src/test/java/io/trino/plugin/exchange/filesystem/TestFileSystemExchangeConfig.java index 16e903848615..f3727fe27bac 100644 --- a/plugin/trino-exchange-filesystem/src/test/java/io/trino/plugin/exchange/filesystem/TestFileSystemExchangeConfig.java +++ b/plugin/trino-exchange-filesystem/src/test/java/io/trino/plugin/exchange/filesystem/TestFileSystemExchangeConfig.java @@ -48,7 +48,7 @@ public void testDefaults() public void testExplicitPropertyMappings() { Map properties = ImmutableMap.builder() - .put("exchange.base-directories", "s3n://exchange-spooling-test/") + .put("exchange.base-directories", "s3://exchange-spooling-test/") .put("exchange.encryption-enabled", "false") .put("exchange.max-page-storage-size", "32MB") .put("exchange.sink-buffer-pool-min-size", "20") @@ -62,7 +62,7 @@ public void testExplicitPropertyMappings() .buildOrThrow(); FileSystemExchangeConfig expected = new FileSystemExchangeConfig() - .setBaseDirectories("s3n://exchange-spooling-test/") + .setBaseDirectories("s3://exchange-spooling-test/") .setExchangeEncryptionEnabled(false) .setMaxPageStorageSize(DataSize.of(32, MEGABYTE)) .setExchangeSinkBufferPoolMinSize(20)