Skip to content

Commit debe771

Browse files
electrumwendigo
electrum
authored and
wendigo
committed
Use AWS v2 SDK for S3 in product tests
1 parent beb12d2 commit debe771

12 files changed

+208
-140
lines changed

‎testing/trino-product-tests/pom.xml

Lines changed: 25 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -19,16 +19,6 @@
1919
</properties>
2020

2121
<dependencies>
22-
<dependency>
23-
<groupId>com.amazonaws</groupId>
24-
<artifactId>aws-java-sdk-core</artifactId>
25-
</dependency>
26-
27-
<dependency>
28-
<groupId>com.amazonaws</groupId>
29-
<artifactId>aws-java-sdk-s3</artifactId>
30-
</dependency>
31-
3222
<dependency>
3323
<groupId>com.google.guava</groupId>
3424
<artifactId>guava</artifactId>
@@ -116,12 +106,6 @@
116106
<dependency>
117107
<groupId>io.trino</groupId>
118108
<artifactId>trino-hive</artifactId>
119-
<exclusions>
120-
<exclusion>
121-
<groupId>org.apache.httpcomponents</groupId>
122-
<artifactId>httpclient</artifactId>
123-
</exclusion>
124-
</exclusions>
125109
</dependency>
126110

127111
<dependency>
@@ -162,16 +146,6 @@
162146
<dependency>
163147
<groupId>io.trino.tempto</groupId>
164148
<artifactId>tempto-core</artifactId>
165-
<exclusions>
166-
<exclusion>
167-
<groupId>org.apache.httpcomponents</groupId>
168-
<artifactId>httpclient</artifactId>
169-
</exclusion>
170-
<exclusion>
171-
<groupId>org.apache.httpcomponents</groupId>
172-
<artifactId>httpcore</artifactId>
173-
</exclusion>
174-
</exclusions>
175149
</dependency>
176150

177151
<dependency>
@@ -224,11 +198,36 @@
224198
<artifactId>testng</artifactId>
225199
</dependency>
226200

201+
<dependency>
202+
<groupId>software.amazon.awssdk</groupId>
203+
<artifactId>auth</artifactId>
204+
</dependency>
205+
206+
<dependency>
207+
<groupId>software.amazon.awssdk</groupId>
208+
<artifactId>aws-core</artifactId>
209+
</dependency>
210+
227211
<dependency>
228212
<groupId>software.amazon.awssdk</groupId>
229213
<artifactId>glue</artifactId>
230214
</dependency>
231215

216+
<dependency>
217+
<groupId>software.amazon.awssdk</groupId>
218+
<artifactId>regions</artifactId>
219+
</dependency>
220+
221+
<dependency>
222+
<groupId>software.amazon.awssdk</groupId>
223+
<artifactId>s3</artifactId>
224+
</dependency>
225+
226+
<dependency>
227+
<groupId>software.amazon.awssdk</groupId>
228+
<artifactId>sdk-core</artifactId>
229+
</dependency>
230+
232231
<dependency>
233232
<groupId>com.clickhouse</groupId>
234233
<artifactId>clickhouse-jdbc</artifactId>

‎testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/S3ClientFactory.java

Lines changed: 21 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -13,17 +13,15 @@
1313
*/
1414
package io.trino.tests.product.deltalake;
1515

16-
import com.amazonaws.ClientConfiguration;
17-
import com.amazonaws.Protocol;
18-
import com.amazonaws.auth.AWSCredentials;
19-
import com.amazonaws.auth.AWSCredentialsProvider;
20-
import com.amazonaws.auth.AWSStaticCredentialsProvider;
21-
import com.amazonaws.auth.BasicAWSCredentials;
22-
import com.amazonaws.client.builder.AwsClientBuilder;
23-
import com.amazonaws.services.s3.AmazonS3;
24-
import com.amazonaws.services.s3.AmazonS3Client;
25-
import com.amazonaws.services.s3.model.Region;
2616
import io.trino.testing.minio.MinioClient;
17+
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
18+
import software.amazon.awssdk.auth.credentials.AwsCredentials;
19+
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
20+
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
21+
import software.amazon.awssdk.regions.Region;
22+
import software.amazon.awssdk.services.s3.S3Client;
23+
24+
import java.net.URI;
2725

2826
import static io.trino.testing.SystemEnvironmentUtils.requireEnv;
2927
import static io.trino.testing.minio.MinioClient.DEFAULT_MINIO_ACCESS_KEY;
@@ -34,7 +32,9 @@ final class S3ClientFactory
3432
public static final String AWS_S3_SERVER_TYPE = "aws";
3533
public static final String MINIO_S3_SERVER_TYPE = "minio";
3634

