This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new d6f034d85e [#13154] test: Replace MinIO credential-vending fixture
with RustFS (#13155)
d6f034d85e is described below
commit d6f034d85e40ac5f45d07ff5ade091ea3986a324
Author: Nevin Zheng <[email protected]>
AuthorDate: Mon Sep 14 23:00:32 2026 -0700
[#13154] test: Replace MinIO credential-vending fixture with RustFS (#13155)
### What changes were proposed in this pull request?
Replace the shared MinIO fixture with RustFS **1.0.0-rc.6** for fileset
connection testing and Iceberg credential vending. The fixture consumes
one version-and-digest pin from a packaged Docker Compose resource,
waits for signed S3 readiness, and creates buckets through the AWS SDK.
Preserve the Hadoop and Spark/S3FileIO tests, strengthen Spark
write-denial assertions to require `403 AccessDenied`, and add focused
checks of Gravitino's generated inline STS policy: read-only sessions,
object and bucket boundaries, both listing APIs, Hadoop's deliberate
bare-prefix behavior, deletion, and multipart upload.
Image index:
`rustfs/rustfs:1.0.0-rc.6@sha256:97171b3d72cd47dc81000f92ea84de25608bfc35a94c965501afaeb5d99f6035`.
Published `linux/amd64` and `linux/arm64` manifests were verified.
### Why are the changes needed?
Move these tests beyond the MinIO image-availability workaround in
#13112 while retaining their storage-authorization coverage. Related to
#13154; automated image updates and vulnerability reporting are handled
in a separate follow-up PR.
### Does this PR introduce _any_ user-facing change?
No. This changes test infrastructure only.
RustFS is a release candidate. These checks establish
caller-plus-inline-policy enforcement, not AWS named-role trust or
external-ID enforcement. The generated policy does not grant
`AbortMultipartUpload`; the focused multipart test uses root credentials
only to clean up unfinished uploads after failure. Further boundaries
and commands are documented in [the fixture
README](https://github.com/nevzheng/gravitino/blob/ffd21dd5ca8dedcca023731adb2aa5809889b566/integration-test-common/README-rustfs.md).
### How was this patch tested?
On Corretto 17.0.19 and Docker 29.4/OrbStack with the pinned
`linux/arm64` image:
- Image-pin parsing: **2 passed**.
- Signed STS policy and multipart probes: **5 passed**.
- Spark/S3FileIO owner, modify, select-only, and active-role tests: **4
passed**.
- Hadoop fileset catalog connection: **1 passed**.
- Changed-module Spotless checks and `git diff --check` passed.
Native Linux amd64 and deploy-mode execution remain for the existing
backend CI matrix. This Mac's build-output redirection needed its
documented compatibility switch because the existing catalog loader
expects module-local `build/resources/main` paths.
```shell
./gradlew :catalogs:catalog-lakehouse-iceberg:jar \
:catalogs:catalog-fileset:jar :catalogs:hadoop-common:jar \
-DcentralBuildOutputs.disabled=true
./gradlew :integration-test-common:test \
--tests
org.apache.gravitino.integration.test.container.TestRustFSContainer \
:catalogs:catalog-fileset:test \
--tests
org.apache.gravitino.catalog.fileset.integration.test.FilesetS3TokenConnectionIT
\
:iceberg:iceberg-rest-server:test \
--tests
org.apache.gravitino.iceberg.integration.test.IcebergRESTRustFSTokenAuthorizationIT
\
--tests org.apache.gravitino.iceberg.integration.test.RustFSS3TokenIT \
-PskipDockerTests=false -DcentralBuildOutputs.disabled=true
```
---
catalogs/catalog-fileset/build.gradle.kts | 2 +
.../test/FilesetS3TokenConnectionIT.java | 23 +-
iceberg/iceberg-rest-server/build.gradle.kts | 2 +
.../IcebergRESTCloudTokenAuthorizationBaseIT.java | 17 +-
... => IcebergRESTRustFSTokenAuthorizationIT.java} | 56 ++--
.../iceberg/integration/test/RustFSS3TokenIT.java | 313 +++++++++++++++++++++
integration-test-common/README-rustfs.md | 91 ++++++
integration-test-common/build.gradle.kts | 1 +
.../integration/test/container/ContainerSuite.java | 28 +-
.../integration/test/container/MinIOContainer.java | 149 ----------
.../test/container/RustFSContainer.java | 214 ++++++++++++++
.../test/container/TestRustFSContainer.java | 51 ++++
.../src/test/resources/docker-compose-rustfs.yml | 22 ++
13 files changed, 771 insertions(+), 198 deletions(-)
diff --git a/catalogs/catalog-fileset/build.gradle.kts
b/catalogs/catalog-fileset/build.gradle.kts
index ced821091f..d2879efcbb 100644
--- a/catalogs/catalog-fileset/build.gradle.kts
+++ b/catalogs/catalog-fileset/build.gradle.kts
@@ -79,6 +79,8 @@ dependencies {
testImplementation(libs.testcontainers)
testImplementation(libs.testcontainers.mysql)
+ // The RustFS fixture uses the S3 SDK v2; Hadoop's AWS bundle supplies the
S3 SDK v1.
+ testRuntimeOnly(libs.aws.s3)
testRuntimeOnly(libs.junit.jupiter.engine)
}
diff --git
a/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/integration/test/FilesetS3TokenConnectionIT.java
b/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/integration/test/FilesetS3TokenConnectionIT.java
index 86403dbd14..3e5b56f90f 100644
---
a/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/integration/test/FilesetS3TokenConnectionIT.java
+++
b/catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/integration/test/FilesetS3TokenConnectionIT.java
@@ -29,7 +29,7 @@ import org.apache.gravitino.Catalog;
import org.apache.gravitino.client.GravitinoMetalake;
import org.apache.gravitino.credential.CredentialConstants;
import org.apache.gravitino.credential.S3TokenCredential;
-import org.apache.gravitino.integration.test.container.MinIOContainer;
+import org.apache.gravitino.integration.test.container.RustFSContainer;
import org.apache.gravitino.integration.test.util.BaseIT;
import org.apache.gravitino.integration.test.util.GravitinoITUtils;
import org.apache.gravitino.storage.S3Properties;
@@ -38,13 +38,10 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
-/** Tests an existing Fileset catalog connection with S3 temporary credentials
against MinIO. */
+/** Tests an existing Fileset catalog connection with S3 temporary credentials
against RustFS. */
@Tag("gravitino-docker-test")
public class FilesetS3TokenConnectionIT extends BaseIT {
- private static final String ROLE_ARN = "arn:minio:iam:::role/test";
- private static final String REGION = "us-east-1";
-
private final String bucketName =
"fileset-connection-" + UUID.randomUUID().toString().replace("-", "");
private final String metalakeName =
GravitinoITUtils.genRandomName("fileset_connection_metalake");
@@ -57,25 +54,25 @@ public class FilesetS3TokenConnectionIT extends BaseIT {
@BeforeAll
public void startIntegrationTest() throws Exception {
copyBundleJarsToHadoop("aws-bundle");
- containerSuite.startMinIOContainer();
- MinIOContainer minIOContainer = containerSuite.getMinIOContainer();
- minIOContainer.createBucket(bucketName);
+ containerSuite.startRustFSContainer();
+ RustFSContainer rustFSContainer = containerSuite.getRustFSContainer();
+ rustFSContainer.createBucket(bucketName);
super.startIntegrationTest();
client.createMetalake(metalakeName, "comment", new HashMap<>());
metalake = client.loadMetalake(metalakeName);
- String endpoint = minIOContainer.getS3Endpoint();
+ String endpoint = rustFSContainer.getS3Endpoint();
Map<String, String> properties = new HashMap<>();
properties.put(LOCATION, String.format("s3a://%s", bucketName));
properties.put(FILESYSTEM_PROVIDERS, "s3");
properties.put(
CredentialConstants.CREDENTIAL_PROVIDERS,
S3TokenCredential.S3_TOKEN_CREDENTIAL_TYPE);
- properties.put(S3Properties.GRAVITINO_S3_REGION, REGION);
- properties.put(S3Properties.GRAVITINO_S3_ACCESS_KEY_ID,
MinIOContainer.ACCESS_KEY);
- properties.put(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY,
MinIOContainer.SECRET_KEY);
- properties.put(S3Properties.GRAVITINO_S3_ROLE_ARN, ROLE_ARN);
+ properties.put(S3Properties.GRAVITINO_S3_REGION, RustFSContainer.REGION);
+ properties.put(S3Properties.GRAVITINO_S3_ACCESS_KEY_ID,
RustFSContainer.ACCESS_KEY);
+ properties.put(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY,
RustFSContainer.SECRET_KEY);
+ properties.put(S3Properties.GRAVITINO_S3_ROLE_ARN,
RustFSContainer.ROLE_ARN);
properties.put(S3Properties.GRAVITINO_S3_ENDPOINT, endpoint);
properties.put(S3Properties.GRAVITINO_S3_STS_ENDPOINT, endpoint);
properties.put(S3Properties.GRAVITINO_S3_PATH_STYLE_ACCESS, "true");
diff --git a/iceberg/iceberg-rest-server/build.gradle.kts
b/iceberg/iceberg-rest-server/build.gradle.kts
index f69bb0ea66..20a8abe8a9 100644
--- a/iceberg/iceberg-rest-server/build.gradle.kts
+++ b/iceberg/iceberg-rest-server/build.gradle.kts
@@ -98,6 +98,8 @@ dependencies {
}
testImplementation(libs.awaitility)
+ // At runtime use the SDK supplied by iceberg-aws-bundle, keeping its S3 and
STS versions aligned.
+ testCompileOnly(libs.aws.s3)
testImplementation(libs.h2db)
testImplementation(libs.mysql.driver)
testImplementation(libs.postgresql.driver)
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTCloudTokenAuthorizationBaseIT.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTCloudTokenAuthorizationBaseIT.java
index c7912761b1..babce16fe7 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTCloudTokenAuthorizationBaseIT.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTCloudTokenAuthorizationBaseIT.java
@@ -42,6 +42,7 @@ import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.function.Executable;
/**
* Abstract base class for Iceberg REST credential vending integration tests
with cloud storage
@@ -172,8 +173,7 @@ public abstract class
IcebergRESTCloudTokenAuthorizationBaseIT extends IcebergAu
() -> sql("SELECT * FROM %s.%s.%s", SPARK_CATALOG_NAME, SCHEMA_NAME,
tableName));
grantSelectTableRole(tableName);
- Assertions.assertThrows(
- SparkException.class, () -> sql("INSERT INTO %s VALUES (1,1),(2,2)",
tableName));
+ assertStorageWriteDenied(() -> sql("INSERT INTO %s VALUES (1,1),(2,2)",
tableName));
List<Object[]> rows = sql("SELECT * FROM %s", tableName);
Assertions.assertEquals(0, rows.size());
@@ -233,8 +233,7 @@ public abstract class
IcebergRESTCloudTokenAuthorizationBaseIT extends IcebergAu
// Narrowed to the select-only role the caller gets a read-only
credential, so the write fails
// inside Spark instead of being rejected by Gravitino.
- Assertions.assertThrows(
- SparkException.class,
+ assertStorageWriteDenied(
() ->
sql(
"INSERT INTO %s.%s.%s VALUES (3,3)",
@@ -245,6 +244,16 @@ public abstract class
IcebergRESTCloudTokenAuthorizationBaseIT extends IcebergAu
Assertions.assertEquals(2, rows.size());
}
+ /**
+ * Checks that a storage write fails inside Spark. Providers may
additionally check the storage
+ * error code to distinguish policy denial from an unrelated client failure.
+ *
+ * @param operation the write operation to reject
+ */
+ protected void assertStorageWriteDenied(Executable operation) {
+ Assertions.assertThrows(SparkException.class, operation);
+ }
+
/**
* The narrowed catalog names this role in a static header, so it has to
carry every privilege the
* read path needs; narrowing deactivates the per-test USE_SCHEMA role.
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTMinIOTokenAuthorizationIT.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTRustFSTokenAuthorizationIT.java
similarity index 62%
rename from
iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTMinIOTokenAuthorizationIT.java
rename to
iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTRustFSTokenAuthorizationIT.java
index 5b7fb88bcd..3c58d5f847 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTMinIOTokenAuthorizationIT.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTRustFSTokenAuthorizationIT.java
@@ -24,30 +24,30 @@ import
org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
import org.apache.gravitino.credential.CredentialConstants;
import org.apache.gravitino.credential.S3TokenCredential;
import org.apache.gravitino.integration.test.container.ContainerSuite;
-import org.apache.gravitino.integration.test.container.MinIOContainer;
+import org.apache.gravitino.integration.test.container.RustFSContainer;
import org.apache.gravitino.integration.test.util.BaseIT;
import org.apache.gravitino.integration.test.util.ITUtils;
import org.apache.gravitino.storage.S3Properties;
+import org.apache.spark.SparkException;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.function.Executable;
+import software.amazon.awssdk.services.s3.model.S3Exception;
/**
- * Runs the cloud credential vending tests against MinIO. MinIO enforces the
session policy attached
- * to an {@code AssumeRole} request, so the read-only downgrade a caller
receives without {@code
- * MODIFY_TABLE} is observable without a cloud account.
+ * Runs the cloud credential vending tests against RustFS. RustFS enforces the
session policy
+ * attached to an {@code AssumeRole} request, so the read-only downgrade a
caller receives without
+ * {@code MODIFY_TABLE} is observable without a cloud account.
*/
@Tag("gravitino-docker-test")
-public class IcebergRESTMinIOTokenAuthorizationIT extends
IcebergRESTCloudTokenAuthorizationBaseIT {
+public class IcebergRESTRustFSTokenAuthorizationIT
+ extends IcebergRESTCloudTokenAuthorizationBaseIT {
- private static final String BUCKET_NAME = "gravitino-minio-it";
+ private static final String BUCKET_NAME = "gravitino-rustfs-it";
private static final String BUNDLE_NAME = "iceberg-aws-bundle";
- // MinIO does not resolve the account or resource part, but the SDK requires
a well-formed ARN.
- private static final String ROLE_ARN = "arn:minio:iam:::role/test";
-
- private static final String REGION = "us-east-1";
-
private final ContainerSuite containerSuite = ContainerSuite.getInstance();
private String s3Endpoint;
@@ -55,10 +55,10 @@ public class IcebergRESTMinIOTokenAuthorizationIT extends
IcebergRESTCloudTokenA
@BeforeAll
public void startIntegrationTest() throws Exception {
- containerSuite.startMinIOContainer();
- MinIOContainer minIOContainer = containerSuite.getMinIOContainer();
- minIOContainer.createBucket(BUCKET_NAME);
- this.s3Endpoint = minIOContainer.getS3Endpoint();
+ containerSuite.startRustFSContainer();
+ RustFSContainer rustFSContainer = containerSuite.getRustFSContainer();
+ rustFSContainer.createBucket(BUCKET_NAME);
+ this.s3Endpoint = rustFSContainer.getS3Endpoint();
this.warehouse = String.format("s3://%s/test1", BUCKET_NAME);
// In deploy mode the server resolves S3FileIO from its own classpath, so
the bundle has to be
@@ -75,22 +75,36 @@ public class IcebergRESTMinIOTokenAuthorizationIT extends
IcebergRESTCloudTokenA
Map<String, String> configMap = new HashMap<>();
configMap.put(
CredentialConstants.CREDENTIAL_PROVIDERS,
S3TokenCredential.S3_TOKEN_CREDENTIAL_TYPE);
- configMap.put(S3Properties.GRAVITINO_S3_REGION, REGION);
- configMap.put(S3Properties.GRAVITINO_S3_ACCESS_KEY_ID,
MinIOContainer.ACCESS_KEY);
- configMap.put(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY,
MinIOContainer.SECRET_KEY);
- configMap.put(S3Properties.GRAVITINO_S3_ROLE_ARN, ROLE_ARN);
+ configMap.put(S3Properties.GRAVITINO_S3_REGION, RustFSContainer.REGION);
+ configMap.put(S3Properties.GRAVITINO_S3_ACCESS_KEY_ID,
RustFSContainer.ACCESS_KEY);
+ configMap.put(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY,
RustFSContainer.SECRET_KEY);
+ configMap.put(S3Properties.GRAVITINO_S3_ROLE_ARN,
RustFSContainer.ROLE_ARN);
configMap.put(S3Properties.GRAVITINO_S3_ENDPOINT, s3Endpoint);
configMap.put(S3Properties.GRAVITINO_S3_STS_ENDPOINT, s3Endpoint);
- // MinIO serves buckets as a path segment rather than a subdomain.
+ // RustFS serves buckets as a path segment rather than a subdomain.
configMap.put(S3Properties.GRAVITINO_S3_PATH_STYLE_ACCESS, "true");
configMap.put(IcebergConstants.IO_IMPL,
"org.apache.iceberg.aws.s3.S3FileIO");
configMap.put(IcebergConstants.WAREHOUSE, warehouse);
return configMap;
}
+ @Override
+ protected void assertStorageWriteDenied(Executable operation) {
+ SparkException error = Assertions.assertThrows(SparkException.class,
operation);
+ Throwable cause = error;
+ while (cause != null && !(cause instanceof S3Exception)) {
+ cause = cause.getCause();
+ }
+ Assertions.assertNotNull(
+ cause, "Expected an S3 authorization failure in the Spark cause
chain");
+ S3Exception denial = (S3Exception) cause;
+ Assertions.assertEquals(403, denial.statusCode());
+ Assertions.assertEquals("AccessDenied",
denial.awsErrorDetails().errorCode());
+ }
+
@Override
protected String getCloudProviderName() {
- return "minio";
+ return "rustfs";
}
@Override
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/RustFSS3TokenIT.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/RustFSS3TokenIT.java
new file mode 100644
index 0000000000..ebea11bce8
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/RustFSS3TokenIT.java
@@ -0,0 +1,313 @@
+/*
+ * 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.gravitino.iceberg.integration.test;
+
+import java.io.IOException;
+import java.net.URI;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.gravitino.credential.CredentialConstants;
+import org.apache.gravitino.credential.PathBasedCredentialContext;
+import org.apache.gravitino.credential.S3TokenCredential;
+import org.apache.gravitino.integration.test.container.RustFSContainer;
+import org.apache.gravitino.s3.credential.S3TokenGenerator;
+import org.apache.gravitino.storage.S3Properties;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.function.Executable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+/** Exercises Gravitino's generated inline STS policies through signed
requests to RustFS. */
+@Tag("gravitino-docker-test")
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class RustFSS3TokenIT {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(RustFSS3TokenIT.class);
+
+ private final String bucket = "gravitino-policy-" + UUID.randomUUID();
+ private final String otherBucket = "gravitino-other-" + UUID.randomUUID();
+ private RustFSContainer container;
+ private S3Client rootClient;
+
+ @BeforeAll
+ void startStorage() {
+ container = RustFSContainer.builder().build();
+ container.start();
+ container.createBucket(bucket);
+ container.createBucket(otherBucket);
+ rootClient =
+ client(AwsBasicCredentials.create(RustFSContainer.ACCESS_KEY,
RustFSContainer.SECRET_KEY));
+ for (String key : List.of("read/file", "write/file", "write_sibling/file",
"other/file")) {
+ put(rootClient, bucket, key);
+ }
+ put(rootClient, otherBucket, "write/file");
+ }
+
+ @AfterAll
+ void stopStorage() {
+ if (rootClient != null) {
+ rootClient.close();
+ }
+ if (container != null) {
+ container.close();
+ }
+ }
+
+ @Test
+ void testObjectResourcesAndReadOnlyPermissions() throws IOException {
+ try (S3Client scoped = scopedClient(false)) {
+ Assertions.assertEquals("data", read(scoped, bucket, "read/file"));
+ Assertions.assertEquals("data", read(scoped, bucket, "write/file"));
+ put(scoped, bucket, "write/new");
+ Assertions.assertEquals("data", read(scoped, bucket, "write/new"));
+ scoped.deleteObject(request -> request.bucket(bucket).key("write/new"));
+ Assertions.assertTrue(
+ rootClient
+ .listObjectsV2(request ->
request.bucket(bucket).prefix("write/new"))
+ .contents()
+ .isEmpty());
+
+ assertAccessDenied(() -> put(scoped, bucket, "read/new"));
+ assertAccessDenied(
+ () -> scoped.deleteObject(request ->
request.bucket(bucket).key("read/file")));
+ for (String key : List.of("other/file", "write_sibling/file", "write")) {
+ assertAccessDenied(() -> read(scoped, bucket, key));
+ assertAccessDenied(() -> put(scoped, bucket, key));
+ assertAccessDenied(() -> scoped.deleteObject(request ->
request.bucket(bucket).key(key)));
+ }
+ assertAccessDenied(() -> read(scoped, otherBucket, "write/file"));
+ assertAccessDenied(() -> put(scoped, otherBucket, "write/new"));
+ assertAccessDenied(
+ () -> scoped.deleteObject(request ->
request.bucket(otherBucket).key("write/file")));
+ scoped.getBucketLocation(request -> request.bucket(bucket));
+ assertAccessDenied(() -> scoped.getBucketLocation(request ->
request.bucket(otherBucket)));
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void testListPrefixes(boolean includeLocationPrefix) throws IOException {
+ try (S3Client scoped = scopedClient(includeLocationPrefix)) {
+ Assertions.assertEquals(
+ List.of("read/file"),
+ scoped.listObjects(request ->
request.bucket(bucket).prefix("read/")).contents().stream()
+ .map(S3Object::key)
+ .collect(Collectors.toList()));
+ Assertions.assertEquals(
+ List.of("read/file"),
+ scoped
+ .listObjectsV2(request -> request.bucket(bucket).prefix("read/"))
+ .contents()
+ .stream()
+ .map(S3Object::key)
+ .collect(Collectors.toList()));
+ scoped.listObjectsV2(request ->
request.bucket(bucket).prefix("write/subdirectory/"));
+ for (String prefix : List.of("", "other/", "write_sibling/")) {
+ assertAccessDenied(
+ () -> scoped.listObjects(request ->
request.bucket(bucket).prefix(prefix)));
+ assertAccessDenied(
+ () -> scoped.listObjectsV2(request ->
request.bucket(bucket).prefix(prefix)));
+ }
+ assertAccessDenied(() -> scoped.listObjects(request ->
request.bucket(bucket)));
+ assertAccessDenied(() -> scoped.listObjectsV2(request ->
request.bucket(bucket)));
+ assertAccessDenied(
+ () -> scoped.listObjectsV2(request ->
request.bucket(otherBucket).prefix("write/")));
+ if (includeLocationPrefix) {
+ // Hadoop's directory probe deliberately permits listing the bare
location prefix. That
+ // also reveals sibling key names sharing it, but grants no access to
their object data.
+ Assertions.assertTrue(
+ scoped
+ .listObjectsV2(request ->
request.bucket(bucket).prefix("write"))
+ .contents()
+ .stream()
+ .anyMatch(object ->
object.key().equals("write_sibling/file")));
+ scoped.listObjects(request -> request.bucket(bucket).prefix("write"));
+ assertAccessDenied(() -> read(scoped, bucket, "write_sibling/file"));
+ } else {
+ assertAccessDenied(
+ () -> scoped.listObjects(request ->
request.bucket(bucket).prefix("write")));
+ assertAccessDenied(
+ () -> scoped.listObjectsV2(request ->
request.bucket(bucket).prefix("write")));
+ }
+ }
+ }
+
+ @Test
+ void testReadOnlySession() throws IOException {
+ try (S3Client scoped = scopedClient(false, false)) {
+ Assertions.assertEquals("data", read(scoped, bucket, "read/file"));
+ Assertions.assertEquals("data", read(scoped, bucket, "write/file"));
+ scoped.listObjectsV2(request -> request.bucket(bucket).prefix("write/"));
+ assertAccessDenied(() -> put(scoped, bucket, "write/read-only"));
+ assertAccessDenied(
+ () -> scoped.deleteObject(request ->
request.bucket(bucket).key("write/file")));
+ assertAccessDenied(
+ () ->
+ scoped.createMultipartUpload(
+ request -> request.bucket(bucket).key("write/read-only")));
+ }
+ }
+
+ @Test
+ void testMultipartUploadWithGeneratedWritePolicy() throws IOException {
+ String key = "write/multipart";
+ try (S3Client scoped = scopedClient(false)) {
+ assertAccessDenied(
+ () ->
+ scoped.createMultipartUpload(
+ request -> request.bucket(bucket).key("read/multipart")));
+ assertAccessDenied(
+ () ->
+ scoped.createMultipartUpload(
+ request -> request.bucket(bucket).key("other/multipart")));
+ String uploadId =
+ scoped.createMultipartUpload(request ->
request.bucket(bucket).key(key)).uploadId();
+ boolean completed = false;
+ try {
+ byte[] firstPart = new byte[5 * 1024 * 1024];
+ String firstETag =
+ scoped
+ .uploadPart(
+ request ->
request.bucket(bucket).key(key).uploadId(uploadId).partNumber(1),
+ RequestBody.fromBytes(firstPart))
+ .eTag();
+ String secondETag =
+ scoped
+ .uploadPart(
+ request ->
request.bucket(bucket).key(key).uploadId(uploadId).partNumber(2),
+ RequestBody.fromString("tail"))
+ .eTag();
+ scoped.completeMultipartUpload(
+ request ->
+ request
+ .bucket(bucket)
+ .key(key)
+ .uploadId(uploadId)
+ .multipartUpload(
+ upload ->
+ upload.parts(
+
CompletedPart.builder().partNumber(1).eTag(firstETag).build(),
+
CompletedPart.builder().partNumber(2).eTag(secondETag).build())));
+ completed = true;
+ Assertions.assertEquals(
+ firstPart.length + 4,
+ scoped
+ .getObjectAsBytes(request -> request.bucket(bucket).key(key))
+ .asByteArray()
+ .length);
+ scoped.deleteObject(request -> request.bucket(bucket).key(key));
+ } finally {
+ // Abort is not granted by Gravitino's generated policy. Root
credentials clean up an
+ // unfinished upload if any assertion or request above fails.
+ if (!completed) {
+ try {
+ rootClient.abortMultipartUpload(
+ request -> request.bucket(bucket).key(key).uploadId(uploadId));
+ } catch (Exception cleanupFailure) {
+ LOG.warn(
+ "Failed to clean up multipart upload {} after a test failure",
+ uploadId,
+ cleanupFailure);
+ }
+ }
+ }
+ }
+ }
+
+ private S3Client scopedClient(boolean includeLocationPrefix) throws
IOException {
+ return scopedClient(includeLocationPrefix, true);
+ }
+
+ private S3Client scopedClient(boolean includeLocationPrefix, boolean
writable)
+ throws IOException {
+ Map<String, String> properties = new HashMap<>();
+ properties.put(S3Properties.GRAVITINO_S3_REGION, RustFSContainer.REGION);
+ properties.put(S3Properties.GRAVITINO_S3_ACCESS_KEY_ID,
RustFSContainer.ACCESS_KEY);
+ properties.put(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY,
RustFSContainer.SECRET_KEY);
+ properties.put(S3Properties.GRAVITINO_S3_ROLE_ARN,
RustFSContainer.ROLE_ARN);
+ properties.put(S3Properties.GRAVITINO_S3_STS_ENDPOINT,
container.getS3Endpoint());
+ properties.put(
+ CredentialConstants.S3_CREDENTIAL_LIST_LOCATION_PREFIX,
+ Boolean.toString(includeLocationPrefix));
+ try (S3TokenGenerator generator = new S3TokenGenerator()) {
+ generator.initialize(properties);
+ S3TokenCredential credential =
+ generator.generate(
+ new PathBasedCredentialContext(
+ "rustfs-test",
+ writable ? Set.of("s3://" + bucket + "/write") : Set.of(),
+ Set.of("s3://" + bucket + "/read", "s3://" + bucket +
"/write")));
+ Assertions.assertNotEquals(RustFSContainer.ACCESS_KEY,
credential.accessKeyId());
+ Assertions.assertFalse(credential.sessionToken().isEmpty());
+ return client(
+ AwsSessionCredentials.create(
+ credential.accessKeyId(), credential.secretAccessKey(),
credential.sessionToken()));
+ }
+ }
+
+ private S3Client client(AwsCredentials credentials) {
+ return S3Client.builder()
+ .endpointOverride(URI.create(container.getS3Endpoint()))
+ .region(Region.of(RustFSContainer.REGION))
+ .forcePathStyle(true)
+ .credentialsProvider(StaticCredentialsProvider.create(credentials))
+ .overrideConfiguration(
+ config ->
+ config
+ .apiCallTimeout(Duration.ofSeconds(30))
+ .apiCallAttemptTimeout(Duration.ofSeconds(10)))
+ .build();
+ }
+
+ private static void put(S3Client client, String bucket, String key) {
+ client.putObject(request -> request.bucket(bucket).key(key),
RequestBody.fromString("data"));
+ }
+
+ private static String read(S3Client client, String bucket, String key) {
+ return client.getObjectAsBytes(request ->
request.bucket(bucket).key(key)).asUtf8String();
+ }
+
+ private static void assertAccessDenied(Executable request) {
+ S3Exception error = Assertions.assertThrows(S3Exception.class, request);
+ Assertions.assertEquals(403, error.statusCode());
+ Assertions.assertEquals("AccessDenied",
error.awsErrorDetails().errorCode());
+ }
+}
diff --git a/integration-test-common/README-rustfs.md
b/integration-test-common/README-rustfs.md
new file mode 100644
index 0000000000..eb0afc3b80
--- /dev/null
+++ b/integration-test-common/README-rustfs.md
@@ -0,0 +1,91 @@
+<!--
+ 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.
+-->
+
+# RustFS credential-vending test fixture
+
+The fileset connection and Iceberg credential-vending integration tests use
+RustFS instead of MinIO. The authoritative version and multi-platform digest
+are in
[`src/test/resources/docker-compose-rustfs.yml`](src/test/resources/docker-compose-rustfs.yml).
+`RustFSContainer` reads this packaged resource directly; changing that image
+entry changes the image used by the tests. Keep exactly one unquoted image
+entry with a release tag and SHA-256 digest.
+
+The initial pin is RustFS **1.0.0-rc.6**, a release candidate. Its published
+image index includes `linux/amd64` and `linux/arm64`. The fixture uses the
+image's default startup command, waits for a successful signed S3 request,
+and creates buckets with the AWS SDK. It needs no MinIO client or additional
+setup container. Testcontainers publishes the S3/STS port for host JVMs,
+forwards container logs, and removes the container and its temporary data
+when the owning test or shared container suite closes.
+
+## Run the compatibility checks
+
+Use JDK 17 and a running Docker daemon. The image-pin unit tests and focused
+signed STS policy checks do not start a Gravitino server:
+
+```shell
+./gradlew :integration-test-common:test \
+ --tests org.apache.gravitino.integration.test.container.TestRustFSContainer \
+ :iceberg:iceberg-rest-server:test \
+ --tests org.apache.gravitino.iceberg.integration.test.RustFSS3TokenIT \
+ -PskipDockerTests=false
+```
+
+Build the catalog runtime before running the existing embedded client suites
+from a fresh checkout:
+
+```shell
+./gradlew :catalogs:catalog-lakehouse-iceberg:jar
+./gradlew :catalogs:catalog-fileset:test \
+ --tests
org.apache.gravitino.catalog.fileset.integration.test.FilesetS3TokenConnectionIT
\
+ :iceberg:iceberg-rest-server:test \
+ --tests
org.apache.gravitino.iceberg.integration.test.IcebergRESTRustFSTokenAuthorizationIT
\
+ -PskipDockerTests=false
+```
+
+The backend integration workflow also runs the affected suites in its existing
+Linux amd64 embedded/deploy matrix when the image resource changes. Deploy
+tests require the normal assembled Gravitino distribution.
+
+The focused tests obtain temporary credentials through `S3TokenGenerator`,
+using its actual inline policy and signed `AssumeRole` request. They check
+read/write/delete permissions, another bucket, adjacent object paths,
+`GetBucketLocation`, both listing APIs with allowed and denied prefixes,
+read-only sessions, and a completed two-part multipart upload. Denials must
+be S3 `403 AccessDenied` responses. The RustFS Spark tests also require that
+error in the cause chain for select-only and narrowed-active-role writes.
+
+## Compatibility boundaries
+
+- RustFS derives a session's permissions from its caller and inline policy.
+ The fixture's role ARN satisfies the SDK's request format; these tests do
+ not establish AWS named-role trust or external-ID enforcement.
+- Fileset policies deliberately allow the bare location prefix for Hadoop
+ directory probing. This permits listing sibling key names sharing that
+ prefix, while the object-resource policy still denies access to their
+ contents. The focused checks preserve this distinction.
+- Gravitino's generated write policy grants `PutObject` and `DeleteObject`;
+ it does not grant `AbortMultipartUpload`. The multipart test uses root
+ credentials to clean up an unfinished upload after a failure. Passing
+ these checks does not establish every S3 multipart edge case.
+- Image scanning cannot establish Rust dependency coverage when an image
+ lacks usable Rust package metadata. See the linked image-maintenance
+ follow-up for update detection, scan coverage, and maintainer triage.
+
+Related to [#13154](https://github.com/apache/gravitino/issues/13154).
diff --git a/integration-test-common/build.gradle.kts
b/integration-test-common/build.gradle.kts
index a73ec0a627..7adeffc08b 100644
--- a/integration-test-common/build.gradle.kts
+++ b/integration-test-common/build.gradle.kts
@@ -36,6 +36,7 @@ dependencies {
testImplementation(project(":server"))
testImplementation(project(":server-common"))
testImplementation(libs.awaitility)
+ testImplementation(libs.aws.s3)
testImplementation(libs.bundles.jetty)
testImplementation(libs.bundles.jersey)
testImplementation(libs.bundles.jwt)
diff --git
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/ContainerSuite.java
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/ContainerSuite.java
index 05ecac0186..990a6b6f9e 100644
---
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/ContainerSuite.java
+++
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/ContainerSuite.java
@@ -93,7 +93,7 @@ public class ContainerSuite implements Closeable {
private static volatile GravitinoLocalStackContainer
gravitinoLocalStackContainer;
- private static volatile MinIOContainer minIOContainer;
+ private static volatile RustFSContainer rustFSContainer;
/**
* We can share the same Hive container as Hive container with S3 contains
the following
@@ -698,27 +698,33 @@ public class ContainerSuite implements Closeable {
return gravitinoLocalStackContainer;
}
- public void startMinIOContainer() {
+ /** Starts the shared RustFS object store for S3 credential-vending tests. */
+ public void startRustFSContainer() {
ITUtils.cleanDisk();
- if (minIOContainer == null) {
+ if (rustFSContainer == null) {
synchronized (ContainerSuite.class) {
- if (minIOContainer == null) {
- MinIOContainer.Builder builder =
MinIOContainer.builder().withNetwork(network);
- MinIOContainer container = closer.register(builder.build());
+ if (rustFSContainer == null) {
+ RustFSContainer.Builder builder =
RustFSContainer.builder().withNetwork(network);
+ RustFSContainer container = closer.register(builder.build());
try {
container.start();
} catch (Exception e) {
- LOG.error("Failed to start MinIO container", e);
- throw new RuntimeException("Failed to start MinIO container", e);
+ LOG.error("Failed to start RustFS container", e);
+ throw new RuntimeException("Failed to start RustFS container", e);
}
- minIOContainer = container;
+ rustFSContainer = container;
}
}
}
}
- public MinIOContainer getMinIOContainer() {
- return minIOContainer;
+ /**
+ * Returns the shared RustFS fixture after {@link #startRustFSContainer()}.
+ *
+ * @return the RustFS fixture
+ */
+ public RustFSContainer getRustFSContainer() {
+ return rustFSContainer;
}
public HiveContainer getHiveContainerWithS3() {
diff --git
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/MinIOContainer.java
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/MinIOContainer.java
deleted file mode 100644
index 79c8252020..0000000000
---
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/MinIOContainer.java
+++ /dev/null
@@ -1,149 +0,0 @@
-/*
- * 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.gravitino.integration.test.container;
-
-import static java.lang.String.format;
-import static org.awaitility.Awaitility.await;
-
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableSet;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import org.rnorth.ducttape.Preconditions;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.testcontainers.containers.Container;
-import org.testcontainers.containers.Network;
-
-/**
- * A MinIO container providing S3-compatible object storage. MinIO enforces
the session policy of an
- * {@code AssumeRole} request, so it can exercise credential vending,
including the read-only
- * downgrade, without a cloud account.
- */
-public class MinIOContainer extends BaseContainer {
- public static final Logger LOG =
LoggerFactory.getLogger(MinIOContainer.class);
-
- // MinIO no longer publishes this image on Docker Hub; quay.io carries the
same release.
- public static final String DEFAULT_IMAGE =
"quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z";
- public static final String HOST_NAME = "gravitino-ci-minio";
- public static final int PORT = 9000;
- public static final String ACCESS_KEY = "minioadmin";
- public static final String SECRET_KEY = "minioadmin123";
-
- public static Builder builder() {
- return new Builder();
- }
-
- private MinIOContainer(
- String image,
- String hostName,
- Set<Integer> ports,
- Map<String, String> extraHosts,
- Map<String, String> filesToMount,
- Map<String, String> envVars,
- Optional<Network> network) {
- super(image, hostName, ports, extraHosts, filesToMount, envVars, network);
- }
-
- @Override
- protected void setupContainer() {
- super.setupContainer();
- // The image defines no default command; the server needs to be told where
to keep its data.
- container.withCommand("server", "/data");
- withLogConsumer(new PrintingContainerLog(format("%-14s| ",
"MinIOContainer")));
- }
-
- @Override
- public void start() {
- super.start();
- Preconditions.check("MinIO container startup failed!",
checkContainerStatus(5));
- }
-
- @Override
- protected boolean checkContainerStatus(int retryLimit) {
- // `mc alias set` contacts the server, so it only succeeds once MinIO is
accepting requests.
- await()
- .atMost(30, TimeUnit.SECONDS)
- .pollInterval(30 / retryLimit, TimeUnit.SECONDS)
- .until(
- () -> {
- try {
- return setLocalAlias().getExitCode() == 0;
- } catch (Exception e) {
- LOG.warn("MinIO is not ready yet", e);
- return false;
- }
- });
- return true;
- }
-
- /**
- * Creates a bucket, so that a warehouse location inside it can be used.
- *
- * @param bucketName the bucket to create
- */
- public void createBucket(String bucketName) {
- try {
- setLocalAlias();
- Container.ExecResult result = executeInContainer("mc", "mb", "local/" +
bucketName);
- if (result.getExitCode() != 0) {
- throw new RuntimeException(
- format("Failed to create bucket %s: %s", bucketName,
result.getStderr()));
- }
- } catch (Exception e) {
- throw new RuntimeException("Failed to create bucket " + bucketName, e);
- }
- }
-
- private Container.ExecResult setLocalAlias() throws Exception {
- return executeInContainer(
- "mc", "alias", "set", "local", "http://localhost:"; + PORT, ACCESS_KEY,
SECRET_KEY);
- }
-
- /**
- * Returns the S3 endpoint of this container, reachable from the host once
container addresses are
- * routed.
- *
- * @return the endpoint, for example {@code http://10.20.30.5:9000}
- */
- public String getS3Endpoint() {
- return format("http://%s:%d";, getContainerIpAddress(), PORT);
- }
-
- /** Builder for {@link MinIOContainer}. */
- public static class Builder
- extends BaseContainer.Builder<MinIOContainer.Builder, MinIOContainer> {
-
- private Builder() {
- this.image = DEFAULT_IMAGE;
- this.hostName = HOST_NAME;
- this.exposePorts = ImmutableSet.of(PORT);
- this.envVars =
- ImmutableMap.of("MINIO_ROOT_USER", ACCESS_KEY,
"MINIO_ROOT_PASSWORD", SECRET_KEY);
- }
-
- @Override
- public MinIOContainer build() {
- return new MinIOContainer(
- image, hostName, exposePorts, extraHosts, filesToMount, envVars,
network);
- }
- }
-}
diff --git
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/RustFSContainer.java
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/RustFSContainer.java
new file mode 100644
index 0000000000..7177198033
--- /dev/null
+++
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/RustFSContainer.java
@@ -0,0 +1,214 @@
+/*
+ * 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.gravitino.integration.test.container;
+
+import static java.lang.String.format;
+import static org.awaitility.Awaitility.await;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Network;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+
+/**
+ * A RustFS container providing S3-compatible object storage. RustFS enforces
the session policy of
+ * an {@code AssumeRole} request, so it can exercise credential vending,
including the read-only
+ * downgrade, without a cloud account.
+ */
+public class RustFSContainer extends BaseContainer {
+ private static final Logger LOG =
LoggerFactory.getLogger(RustFSContainer.class);
+
+ /** The image pin consumed from the Docker Compose manifest tracked by image
updates. */
+ public static final String DEFAULT_IMAGE = loadImage();
+
+ /** The container hostname on the shared integration-test network. */
+ public static final String HOST_NAME = "gravitino-ci-rustfs";
+
+ /** The S3 and STS port inside the container. */
+ public static final int PORT = 9000;
+
+ /** Root access key used only by this local test fixture. */
+ public static final String ACCESS_KEY = "rustfsadmin";
+
+ /** Root secret key used only by this local test fixture. */
+ public static final String SECRET_KEY = "rustfsadmin123";
+
+ /** Region used when signing requests to the fixture. */
+ public static final String REGION = "us-east-1";
+
+ /**
+ * Well-formed role identifier for the SDK. RustFS scopes sessions by caller
and inline policy;
+ * this fixture does not emulate AWS named-role trust or external-ID
enforcement.
+ */
+ public static final String ROLE_ARN =
"arn:aws:iam::123456789012:role/gravitino-test";
+
+ /**
+ * Creates a builder with the default test configuration.
+ *
+ * @return the container builder
+ */
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ private RustFSContainer(
+ String image,
+ String hostName,
+ Set<Integer> ports,
+ Map<String, String> extraHosts,
+ Map<String, String> filesToMount,
+ Map<String, String> envVars,
+ Optional<Network> network) {
+ super(image, hostName, ports, extraHosts, filesToMount, envVars, network);
+ }
+
+ @Override
+ protected void setupContainer() {
+ super.setupContainer();
+ withLogConsumer(new PrintingContainerLog(format("%-14s| ",
"RustFSContainer")));
+ }
+
+ @Override
+ public void start() {
+ super.start();
+ checkContainerStatus(60);
+ }
+
+ @Override
+ protected boolean checkContainerStatus(int retryLimit) {
+ // A listening socket alone does not establish that IAM and the S3 API are
initialized.
+ try (S3Client client = createS3Client()) {
+ await()
+ .atMost(60, TimeUnit.SECONDS)
+ .pollInterval(Math.max(1, 60 / retryLimit), TimeUnit.SECONDS)
+ .until(
+ () -> {
+ try {
+ client.listBuckets();
+ return true;
+ } catch (Exception e) {
+ LOG.info("RustFS is not ready yet: {}", e.getMessage());
+ return false;
+ }
+ });
+ }
+ return true;
+ }
+
+ /**
+ * Creates a bucket, so that a warehouse location inside it can be used.
+ *
+ * @param bucketName the bucket to create
+ */
+ public void createBucket(String bucketName) {
+ try (S3Client client = createS3Client()) {
+ client.createBucket(request -> request.bucket(bucketName));
+ client.headBucket(request -> request.bucket(bucketName));
+ }
+ }
+
+ /**
+ * Returns the S3 and STS endpoint reachable from the host through the
published port.
+ *
+ * @return the endpoint, for example {@code http://localhost:32768}
+ */
+ public String getS3Endpoint() {
+ return format("http://%s:%d";, container.getHost(), getMappedPort(PORT));
+ }
+
+ static String imageFromCompose(String compose) {
+ // This single-service manifest is deliberately limited to one unquoted
image reference.
+ // Fail closed if its structure changes, rather than silently using a
different image pin.
+ Matcher matcher = Pattern.compile("(?m)^\\s+image:
(\\S+)\\s*$").matcher(compose);
+ if (!matcher.find()) {
+ throw new IllegalArgumentException("RustFS Compose manifest must contain
an image");
+ }
+ String image = matcher.group(1);
+ if (matcher.find() ||
!image.matches("rustfs/rustfs:[^@\\s]+@sha256:[0-9a-f]{64}")) {
+ throw new IllegalArgumentException(
+ "RustFS Compose manifest must contain one version and digest pin");
+ }
+ return image;
+ }
+
+ private static String loadImage() {
+ try (InputStream stream =
+
RustFSContainer.class.getResourceAsStream("/docker-compose-rustfs.yml")) {
+ if (stream == null) {
+ throw new IllegalStateException("Missing docker-compose-rustfs.yml
test resource");
+ }
+ return imageFromCompose(new String(stream.readAllBytes(),
StandardCharsets.UTF_8));
+ } catch (IOException e) {
+ throw new IllegalStateException("Unable to read RustFS image pin", e);
+ }
+ }
+
+ private S3Client createS3Client() {
+ return S3Client.builder()
+ .endpointOverride(URI.create(getS3Endpoint()))
+ .region(Region.of(REGION))
+ .forcePathStyle(true)
+ .credentialsProvider(
+
StaticCredentialsProvider.create(AwsBasicCredentials.create(ACCESS_KEY,
SECRET_KEY)))
+ .overrideConfiguration(
+ config ->
+ config
+ .apiCallTimeout(Duration.ofSeconds(5))
+ .apiCallAttemptTimeout(Duration.ofSeconds(3)))
+ .build();
+ }
+
+ /** Builder for {@link RustFSContainer}. */
+ public static class Builder
+ extends BaseContainer.Builder<RustFSContainer.Builder, RustFSContainer> {
+
+ private Builder() {
+ this.image = DEFAULT_IMAGE;
+ this.hostName = HOST_NAME;
+ this.exposePorts = ImmutableSet.of(PORT);
+ this.envVars =
+ ImmutableMap.of(
+ "RUSTFS_ACCESS_KEY", ACCESS_KEY,
+ "RUSTFS_SECRET_KEY", SECRET_KEY,
+ "RUSTFS_CONSOLE_ENABLE", "false");
+ }
+
+ @Override
+ public RustFSContainer build() {
+ return new RustFSContainer(
+ image, hostName, exposePorts, extraHosts, filesToMount, envVars,
network);
+ }
+ }
+}
diff --git
a/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/TestRustFSContainer.java
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/TestRustFSContainer.java
new file mode 100644
index 0000000000..50e0379a7c
--- /dev/null
+++
b/integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/TestRustFSContainer.java
@@ -0,0 +1,51 @@
+/*
+ * 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.gravitino.integration.test.container;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Tests that the fixture consumes exactly one versioned, immutable image
reference. */
+public class TestRustFSContainer {
+
+ @Test
+ void testImagePinFromPackagedResource() {
+ Assertions.assertTrue(
+
RustFSContainer.DEFAULT_IMAGE.matches("rustfs/rustfs:[^@\\s]+@sha256:[0-9a-f]{64}"));
+ }
+
+ @Test
+ void testImagePinParsing() {
+ String manifest = "services:\n rustfs:\n image: " +
RustFSContainer.DEFAULT_IMAGE + "\n";
+ Assertions.assertEquals(
+ RustFSContainer.DEFAULT_IMAGE,
RustFSContainer.imageFromCompose(manifest));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> RustFSContainer.imageFromCompose(manifest + " image:
another/image:latest\n"));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> RustFSContainer.imageFromCompose("services:\n rustfs:\n"));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> RustFSContainer.imageFromCompose(" image:
rustfs/rustfs:latest\n"));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> RustFSContainer.imageFromCompose(" image:
rustfs/rustfs:1.0.0@sha256:bad\n"));
+ }
+}
diff --git
a/integration-test-common/src/test/resources/docker-compose-rustfs.yml
b/integration-test-common/src/test/resources/docker-compose-rustfs.yml
new file mode 100644
index 0000000000..cad91610a2
--- /dev/null
+++ b/integration-test-common/src/test/resources/docker-compose-rustfs.yml
@@ -0,0 +1,22 @@
+# 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.
+
+# RustFSContainer reads this authoritative image pin from its test resources.
+# Keep a single, unquoted image entry so fixture and image monitoring agree.
+services:
+ rustfs:
+ image:
rustfs/rustfs:1.0.0-rc.6@sha256:97171b3d72cd47dc81000f92ea84de25608bfc35a94c965501afaeb5d99f6035
