This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 42e885f96783 feat(kinesis): Cross-account Kinesis source support via
STS assume-role in JsonKinesisSource (#19383)
42e885f96783 is described below
commit 42e885f9678321711e0d1563b1f651efac5794f5
Author: Surya Dhaneshwar <[email protected]>
AuthorDate: Tue Aug 18 05:30:38 2026 -0700
feat(kinesis): Cross-account Kinesis source support via STS assume-role in
JsonKinesisSource (#19383)
* Cross-account Kinesis: assume-role support in JsonKinesisSource
Kinesis has no resource-based policy, so reading a stream in a different AWS
account requires IAM role assumption (sts:AssumeRole). Add an optional
role-ARN
config (hoodie.streamer.source.kinesis.role.arn) that, when set, builds the
KinesisClient with an STS assume-role credentials provider (AWS SDK v2). The
base STS client uses the default credential chain; static access/secret keys
still take precedence. Empty/absent preserves the existing same-account
behavior. No external ID is used.
The assume-role provider is cached per region|roleArn so a long-lived
streaming
executor reuses one StsClient per distinct role instead of leaking one per
mapPartitions task (AWS SDK v2 does not close a user-supplied credentials
provider when the KinesisClient closes).
Adds software.amazon.awssdk:sts to hudi-utilities and unit coverage for the
config-key contract, ARN serialization to executors, and the credential-
provider branch selection.
* review(19383): add external id/session name, non-tautological tests,
javadoc fixes
Addresses the review findings on the cross-account Kinesis assume-role
support:
- KinesisSourceConfig: add hoodie.streamer.source.kinesis.role.external.id
(no default) and
role.session.name (default "hudi-kinesis-source"), mirroring
HoodieAWSConfig; the same gap
HUDI-7699 (#11134) closed for the Glue assume-role provider.
KINESIS_ROLE_ARN moves to
sinceVersion 1.3.0 and its doc now states the static-key precedence and
that endpoint.url
applies to Kinesis only (STS always uses the regional endpoint).
- KinesisOffsetGen: thread external id + session name into the
AssumeRoleRequest and the
provider cache key; rename assumeRoleProvider to
getOrCreateAssumeRoleProvider; correct the
javadoc (AWS SDK v2 does close a user-supplied provider on client close;
the cache is safe
only because StsAssumeRoleCredentialsProvider.close() never closes the
injected StsClient,
so asyncCredentialUpdateEnabled must stay off); ASCII-only comments.
- KinesisReadConfig / JsonKinesisSource: carry the two new fields to
executors; extract the
read-config construction into a package-private buildReadConfig for
testability.
- Tests: TestKinesisOffsetGenClient now asserts the selected credentials
provider via
serviceClientConfiguration().credentialsProvider() (previously
assertNotNull(client), which
passed with the roleArn branch deleted) and pins the per-role/external-id
provider cache;
TestJsonKinesisSource gains testBuildReadConfigCarriesCredentialProps
covering the props to
KinesisReadConfig plumbing; TestKinesisReadConfig (JDK/Lombok/constant
checks) is dropped.
- pom.xml: correct the sts dependency comment (already transitive via
hudi-aws).
* review(19383): make the STS request testable, normalize blank role configs
Round-2 review follow-ups on the cross-account Kinesis assume-role support:
- KinesisOffsetGen: extract the AssumeRoleRequest into a package-private
buildAssumeRoleRequest so
role ARN, external id and session name are asserted directly (the
provider hides its request);
treat blank role.external.id / role.session.name as unset / default
instead of sending "" to STS
and failing lazily on an executor; cache key uses "" for a missing
external id.
- TestKinesisOffsetGenClient: cache test now varies the session name and
closes the first client
before asserting reuse; the props test uses the cache as oracle for the
full role tuple; new
tests for the request fields and the blank-value normalization.
---------
Co-authored-by: voon <[email protected]>
---
hudi-utilities/pom.xml | 7 +
.../hudi/utilities/config/KinesisSourceConfig.java | 30 +++++
.../hudi/utilities/sources/JsonKinesisSource.java | 22 +++-
.../sources/helpers/KinesisOffsetGen.java | 66 +++++++++-
.../sources/helpers/KinesisReadConfig.java | 3 +
.../utilities/sources/TestJsonKinesisSource.java | 40 ++++++
.../helpers/TestKinesisOffsetGenClient.java | 145 +++++++++++++++++++++
7 files changed, 306 insertions(+), 7 deletions(-)
diff --git a/hudi-utilities/pom.xml b/hudi-utilities/pom.xml
index 05906ac8580d..35496f4051f2 100644
--- a/hudi-utilities/pom.xml
+++ b/hudi-utilities/pom.xml
@@ -571,6 +571,13 @@
<artifactId>kinesis</artifactId>
<version>${aws.sdk.version}</version>
</dependency>
+ <!-- STS: assume-role credentials for cross-account Kinesis reads. Already
on the classpath transitively
+ via hudi-aws; declared directly because KinesisOffsetGen references
StsClient itself. -->
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>sts</artifactId>
+ <version>${aws.sdk.version}</version>
+ </dependency>
<!-- KPL de-aggregation: extracts user records from Kinesis Producer
Library aggregated records -->
<dependency>
<groupId>com.amazonaws</groupId>
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/KinesisSourceConfig.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/KinesisSourceConfig.java
index 1ad17d5efc71..a9f3fd4688f9 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/KinesisSourceConfig.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/config/KinesisSourceConfig.java
@@ -76,6 +76,36 @@ public class KinesisSourceConfig extends HoodieConfig {
.withDocumentation("AWS secret key for Kinesis. Used when connecting to
custom endpoints (e.g., LocalStack). "
+ "If not set with endpoint, uses the default AWS credential
chain.");
+ public static final ConfigProperty<String> KINESIS_ROLE_ARN = ConfigProperty
+ .key(PREFIX + "role.arn")
+ .noDefaultValue()
+ .sinceVersion("1.3.0")
+ .markAdvanced()
+ .withDocumentation("IAM role ARN to assume via STS when the Kinesis
stream lives in a different AWS "
+ + "account than the application. When set, and no static
access/secret key is configured (static keys "
+ + "take precedence), the Kinesis client uses an auto-refreshing
StsAssumeRoleCredentialsProvider whose "
+ + "base credentials come from the default credential chain, which
must be allowed sts:AssumeRole on this "
+ + "role. When empty/absent, the stream is read with the default
credential chain (same-account behavior). "
+ + "The STS call always goes to the regional STS endpoint of " +
PREFIX + "region; " + PREFIX
+ + "endpoint.url applies to Kinesis only. See also " + PREFIX +
"role.external.id and " + PREFIX
+ + "role.session.name.");
+
+ public static final ConfigProperty<String> KINESIS_ROLE_EXTERNAL_ID =
ConfigProperty
+ .key(PREFIX + "role.external.id")
+ .noDefaultValue()
+ .sinceVersion("1.3.0")
+ .markAdvanced()
+ .withDocumentation("External ID passed to sts:AssumeRole when assuming "
+ PREFIX + "role.arn, for roles "
+ + "whose trust policy carries an sts:ExternalId condition. Ignored
when the role ARN is not set.");
+
+ public static final ConfigProperty<String> KINESIS_ROLE_SESSION_NAME =
ConfigProperty
+ .key(PREFIX + "role.session.name")
+ .defaultValue("hudi-kinesis-source")
+ .sinceVersion("1.3.0")
+ .markAdvanced()
+ .withDocumentation("STS role session name used when assuming " + PREFIX
+ "role.arn. Shows up in the "
+ + "target account's CloudTrail, so set it per pipeline when several
jobs share one role.");
+
public static final ConfigProperty<Long> MAX_EVENTS_FROM_KINESIS_SOURCE =
ConfigProperty
.key(PREFIX + "max.events")
.defaultValue(5000000L)
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKinesisSource.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKinesisSource.java
index 3a297467c82e..42e850efc78b 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKinesisSource.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/JsonKinesisSource.java
@@ -20,6 +20,7 @@ package org.apache.hudi.utilities.sources;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.VisibleForTesting;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.utilities.config.KinesisSourceConfig;
import org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics;
@@ -108,15 +109,22 @@ public class JsonKinesisSource extends
KinesisSource<JavaRDD<String>> {
this.offsetGen = new KinesisOffsetGen(props);
}
- @Override
- protected JavaRDD<String> toBatch(KinesisOffsetGen.KinesisShardRange[]
shardRanges, long sourceLimit) {
+ /**
+ * Builds the serializable per-batch read config shipped to executors; the
Kinesis client is rebuilt from
+ * it inside {@code mapPartitions}, so every credential-related property
must be carried here.
+ */
+ @VisibleForTesting
+ KinesisReadConfig buildReadConfig(KinesisOffsetGen.KinesisShardRange[]
shardRanges, long sourceLimit) {
long numEvents = calculateNumEvents(sourceLimit, props);
- KinesisReadConfig readConfig = new KinesisReadConfig(
+ return new KinesisReadConfig(
offsetGen.getStreamName(),
offsetGen.getRegion(),
offsetGen.getEndpointUrl().orElse(null),
getStringWithAltKeys(props, KinesisSourceConfig.KINESIS_ACCESS_KEY,
null),
getStringWithAltKeys(props, KinesisSourceConfig.KINESIS_SECRET_KEY,
null),
+ getStringWithAltKeys(props, KinesisSourceConfig.KINESIS_ROLE_ARN,
null),
+ getStringWithAltKeys(props,
KinesisSourceConfig.KINESIS_ROLE_EXTERNAL_ID, null),
+ getStringWithAltKeys(props,
KinesisSourceConfig.KINESIS_ROLE_SESSION_NAME, true),
offsetGen.getStartingPositionStrategy(),
shouldAddMetaFields,
getBooleanWithAltKeys(props,
KinesisSourceConfig.KINESIS_ENABLE_DEAGGREGATION),
@@ -127,6 +135,11 @@ public class JsonKinesisSource extends
KinesisSource<JavaRDD<String>> {
getLongWithAltKeys(props,
KinesisSourceConfig.KINESIS_RETRY_INITIAL_INTERVAL_MS),
getLongWithAltKeys(props,
KinesisSourceConfig.KINESIS_RETRY_MAX_INTERVAL_MS),
getLongWithAltKeys(props,
KinesisSourceConfig.KINESIS_THROTTLE_TIMEOUT_MS));
+ }
+
+ @Override
+ protected JavaRDD<String> toBatch(KinesisOffsetGen.KinesisShardRange[]
shardRanges, long sourceLimit) {
+ KinesisReadConfig readConfig = buildReadConfig(shardRanges, sourceLimit);
JavaRDD<ShardFetchResult> fetchRdd = sparkContext.parallelize(
java.util.Arrays.asList(shardRanges), shardRanges.length)
@@ -134,7 +147,8 @@ public class JsonKinesisSource extends
KinesisSource<JavaRDD<String>> {
List<ShardFetchResult> results = new ArrayList<>();
try (KinesisClient client = KinesisOffsetGen.createKinesisClient(
readConfig.getRegion(), readConfig.getEndpointUrl(),
- readConfig.getAccessKey(), readConfig.getSecretKey())) {
+ readConfig.getAccessKey(), readConfig.getSecretKey(),
+ readConfig.getRoleArn(), readConfig.getRoleExternalId(),
readConfig.getRoleSessionName())) {
while (shardRangeIt.hasNext()) {
KinesisOffsetGen.KinesisShardRange range = shardRangeIt.next();
// Lazy iterator: fetches one GetRecords page at a time, keeping
only one page in
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisOffsetGen.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisOffsetGen.java
index 27d160c3de0f..9085e2522445 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisOffsetGen.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisOffsetGen.java
@@ -21,6 +21,7 @@ package org.apache.hudi.utilities.sources.helpers;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.table.checkpoint.Checkpoint;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.VisibleForTesting;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.utilities.config.KinesisSourceConfig;
@@ -41,6 +42,9 @@ import
software.amazon.awssdk.services.kinesis.model.ListShardsResponse;
import
software.amazon.awssdk.services.kinesis.model.ProvisionedThroughputExceededException;
import software.amazon.awssdk.services.kinesis.model.ResourceNotFoundException;
import software.amazon.awssdk.services.kinesis.model.Shard;
+import software.amazon.awssdk.services.sts.StsClient;
+import
software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
+import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
import java.math.BigInteger;
import java.net.URI;
@@ -50,6 +54,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
@@ -304,20 +309,71 @@ public class KinesisOffsetGen {
getStringWithAltKeys(props,
KinesisSourceConfig.KINESIS_STARTING_POSITION, true));
}
+ /**
+ * Assume-role credentials providers cached by {@code
region|roleArn|externalId|sessionName}. Kinesis
+ * clients are built per {@code mapPartitions} task, so building a fresh
StsClient (with its own HTTP
+ * connection pool) each time would pile one up per micro-batch on a
long-lived streaming executor. Note
+ * that AWS SDK v2 DOES close a user-supplied credentials provider when the
KinesisClient closes, but
+ * StsAssumeRoleCredentialsProvider.close() only stops background prefetch
(a no-op with the default
+ * synchronous refresh strategy) and never closes the injected StsClient, so
a cached provider stays
+ * usable across clients. For that reason asyncCredentialUpdateEnabled must
never be turned on for this
+ * shared provider: the first client close would permanently disable its
background refresh. Providers
+ * live for the JVM lifetime and are intentionally never closed.
+ */
+ private static final Map<String, StsAssumeRoleCredentialsProvider>
ASSUME_ROLE_PROVIDERS =
+ new ConcurrentHashMap<>();
+
+ private static StsAssumeRoleCredentialsProvider
getOrCreateAssumeRoleProvider(
+ String region, String roleArn, String roleExternalId, String
roleSessionName) {
+ String cacheKey = String.join("|", region, roleArn, roleExternalId == null
? "" : roleExternalId, roleSessionName);
+ return ASSUME_ROLE_PROVIDERS.computeIfAbsent(cacheKey, ignored ->
+ StsAssumeRoleCredentialsProvider.builder()
+ // Regional STS endpoint of the stream's region; the Kinesis
endpoint.url override is
+ // deliberately not applied here (it may be a Kinesis-only VPC
interface endpoint).
+ .stsClient(StsClient.builder().region(Region.of(region)).build())
+ .refreshRequest(buildAssumeRoleRequest(roleArn, roleExternalId,
roleSessionName))
+ .build());
+ }
+
+ /**
+ * The STS request behind the cached provider; a null external id is omitted
from the request.
+ */
+ @VisibleForTesting
+ static AssumeRoleRequest buildAssumeRoleRequest(String roleArn, String
roleExternalId, String roleSessionName) {
+ return AssumeRoleRequest.builder()
+ .roleArn(roleArn)
+ .roleSessionName(roleSessionName)
+ .externalId(roleExternalId)
+ .build();
+ }
+
/**
* Builds a Kinesis client from explicit parameters. Used by both the
instance method
* {@link #createKinesisClient()} and by {@link
org.apache.hudi.utilities.sources.JsonKinesisSource}
- * from serializable {@link KinesisReadConfig} in Spark closures.
+ * from serializable {@link KinesisReadConfig} in Spark closures. Credential
precedence: static
+ * access/secret keys, then STS assume-role when {@code roleArn} is set,
else the default chain.
*/
public static KinesisClient createKinesisClient(String region, String
endpointUrl,
- String accessKey, String secretKey) {
+ String accessKey, String secretKey, String roleArn, String
roleExternalId, String roleSessionName) {
KinesisClientBuilder builder =
KinesisClient.builder().region(Region.of(region));
if (endpointUrl != null && !endpointUrl.isEmpty()) {
builder = builder.endpointOverride(URI.create(endpointUrl));
}
if (accessKey != null && !accessKey.isEmpty() && secretKey != null &&
!secretKey.isEmpty()) {
+ // Static credentials (e.g. LocalStack / custom endpoint) take
precedence.
builder = builder.credentialsProvider(
StaticCredentialsProvider.create(AwsBasicCredentials.create(accessKey,
secretKey)));
+ } else if (roleArn != null && !roleArn.isEmpty()) {
+ // Cross-account stream: assume the role via STS with a per-JVM cached
provider (see
+ // ASSUME_ROLE_PROVIDERS). The base STS client uses the default
credential chain, which must be
+ // granted sts:AssumeRole on this ARN. Blank values (e.g.
"role.external.id=" in a templated
+ // properties file) would be sent to STS verbatim and rejected lazily on
the first request, so
+ // treat them as unset.
+ String externalId = StringUtils.isNullOrEmpty(roleExternalId) ? null :
roleExternalId;
+ String sessionName = StringUtils.isNullOrEmpty(roleSessionName)
+ ? KinesisSourceConfig.KINESIS_ROLE_SESSION_NAME.defaultValue() :
roleSessionName;
+ builder = builder.credentialsProvider(
+ getOrCreateAssumeRoleProvider(region, roleArn, externalId,
sessionName));
}
return builder.build();
}
@@ -325,7 +381,11 @@ public class KinesisOffsetGen {
public KinesisClient createKinesisClient() {
String accessKey = getStringWithAltKeys(props,
KinesisSourceConfig.KINESIS_ACCESS_KEY, null);
String secretKey = getStringWithAltKeys(props,
KinesisSourceConfig.KINESIS_SECRET_KEY, null);
- return createKinesisClient(region, endpointUrl.orElse(null), accessKey,
secretKey);
+ String roleArn = getStringWithAltKeys(props,
KinesisSourceConfig.KINESIS_ROLE_ARN, null);
+ String roleExternalId = getStringWithAltKeys(props,
KinesisSourceConfig.KINESIS_ROLE_EXTERNAL_ID, null);
+ String roleSessionName = getStringWithAltKeys(props,
KinesisSourceConfig.KINESIS_ROLE_SESSION_NAME, true);
+ return createKinesisClient(region, endpointUrl.orElse(null), accessKey,
secretKey,
+ roleArn, roleExternalId, roleSessionName);
}
/**
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisReadConfig.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisReadConfig.java
index 7fe53f24d0f9..c1331da46503 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisReadConfig.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisReadConfig.java
@@ -40,6 +40,9 @@ public class KinesisReadConfig implements Serializable {
private final String endpointUrl; // null if not set
private final String accessKey; // null if not set
private final String secretKey; // null if not set
+ private final String roleArn; // null if not set; cross-account stream reads
via STS assume-role
+ private final String roleExternalId; // null if not set; sts:ExternalId for
the assumed role
+ private final String roleSessionName; // STS session name for the assumed
role
private final KinesisSourceConfig.KinesisStartingPositionStrategy
startingPosition;
private final boolean metaFieldsEnabled;
private final boolean deaggregationEnabled;
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKinesisSource.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKinesisSource.java
index 25450dd7c8a1..cce8c3af3aae 100644
---
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKinesisSource.java
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestJsonKinesisSource.java
@@ -21,9 +21,11 @@ package org.apache.hudi.utilities.sources;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
+import org.apache.hudi.utilities.config.KinesisSourceConfig;
import org.apache.hudi.utilities.ingestion.HoodieIngestionMetrics;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.sources.helpers.KinesisOffsetGen;
+import org.apache.hudi.utilities.sources.helpers.KinesisReadConfig;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -446,6 +448,44 @@ class TestJsonKinesisSource extends
SparkClientFunctionalTestHarness {
"Unread shard should not have arrival time in checkpoint");
}
+ // --- buildReadConfig tests ---
+
+ @Test
+ void testBuildReadConfigCarriesCredentialProps() {
+ TypedProperties props = new TypedProperties();
+ props.setProperty(KINESIS_STREAM_NAME.key(), STREAM_NAME);
+ props.setProperty(KINESIS_REGION.key(), "us-east-1");
+ props.setProperty(KINESIS_STARTING_POSITION.key(), "TRIM_HORIZON");
+ props.setProperty(KinesisSourceConfig.KINESIS_ENDPOINT_URL.key(),
"http://localhost:4566";);
+ props.setProperty(KinesisSourceConfig.KINESIS_ACCESS_KEY.key(),
"access-1");
+ props.setProperty(KinesisSourceConfig.KINESIS_SECRET_KEY.key(),
"secret-1");
+ props.setProperty(KinesisSourceConfig.KINESIS_ROLE_ARN.key(),
"arn:aws:iam::123456789012:role/reader");
+ props.setProperty(KinesisSourceConfig.KINESIS_ROLE_EXTERNAL_ID.key(),
"ext-1");
+ props.setProperty(KinesisSourceConfig.KINESIS_ROLE_SESSION_NAME.key(),
"session-1");
+ TestableJsonKinesisSource withCreds = new TestableJsonKinesisSource(
+ props, jsc(), spark(), null, mock(HoodieIngestionMetrics.class));
+
+ KinesisReadConfig readConfig = withCreds.buildReadConfig(new
KinesisOffsetGen.KinesisShardRange[0], 1000L);
+
+ assertEquals(STREAM_NAME, readConfig.getStreamName());
+ assertEquals("us-east-1", readConfig.getRegion());
+ assertEquals("http://localhost:4566";, readConfig.getEndpointUrl());
+ assertEquals("access-1", readConfig.getAccessKey());
+ assertEquals("secret-1", readConfig.getSecretKey());
+ assertEquals("arn:aws:iam::123456789012:role/reader",
readConfig.getRoleArn());
+ assertEquals("ext-1", readConfig.getRoleExternalId());
+ assertEquals("session-1", readConfig.getRoleSessionName());
+
+ // Defaults: no credentials configured means nulls, and the session name
falls back to the config default.
+ KinesisReadConfig defaults = source.buildReadConfig(new
KinesisOffsetGen.KinesisShardRange[0], 1000L);
+ assertNull(defaults.getEndpointUrl());
+ assertNull(defaults.getAccessKey());
+ assertNull(defaults.getSecretKey());
+ assertNull(defaults.getRoleArn());
+ assertNull(defaults.getRoleExternalId());
+ assertEquals(KinesisSourceConfig.KINESIS_ROLE_SESSION_NAME.defaultValue(),
defaults.getRoleSessionName());
+ }
+
private JavaRDD<String> emptyRdd() {
return jsc().emptyRDD();
}
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKinesisOffsetGenClient.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKinesisOffsetGenClient.java
new file mode 100644
index 000000000000..87ec2d99e648
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKinesisOffsetGenClient.java
@@ -0,0 +1,145 @@
+/*
+ * 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 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.hudi.utilities.sources.helpers;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.utilities.config.KinesisSourceConfig;
+
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.identity.spi.IdentityProvider;
+import software.amazon.awssdk.services.kinesis.KinesisClient;
+import
software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
+import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+/**
+ * Covers the credential-provider branches of {@link
KinesisOffsetGen#createKinesisClient}, the per-JVM
+ * assume-role provider cache and the STS request it builds. AWS SDK v2
clients resolve credentials
+ * lazily (only on the first request), so a client can be built for each
branch without any AWS
+ * environment or network access, and the chosen provider is observable through
+ * {@code serviceClientConfiguration().credentialsProvider()}.
+ */
+class TestKinesisOffsetGenClient {
+
+ private static final String REGION = "us-west-2";
+ private static final String ROLE_ARN_PREFIX =
"arn:aws:iam::123456789012:role/kinesis-cross-account-";
+ private static final String SESSION_NAME = "hudi-kinesis-source";
+
+ private static IdentityProvider<?> providerOf(KinesisClient client) {
+ return client.serviceClientConfiguration().credentialsProvider();
+ }
+
+ @Test
+ void testAssumeRoleProviderUsedWhenRoleArnPresent() {
+ try (KinesisClient client = KinesisOffsetGen.createKinesisClient(
+ REGION, null, null, null, ROLE_ARN_PREFIX + "arn-only", null,
SESSION_NAME)) {
+ assertInstanceOf(StsAssumeRoleCredentialsProvider.class,
providerOf(client));
+ }
+ }
+
+ @Test
+ void testDefaultChainUsedWhenNoRoleArnOrKeys() {
+ try (KinesisClient client = KinesisOffsetGen.createKinesisClient(
+ REGION, null, null, null, null, null, SESSION_NAME)) {
+ assertInstanceOf(DefaultCredentialsProvider.class, providerOf(client));
+ }
+ }
+
+ @Test
+ void testStaticKeysTakePrecedenceOverAssumeRole() {
+ // Both static keys and a role ARN set: the static-credentials branch wins
(no STS assume-role).
+ try (KinesisClient client = KinesisOffsetGen.createKinesisClient(
+ REGION, null, "access", "secret", ROLE_ARN_PREFIX + "with-keys", null,
SESSION_NAME)) {
+ assertInstanceOf(StaticCredentialsProvider.class, providerOf(client));
+ }
+ }
+
+ @Test
+ void testAssumeRoleProviderCachedPerRoleExternalIdAndSession() {
+ String roleArn = ROLE_ARN_PREFIX + "cached";
+ IdentityProvider<?> firstProvider;
+ try (KinesisClient first = KinesisOffsetGen.createKinesisClient(
+ REGION, null, null, null, roleArn, "ext-1", SESSION_NAME)) {
+ firstProvider = providerOf(first);
+ }
+ // The first client is closed above; the cached provider must still be
handed to later clients with
+ // the same role tuple, and any change to external id, session name or
role must yield a new one.
+ try (KinesisClient second = KinesisOffsetGen.createKinesisClient(
+ REGION, null, null, null, roleArn, "ext-1", SESSION_NAME);
+ KinesisClient otherExternalId = KinesisOffsetGen.createKinesisClient(
+ REGION, null, null, null, roleArn, "ext-2", SESSION_NAME);
+ KinesisClient otherSession = KinesisOffsetGen.createKinesisClient(
+ REGION, null, null, null, roleArn, "ext-1", "other-session");
+ KinesisClient otherRole = KinesisOffsetGen.createKinesisClient(
+ REGION, null, null, null, ROLE_ARN_PREFIX + "cached-other",
"ext-1", SESSION_NAME)) {
+ assertSame(firstProvider, providerOf(second));
+ assertNotSame(firstProvider, providerOf(otherExternalId));
+ assertNotSame(firstProvider, providerOf(otherSession));
+ assertNotSame(firstProvider, providerOf(otherRole));
+ }
+ }
+
+ @Test
+ void testInstanceClientForwardsRoleTupleFromProps() {
+ String roleArn = ROLE_ARN_PREFIX + "from-props";
+ TypedProperties props = new TypedProperties();
+ props.setProperty(KinesisSourceConfig.KINESIS_STREAM_NAME.key(),
"test-stream");
+ props.setProperty(KinesisSourceConfig.KINESIS_REGION.key(), REGION);
+ props.setProperty(KinesisSourceConfig.KINESIS_ROLE_ARN.key(), roleArn);
+ props.setProperty(KinesisSourceConfig.KINESIS_ROLE_EXTERNAL_ID.key(),
"ext-props");
+
+ try (KinesisClient fromProps = new
KinesisOffsetGen(props).createKinesisClient();
+ KinesisClient expected = KinesisOffsetGen.createKinesisClient(REGION,
null, null, null,
+ roleArn, "ext-props",
KinesisSourceConfig.KINESIS_ROLE_SESSION_NAME.defaultValue())) {
+ // The cache is keyed by the full role tuple, so provider identity
proves that role ARN, external
+ // id and the defaulted session name all reached the client in the right
slots.
+ assertSame(providerOf(expected), providerOf(fromProps));
+ }
+ }
+
+ @Test
+ void testAssumeRoleRequestCarriesRoleExternalIdAndSession() {
+ String roleArn = ROLE_ARN_PREFIX + "request";
+ AssumeRoleRequest request =
KinesisOffsetGen.buildAssumeRoleRequest(roleArn, "ext-req", "sess-req");
+ assertEquals(roleArn, request.roleArn());
+ assertEquals("ext-req", request.externalId());
+ assertEquals("sess-req", request.roleSessionName());
+ // No external id: omitted from the request rather than sent as an empty
string.
+ assertNull(KinesisOffsetGen.buildAssumeRoleRequest(roleArn, null,
"sess-req").externalId());
+ }
+
+ @Test
+ void testBlankExternalIdAndSessionNameTreatedAsUnset() {
+ String roleArn = ROLE_ARN_PREFIX + "blank";
+ try (KinesisClient blank = KinesisOffsetGen.createKinesisClient(
+ REGION, null, null, null, roleArn, "", "");
+ KinesisClient unset = KinesisOffsetGen.createKinesisClient(
+ REGION, null, null, null, roleArn, null,
KinesisSourceConfig.KINESIS_ROLE_SESSION_NAME.defaultValue())) {
+ // Blank external id / session name normalize to unset / default, so
both hit the same cache entry.
+ assertSame(providerOf(unset), providerOf(blank));
+ }
+ }
+}