37-
public AmazonS3 createS3Client(String serverType)
35+
private S3ClientFactory() {}
36+
37+
public static S3Client createS3Client(String serverType)
3838
{
3939
return switch (serverType) {
4040
case AWS_S3_SERVER_TYPE -> createAwsS3Client();
@@ -43,25 +43,22 @@ public AmazonS3 createS3Client(String serverType)
4343
};
4444
}
4545

46-
private AmazonS3 createAwsS3Client()
46+
private static S3Client createAwsS3Client()
4747
{
4848
String region = requireEnv("AWS_REGION");
49-
return AmazonS3Client.builder().withRegion(region).build();
49+
return S3Client.builder().region(Region.of(region)).build();
5050
}
5151

52-
private AmazonS3 createMinioS3Client()
52+
private static S3Client createMinioS3Client()
5353
{
54-
AWSCredentials credentials = new BasicAWSCredentials(DEFAULT_MINIO_ACCESS_KEY, DEFAULT_MINIO_SECRET_KEY);
55-
AWSCredentialsProvider credentialsProvider = new AWSStaticCredentialsProvider(credentials);
56-
ClientConfiguration clientConfiguration = new ClientConfiguration()
57-
.withProtocol(Protocol.HTTP)
58-
.withSignerOverride("AWSS3V4SignerType");
54+
AwsCredentials credentials = AwsBasicCredentials.create(DEFAULT_MINIO_ACCESS_KEY, DEFAULT_MINIO_SECRET_KEY);
55+
AwsCredentialsProvider credentialsProvider = StaticCredentialsProvider.create(credentials);
5956

60-
return AmazonS3Client.builder()
61-
.withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(MinioClient.DEFAULT_MINIO_ENDPOINT, Region.US_East_2.name()))
62-
.withPathStyleAccessEnabled(true)
63-
.withClientConfiguration(clientConfiguration)
64-
.withCredentials(credentialsProvider)
57+
return S3Client.builder()
58+
.endpointOverride(URI.create(MinioClient.DEFAULT_MINIO_ENDPOINT))
59+
.region(Region.US_EAST_2)
60+
.forcePathStyle(true)
61+
.credentialsProvider(credentialsProvider)
6562
.build();
6663
}
6764
}

‎testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeActiveFilesCache.java

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,17 +13,19 @@
1313
*/
1414
package io.trino.tests.product.deltalake;
1515

16-
import com.amazonaws.services.s3.AmazonS3;
1716
import com.google.inject.Inject;
1817
import com.google.inject.name.Named;
18+
import io.trino.tempto.AfterMethodWithContext;
1919
import io.trino.tempto.BeforeMethodWithContext;
2020
import org.testng.annotations.Test;
21+
import software.amazon.awssdk.services.s3.S3Client;
2122

2223
import static io.trino.tempto.assertions.QueryAssert.Row.row;
2324
import static io.trino.tempto.assertions.QueryAssert.assertQueryFailure;
2425
import static io.trino.testing.TestingNames.randomNameSuffix;
2526
import static io.trino.tests.product.TestGroups.DELTA_LAKE_OSS;
2627
import static io.trino.tests.product.TestGroups.PROFILE_SPECIFIC_TESTS;
28+
import static io.trino.tests.product.deltalake.S3ClientFactory.createS3Client;
2729
import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.dropDeltaTableWithRetry;
2830
import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.removeS3Directory;
2931
import static io.trino.tests.product.utils.QueryExecutors.onDelta;
@@ -38,12 +40,19 @@ public class TestDeltaLakeActiveFilesCache
3840
@Named("s3.server_type")
3941
private String s3ServerType;
4042

41-
private AmazonS3 s3;
43+
private S3Client s3;
4244

4345
@BeforeMethodWithContext
4446
public void setup()
4547
{
46-
s3 = new S3ClientFactory().createS3Client(s3ServerType);
48+
s3 = createS3Client(s3ServerType);
49+
}
50+
51+
@AfterMethodWithContext
52+
public void cleanUp()
53+
{
54+
s3.close();
55+
s3 = null;
4756
}
4857

4958
@Test(groups = {DELTA_LAKE_OSS, PROFILE_SPECIFIC_TESTS})

‎testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeChangeDataFeedCompatibility.java

Lines changed: 15 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -13,16 +13,17 @@
1313
*/
1414
package io.trino.tests.product.deltalake;
1515

16-
import com.amazonaws.services.s3.AmazonS3;
17-
import com.amazonaws.services.s3.model.ListObjectsV2Result;
1816
import com.google.common.collect.ImmutableList;
1917
import com.google.inject.Inject;
2018
import com.google.inject.name.Named;
19+
import io.trino.tempto.AfterMethodWithContext;
2120
import io.trino.tempto.BeforeMethodWithContext;
2221
import io.trino.tempto.assertions.QueryAssert.Row;
2322
import io.trino.testng.services.Flaky;
2423
import org.testng.annotations.DataProvider;
2524
import org.testng.annotations.Test;
25+
import software.amazon.awssdk.services.s3.S3Client;
26+
import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
2627

