From ff07ea334de8c676aed4532b426b2c02af2f178a Mon Sep 17 00:00:00 2001 From: kannar Date: Thu, 23 Jul 2026 16:27:03 +0200 Subject: [PATCH 1/2] [improve][offload] Support credentials from offload policies for S3 and Aliyun OSS drivers The S3 and aliyun-oss offload drivers only read credentials from the ACCESS_KEY_ID/ACCESS_KEY_SECRET (or ALIYUN_OSS_*) environment variables, ignoring the s3ManagedLedgerOffloadCredentialId/Secret properties that OffloadPoliciesImpl carries for every S3-compatible driver. Credentials set in broker.conf or in namespace/topic offload policies therefore never took effect for these drivers. S3_CREDENTIAL_BUILDER now uses the credential pair from the tiered storage configuration when both keys are set (blank values are treated as absent), throws a clear IllegalArgumentException when only one of the pair is supplied, and otherwise falls back to the environment variables unchanged. Assisted-by: Claude Code (Claude Fable 5) Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Tu1LLcX9KGPhnZygeWRZXQ --- .../provider/JCloudBlobStoreProvider.java | 19 ++++++++ .../provider/JCloudBlobStoreProviderTest.java | 46 +++++++++++++++++++ 2 files changed, 65 insertions(+) diff --git a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProvider.java b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProvider.java index 733f6af24a41e..b6ff0308137db 100644 --- a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProvider.java +++ b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProvider.java @@ -433,6 +433,25 @@ public String getAWSSecretKey() { }; static final CredentialBuilder S3_CREDENTIAL_BUILDER = (TieredStorageConfiguration config) -> { + if (config.getCredentials() != null) { + return; + } + // Credentials provided in the tiered storage configuration take + // precedence over the environment variables. Offload policies carry + // credentials under the s3-prefixed keys for every S3-compatible + // driver (see OffloadPoliciesImpl), so those are the keys accepted. + String configId = config.getConfigProperty(S3_ID_FIELD); + String configSecret = config.getConfigProperty(S3_SECRET_FIELD); + if (StringUtils.isNotBlank(configId) || StringUtils.isNotBlank(configSecret)) { + if (StringUtils.isBlank(configId) || StringUtils.isBlank(configSecret)) { + throw new IllegalArgumentException( + "Both " + S3_ID_FIELD + " and " + S3_SECRET_FIELD + + " must be set when providing offload credentials in the configuration"); + } + Credentials credentials = new Credentials(configId, configSecret); + config.setProviderCredentials(() -> credentials); + return; + } String accountName = System.getenv().getOrDefault("ACCESS_KEY_ID", ""); // For forward compatibility if (StringUtils.isEmpty(accountName.trim())) { diff --git a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProviderTest.java b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProviderTest.java index f45cb30ea4c2b..7ad2eef99474f 100644 --- a/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProviderTest.java +++ b/tiered-storage/jcloud/src/test/java/org/apache/bookkeeper/mledger/offload/jcloud/provider/JCloudBlobStoreProviderTest.java @@ -19,8 +19,10 @@ package org.apache.bookkeeper.mledger.offload.jcloud.provider; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNotNull; import java.util.HashMap; import java.util.Map; +import org.jclouds.domain.Credentials; import org.testng.annotations.Test; public class JCloudBlobStoreProviderTest { @@ -130,4 +132,48 @@ public void s3ValidationBucketMissed() { TieredStorageConfiguration configuration = new TieredStorageConfiguration(map); configuration.getProvider().validate(configuration); } + + // Offload policies (OffloadPoliciesImpl) carry credentials under the + // s3-prefixed keys for every S3-compatible driver + private TieredStorageConfiguration credentialConfig(String driver, String id, String secret) { + Map map = new HashMap<>(); + map.put("managedLedgerOffloadDriver", driver); + map.put("managedLedgerOffloadServiceEndpoint", "http://storage.service"); + map.put("managedLedgerOffloadBucket", "test-bucket"); + if (id != null) { + map.put("s3ManagedLedgerOffloadCredentialId", id); + } + if (secret != null) { + map.put("s3ManagedLedgerOffloadCredentialSecret", secret); + } + return new TieredStorageConfiguration(map); + } + + private void assertCredentialsFromConfig(String driver) { + TieredStorageConfiguration configuration = + credentialConfig(driver, "config-access-id", "config-access-secret"); + configuration.getProvider().buildCredentials(configuration); + assertNotNull(configuration.getProviderCredentials()); + Credentials credentials = configuration.getProviderCredentials().get(); + assertEquals(credentials.identity, "config-access-id"); + assertEquals(credentials.credential, "config-access-secret"); + } + + @Test + public void s3CredentialsFromConfigTest() { + assertCredentialsFromConfig("S3"); + } + + @Test + public void aliyunOssCredentialsFromConfigTest() { + assertCredentialsFromConfig("aliyun-oss"); + } + + @Test(expectedExceptions = IllegalArgumentException.class, + expectedExceptionsMessageRegExp = "Both s3ManagedLedgerOffloadCredentialId and " + + "s3ManagedLedgerOffloadCredentialSecret must be set.*") + public void s3PartialCredentialsFromConfigTest() { + TieredStorageConfiguration configuration = credentialConfig("S3", "config-access-id", null); + configuration.getProvider().buildCredentials(configuration); + } } From 7941ebdd64f9c6054349f7d3c585ac906fad9d27 Mon Sep 17 00:00:00 2001 From: kannar Date: Thu, 23 Jul 2026 22:09:59 +0200 Subject: [PATCH 2/2] [improve][offload] Carry offload-policy credentials for aliyun-oss via the admin CLI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address PR review feedback: OffloadPoliciesImpl.create(driver, ...) — the path used by `pulsar-admin namespaces/topics set-offload-policies` — only copied credentialId/credentialSecret into the s3-prefixed fields for the `S3` and `aws-s3` drivers. For `-d aliyun-oss` the CLI silently dropped the credentials before they reached the offloader, so the new config-provided credential support was only reachable for Aliyun via broker.conf or the admin API. aliyun-oss is S3-compatible (shares S3_CREDENTIAL_BUILDER and reads the s3ManagedLedgerOffload* keys), so extend the driver check in create() and isS3Driver() to include it. This also makes bucketValid() consistently check the s3-prefixed bucket for aliyun-oss. Also refactor driver-name handling per review: introduce a named constant for each built-in driver (DRIVER_S3, DRIVER_AWS_S3, DRIVER_GOOGLE_CLOUD_STORAGE, DRIVER_FILESYSTEM, DRIVER_AZUREBLOB, DRIVER_ALIYUN_OSS), build INTERNAL_SUPPORTED_DRIVER from them, and replace the index-based DRIVER_NAMES.get(N) lookups in create(), isS3Driver(), isGcsDriver() and isFileSystemDriver() with the constants. Assisted-by: Claude Code (Claude Opus 4.8 (1M context)) Co-Authored-By: Claude Opus 4.8 (1M context) Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01SFGLyfqZmcYZgWkBV5MkFs --- .../policies/data/OffloadPoliciesImpl.java | 24 ++++++---- .../policies/data/OffloadPoliciesTest.java | 46 +++++++++++++++++++ 2 files changed, 62 insertions(+), 8 deletions(-) diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPoliciesImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPoliciesImpl.java index 14f89ec156e41..acf32e0066646 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPoliciesImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPoliciesImpl.java @@ -63,8 +63,14 @@ public class OffloadPoliciesImpl implements Serializable, OffloadPolicies { CONFIGURATION_FIELDS = Collections.unmodifiableList(temp); } - public static final List INTERNAL_SUPPORTED_DRIVER = List.of("S3", - "aws-s3", "google-cloud-storage", "filesystem", "azureblob", "aliyun-oss"); + public static final String DRIVER_S3 = "S3"; + public static final String DRIVER_AWS_S3 = "aws-s3"; + public static final String DRIVER_GOOGLE_CLOUD_STORAGE = "google-cloud-storage"; + public static final String DRIVER_FILESYSTEM = "filesystem"; + public static final String DRIVER_AZUREBLOB = "azureblob"; + public static final String DRIVER_ALIYUN_OSS = "aliyun-oss"; + public static final List INTERNAL_SUPPORTED_DRIVER = List.of(DRIVER_S3, + DRIVER_AWS_S3, DRIVER_GOOGLE_CLOUD_STORAGE, DRIVER_FILESYSTEM, DRIVER_AZUREBLOB, DRIVER_ALIYUN_OSS); public static final List DRIVER_NAMES; static { String extraDrivers = System.getProperty("pulsar.extra.offload.drivers", ""); @@ -221,7 +227,8 @@ public static OffloadPoliciesImpl create(String driver, String region, String bu .managedLedgerOffloadReadBufferSizeInBytes(readBufferSizeInBytes) .managedLedgerOffloadedReadPriority(readPriority); - if (driver.equalsIgnoreCase(DRIVER_NAMES.get(0)) || driver.equalsIgnoreCase(DRIVER_NAMES.get(1))) { + if (driver.equalsIgnoreCase(DRIVER_S3) || driver.equalsIgnoreCase(DRIVER_AWS_S3) + || driver.equalsIgnoreCase(DRIVER_ALIYUN_OSS)) { if (role != null) { builder.s3ManagedLedgerOffloadRole(role); } @@ -240,7 +247,7 @@ public static OffloadPoliciesImpl create(String driver, String region, String bu .s3ManagedLedgerOffloadServiceEndpoint(endpoint) .s3ManagedLedgerOffloadMaxBlockSizeInBytes(maxBlockSizeInBytes) .s3ManagedLedgerOffloadReadBufferSizeInBytes(readBufferSizeInBytes); - } else if (driver.equalsIgnoreCase(DRIVER_NAMES.get(2))) { + } else if (driver.equalsIgnoreCase(DRIVER_GOOGLE_CLOUD_STORAGE)) { builder.gcsManagedLedgerOffloadRegion(region) .gcsManagedLedgerOffloadBucket(bucket) .gcsManagedLedgerOffloadMaxBlockSizeInBytes(maxBlockSizeInBytes) @@ -310,22 +317,23 @@ public boolean isS3Driver() { if (managedLedgerOffloadDriver == null) { return false; } - return managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_NAMES.get(0)) - || managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_NAMES.get(1)); + return managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_S3) + || managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_AWS_S3) + || managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_ALIYUN_OSS); } public boolean isGcsDriver() { if (managedLedgerOffloadDriver == null) { return false; } - return managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_NAMES.get(2)); + return managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_GOOGLE_CLOUD_STORAGE); } public boolean isFileSystemDriver() { if (managedLedgerOffloadDriver == null) { return false; } - return managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_NAMES.get(3)); + return managedLedgerOffloadDriver.equalsIgnoreCase(DRIVER_FILESYSTEM); } public boolean bucketValid() { diff --git a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/OffloadPoliciesTest.java b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/OffloadPoliciesTest.java index 6755cbec9a93f..56d16061a5305 100644 --- a/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/OffloadPoliciesTest.java +++ b/pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/OffloadPoliciesTest.java @@ -96,6 +96,52 @@ public void testS3Configuration() { Assert.assertEquals(offloadPolicies.getManagedLedgerOffloadThresholdInSeconds(), offloadThresholdInSeconds); } + @Test + public void testAliyunOssConfiguration() { + final String driver = "aliyun-oss"; + final String region = "test-region"; + final String bucket = "test-bucket"; + final String role = "test-role"; + final String roleSessionName = "test-role-session-name"; + final String credentialId = "test-credential-id"; + final String credentialSecret = "test-credential-secret"; + final String endPoint = "test-endpoint"; + final Integer maxBlockSizeInBytes = 5 * M; + final Integer readBufferSizeInBytes = 2 * M; + final Long offloadThresholdInBytes = 10L * M; + final Long offloadThresholdInSeconds = 1000L; + final Long offloadDeletionLagInMillis = 5L * MIN; + + OffloadPoliciesImpl offloadPolicies = OffloadPoliciesImpl.create( + driver, + region, + bucket, + endPoint, + role, + roleSessionName, + credentialId, + credentialSecret, + maxBlockSizeInBytes, + readBufferSizeInBytes, + offloadThresholdInBytes, + offloadThresholdInSeconds, + offloadDeletionLagInMillis, + OffloadedReadPriority.TIERED_STORAGE_FIRST + ); + + Assert.assertTrue(offloadPolicies.isS3Driver()); + Assert.assertEquals(offloadPolicies.getManagedLedgerOffloadDriver(), driver); + Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadRegion(), region); + Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadBucket(), bucket); + Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadServiceEndpoint(), endPoint); + // The CLI create() path must carry the credentials under the s3-prefixed keys the + // S3-compatible aliyun-oss offloader reads, not silently drop them. + Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadCredentialId(), credentialId); + Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadCredentialSecret(), credentialSecret); + Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadMaxBlockSizeInBytes(), maxBlockSizeInBytes); + Assert.assertEquals(offloadPolicies.getS3ManagedLedgerOffloadReadBufferSizeInBytes(), readBufferSizeInBytes); + } + @Test public void testGcsConfiguration() { final String driver = "google-cloud-storage";