yihua commented on code in PR #19383:
URL: https://github.com/apache/hudi/pull/19383#discussion_r3667457269
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/config/KinesisSourceConfig.java:
##########
@@ -76,6 +76,17 @@ 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.2.0")
Review Comment:
```suggestion
.sinceVersion("1.3.0")
```
##########
hudi-utilities/pom.xml:
##########
@@ -537,6 +537,12 @@
<artifactId>kinesis</artifactId>
<version>${aws.sdk.version}</version>
</dependency>
+ <!-- STS: assume-role credentials for reading a Kinesis stream in a
different AWS account. -->
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>sts</artifactId>
+ <version>${aws.sdk.version}</version>
+ </dependency>
Review Comment:
Should this be shaded in `hudi-utilities-bundle`?
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KinesisOffsetGen.java:
##########
@@ -304,28 +308,58 @@ public KinesisOffsetGen(TypedProperties props) {
getStringWithAltKeys(props,
KinesisSourceConfig.KINESIS_STARTING_POSITION, true));
}
+ /**
+ * Assume-role credentials providers cached by {@code region|roleArn}.
Kinesis clients are built
+ * per {@code mapPartitions} task, so creating a fresh STS client each time
would leak one (with its
+ * own HTTP connection pool) per micro-batch on a long-lived streaming
executor — AWS SDK v2 does not
+ * close a user-supplied credentials provider (nor its injected StsClient)
when the KinesisClient
+ * closes. Caching means at most one provider/StsClient per distinct role
per executor JVM, reused
+ * across partitions and micro-batches. Providers auto-refresh and live for
the JVM lifetime, so they
+ * are intentionally never closed.
+ */
+ private static final Map<String, StsAssumeRoleCredentialsProvider>
ASSUME_ROLE_PROVIDERS =
+ new ConcurrentHashMap<>();
+
+ private static StsAssumeRoleCredentialsProvider assumeRoleProvider(String
region, String roleArn) {
+ return ASSUME_ROLE_PROVIDERS.computeIfAbsent(region + "|" + roleArn,
ignored ->
+ StsAssumeRoleCredentialsProvider.builder()
+ .stsClient(StsClient.builder().region(Region.of(region)).build())
+ .refreshRequest(AssumeRoleRequest.builder()
+ .roleArn(roleArn)
+ .roleSessionName("hudi-kinesis-source")
+ .build())
+ .build());
+ }
+
/**
Review Comment:
This is a private method, so it's a minor nit only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