2728
import java.util.List;
2829
import java.util.Map;
@@ -38,6 +39,7 @@
3839
import static io.trino.tests.product.TestGroups.DELTA_LAKE_EXCLUDE_113;
3940
import static io.trino.tests.product.TestGroups.DELTA_LAKE_OSS;
4041
import static io.trino.tests.product.TestGroups.PROFILE_SPECIFIC_TESTS;
42+
import static io.trino.tests.product.deltalake.S3ClientFactory.createS3Client;
4143
import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.DATABRICKS_COMMUNICATION_FAILURE_ISSUE;
4244
import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.DATABRICKS_COMMUNICATION_FAILURE_MATCH;
4345
import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.dropDeltaTableWithRetry;
@@ -55,12 +57,19 @@ public class TestDeltaLakeChangeDataFeedCompatibility
5557
@Named("s3.server_type")
5658
private String s3ServerType;
5759

58-
private AmazonS3 s3Client;
60+
private S3Client s3Client;
5961

6062
@BeforeMethodWithContext
6163
public void setup()
6264
{
63-
s3Client = new S3ClientFactory().createS3Client(s3ServerType);
65+
s3Client = createS3Client(s3ServerType);
66+
}
67+
68+
@AfterMethodWithContext
69+
public void cleanUp()
70+
{
71+
s3Client.close();
72+
s3Client = null;
6473
}
6574

6675
@Test(groups = {DELTA_LAKE_DATABRICKS, DELTA_LAKE_OSS, PROFILE_SPECIFIC_TESTS}, dataProvider = "columnMappingModeDataProvider")
@@ -886,7 +895,7 @@ private void assertThereIsNoCdfFileGenerated(String tableName, String tablePrope
886895
private void assertThatThereIsNoChangeDataFiles(String tableName)
887896
{
888897
String prefix = "databricks-compatibility-test-" + tableName + "/_change_data/";
889-
ListObjectsV2Result listResult = s3Client.listObjectsV2(bucketName, prefix);
890-
assertThat(listResult.getObjectSummaries()).isEmpty();
898+
ListObjectsV2Response response = s3Client.listObjectsV2(request -> request.bucket(bucketName).prefix(prefix));
899+
assertThat(response.contents()).isEmpty();
891900
}
892901
}

‎testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeCheckpointsCompatibility.java

Lines changed: 17 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -13,19 +13,19 @@
1313
*/
1414
package io.trino.tests.product.deltalake;
1515

16-
import com.amazonaws.services.s3.AmazonS3;
17-
import com.amazonaws.services.s3.model.ObjectListing;
18-
import com.amazonaws.services.s3.model.S3ObjectSummary;
1916
import com.google.common.collect.ImmutableList;
2017
import com.google.inject.Inject;
2118
import com.google.inject.name.Named;
19+
import io.trino.tempto.AfterMethodWithContext;
2220
import io.trino.tempto.BeforeMethodWithContext;
2321
import io.trino.tempto.assertions.QueryAssert.Row;
2422
import io.trino.tempto.query.QueryResult;
2523
import io.trino.testng.services.Flaky;
2624
import io.trino.tests.product.deltalake.util.DatabricksVersion;
2725
import org.testng.annotations.DataProvider;
2826
import org.testng.annotations.Test;
27+
import software.amazon.awssdk.services.s3.S3Client;
28+
import software.amazon.awssdk.services.s3.model.S3Object;
2929

3030
import java.math.BigDecimal;
3131
import java.util.List;
@@ -43,6 +43,7 @@
4343
import static io.trino.tests.product.TestGroups.DELTA_LAKE_DATABRICKS_154;
4444
import static io.trino.tests.product.TestGroups.DELTA_LAKE_OSS;
4545
import static io.trino.tests.product.TestGroups.PROFILE_SPECIFIC_TESTS;
46+
import static io.trino.tests.product.deltalake.S3ClientFactory.createS3Client;
4647
import static io.trino.tests.product.deltalake.TransactionLogAssertions.assertLastEntryIsCheckpointed;
4748
import static io.trino.tests.product.deltalake.TransactionLogAssertions.assertTransactionLogVersion;
4849
import static io.trino.tests.product.deltalake.util.DatabricksVersion.DATABRICKS_113_RUNTIME_VERSION;
@@ -62,16 +63,23 @@ public class TestDeltaLakeCheckpointsCompatibility
6263
@Named("s3.server_type")
6364
private String s3ServerType;
6465

65-
private AmazonS3 s3;
66+
private S3Client s3;
6667
private Optional<DatabricksVersion> databricksRuntimeVersion;
6768

6869
@BeforeMethodWithContext
6970
public void setup()
7071
{
71-
s3 = new S3ClientFactory().createS3Client(s3ServerType);
72+
s3 = createS3Client(s3ServerType);
7273
databricksRuntimeVersion = getDatabricksRuntimeVersion();
7374
}
7475

76+
@AfterMethodWithContext
77+
public void cleanUp()
78+
{
79+
s3.close();
80+
s3 = null;
81+
}
82+
7583
@Test(groups = {DELTA_LAKE_OSS, PROFILE_SPECIFIC_TESTS})
7684
public void testSparkCanReadTrinoCheckpoint()
7785
{
@@ -731,13 +739,9 @@ private List<String> listSidecarFiles(String bucketName, String tableDirectory)
731739

732740
private List<String> listS3Directory(String bucketName, String directory)
733741
{
734-
ImmutableList.Builder<String> result = ImmutableList.builder();
735-
ObjectListing listing = s3.listObjects(bucketName, directory);
736-
do {
737-
listing.getObjectSummaries().stream().map(S3ObjectSummary::getKey).forEach(result::add);
738-
listing = s3.listNextBatchOfObjects(listing);
739-
}
740-
while (listing.isTruncated());
741-
return result.build();
742+
return s3.listObjectsV2Paginator(request -> request.bucket(bucketName).prefix(directory))
743+
.contents().stream()
744+
.map(S3Object::key)
745+
.toList();
742746
}
743747
}

‎testing/trino-product-tests/src/main/java/io/trino/tests/product/deltalake/TestDeltaLakeCloneTableCompatibility.java

Lines changed: 18 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -13,14 +13,16 @@
1313
*/
1414
package io.trino.tests.product.deltalake;
1515

16-
import com.amazonaws.services.s3.AmazonS3;
1716
import com.google.common.collect.ImmutableList;
1817
import com.google.inject.Inject;
1918
import com.google.inject.name.Named;
19+
import io.trino.tempto.AfterMethodWithContext;
2020
import io.trino.tempto.BeforeMethodWithContext;
2121
import io.trino.tempto.assertions.QueryAssert.Row;
2222
import io.trino.testng.services.Flaky;
2323
import org.testng.annotations.Test;
24+
import software.amazon.awssdk.services.s3.S3Client;
25+
import software.amazon.awssdk.services.s3.model.S3Object;
2426

2527
import java.sql.Date;
2628
import java.util.List;
@@ -31,6 +33,7 @@
3133
import static io.trino.tests.product.TestGroups.DELTA_LAKE_DATABRICKS;
3234
import static io.trino.tests.product.TestGroups.DELTA_LAKE_OSS;
3335
import static io.trino.tests.product.TestGroups.PROFILE_SPECIFIC_TESTS;
36+
import static io.trino.tests.product.deltalake.S3ClientFactory.createS3Client;
3437
import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.DATABRICKS_COMMUNICATION_FAILURE_ISSUE;
3538
import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.DATABRICKS_COMMUNICATION_FAILURE_MATCH;
3639
import static io.trino.tests.product.deltalake.util.DeltaLakeTestUtils.dropDeltaTableWithRetry;
@@ -46,13 +49,20 @@ public class TestDeltaLakeCloneTableCompatibility
4649
@Named("s3.server_type")
4750
private String s3ServerType;
4851

49-
private AmazonS3 s3;
52+
private S3Client s3;
5053

5154
@BeforeMethodWithContext
5255
public void setup()
5356
{
5457
super.setUp();
55-
s3 = new S3ClientFactory().createS3Client(s3ServerType);
58+
s3 = createS3Client(s3ServerType);
59+
}
60+
61+
@AfterMethodWithContext
62+
public void cleanUp()
63+
{
64+
s3.close();
65+
s3 = null;
5666
}
5767

5868
@Test(groups = {DELTA_LAKE_OSS, PROFILE_SPECIFIC_TESTS})
@@ -546,9 +556,11 @@ private List<String> getToBeVacuumedDataFilesFromDryRun(String tableName)
546556

547557
private List<String> getFilesFromTableDirectory(String directory)
548558
{
549-
return s3.listObjectsV2(bucketName, directory).getObjectSummaries().stream()
550-
.filter(s3ObjectSummary -> !s3ObjectSummary.getKey().contains("/_delta_log"))
551-
.map(s3ObjectSummary -> format("s3://%s/%s", bucketName, s3ObjectSummary.getKey()))
559+
return s3.listObjectsV2Paginator(request -> request.bucket(bucketName).prefix(directory))
560+
.contents().stream()
561+
.map(S3Object::key)
562+
.filter(key -> !key.contains("/_delta_log"))
563+
.map(key -> format("s3://%s/%s", bucketName, key))
552564
.collect(toImmutableList());
553565
}
554566

0 commit comments

Comments
 (0)