[hozkaya2000]

Description

Role-based AWS_MSK_IAM SASL (data-plane / broker) authentication assumed the customer role inside the vendored aws-msk-iam-auth library, which has no hook for sts_header_overrides. As a result the x-amz-source-arn / x-amz-source-account headers were dropped on the data-plane sts:AssumeRole, leaving a confused-deputy gap on the one MSK code path (the bootstrap-broker lookup already carried the headers correctly).

This PR routes the ROLE path through a dedicated callback handler that sources credentials exclusively from Data Prepper's own header-aware STS provider — the same StsAssumeRoleCredentialsProvider already built (with the header overrides applied) for the broker lookup — so both STS call sites on the MSK path now carry the headers.

This supersedes #6925 and reworks it per review feedback.

Key changes

• MskIamAuthCredentialsCallbackHandler now implements AuthenticateCallbackHandler directly instead of extending IAMClientCallbackHandler. Its configure(...) only validates the mechanism — it never reads JAAS options and never constructs an MSKCredentialProvider or falls back to the SDK default chain. Credentials come only from KafkaSecurityConfigurer.getMskCredentialsProvider().
• KafkaSecurityConfigurer.setAwsIamAuthProperties — the ROLE path is now unconditional (single auth path): it always installs our handler + the bare IAMLoginModule required; JAAS config, so the library never assembles its own header-less provider. No longer gated on whether sts_header_overrides are present.
• The library's internals SASL machinery is untouched; the SDK-v2 provider's credentials are translated to SDK-v1 for the AWSCredentialsCallback bridge.
• DEFAULT MSK-IAM mode is unchanged (no role assumption, no headers to carry).

Why implement the interface directly

Review feedback (@dlvenable) on #6925: avoid the split authentication approach and never let credentials load through the library's own MSKCredentialProvider or the SDK default chain. Implementing AuthenticateCallbackHandler directly makes the header-less assembler unreachable and gives a single, always-on auth path.

Testing

• MskIamAuthCredentialsCallbackHandlerTest — 7/7 passing (credential resolution, v2→v1 translation, configure accept/reject, unsupported-callback).
• KafkaSecurityConfigurerTest — 35/35 passing (including the always-on ROLE assertions).
• Built and tested against Java 11.
• Verified end-to-end that sts_header_overrides reach the data-plane sts:AssumeRole on the aws_msk_iam: role branch: driving the real KafkaSecurityConfigurer.setAuthProperties(...) path from a ROLE config, the SDK's own request logger showed the outbound AssumeRole to STS carrying x-amz-source-account / x-amz-source-arn in its headers and returning HTTP 200.

Resolves #6923

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit f33698e)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ Recommended focus areas for review Possible NPE on missing provider handleCallback calls provider.resolveCredentials() without a null check. If KafkaSecurityConfigurer.getMskCredentialsProvider() returns null (e.g., setAwsIamAuthProperties is invoked without the broker-lookup path that populates mskCredentialsProvider), authentication will fail with an opaque NPE wrapped in setLoadingException. Consider validating the provider is non-null and surfacing a clearer error message, since the ROLE path now unconditionally depends on this static field being set. private void handleCallback(final AWSCredentialsCallback callback) { final AwsCredentialsProvider provider = KafkaSecurityConfigurer.getMskCredentialsProvider(); try { LOG.debug("Resolving MSK IAM credentials from header-aware provider {}", provider == null ? "null" : provider.getClass().getSimpleName()); callback.setAwsCredentials(toV1Credentials(provider.resolveCredentials())); } catch (final Exception e) { LOG.debug("Failed to resolve MSK IAM credentials with STS header overrides", e); callback.setLoadingException(e); } } Behavior change for ROLE path The ROLE path previously passed awsRoleArn, awsStsRegion, and optionally awsRoleSessionName into the JAAS config so the library assumed the configured role. Now the JAAS config is bare and credentials must come from mskCredentialsProvider. If this method is ever invoked before the MSK bootstrap-broker lookup populates mskCredentialsProvider (e.g., when a user supplies explicit bootstrap servers in ROLE mode without triggering MSK discovery), authentication will fail. Confirm every ROLE-mode code path initializes mskCredentialsProvider before Kafka client startup. public static void setAwsIamAuthProperties(Properties properties, final AwsIamAuthConfig awsIamAuthConfig, final AwsConfig awsConfig) { properties.put(SECURITY_PROTOCOL, SASL_SSL_PROTOCOL); properties.put(SASL_MECHANISM, "AWS_MSK_IAM"); properties.put(SASL_CLIENT_CALLBACK_HANDLER_CLASS, "software.amazon.msk.auth.iam.IAMClientCallbackHandler"); if (awsIamAuthConfig == AwsIamAuthConfig.ROLE) { if (Objects.isNull(awsConfig)) { throw new RuntimeException("AWS Config needs to be specified when sasl/aws_msk_iam is set to \"role\""); } properties.put(SASL_CLIENT_CALLBACK_HANDLER_CLASS, MskIamAuthCredentialsCallbackHandler.class.getName()); properties.put(SASL_JAAS_CONFIG, "software.amazon.msk.auth.iam.IAMLoginModule required;"); } else if (awsIamAuthConfig == AwsIamAuthConfig.DEFAULT) { properties.put(SASL_JAAS_CONFIG, "software.amazon.msk.auth.iam.IAMLoginModule required;"); }

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to f33698e

Explore these optional code suggestions:

Category	Suggestion	Impact
General	Provide clearer error when provider is null When the provider is null, this throws NPE inside the try/catch and surfaces as a setLoadingException(NullPointerException), which is a confusing diagnostic. Explicitly handle the null case to produce an informative exception message that aids debugging and operations. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/MskIamAuthCredentialsCallbackHandler.java [62-70] final AwsCredentialsProvider provider = KafkaSecurityConfigurer.getMskCredentialsProvider(); try { + if (provider == null) { + throw new IllegalStateException("MSK IAM credentials provider is not initialized"); + } LOG.debug("Resolving MSK IAM credentials from header-aware provider {}", - provider == null ? "null" : provider.getClass().getSimpleName()); + provider.getClass().getSimpleName()); callback.setAwsCredentials(toV1Credentials(provider.resolveCredentials())); } catch (final Exception e) { LOG.debug("Failed to resolve MSK IAM credentials with STS header overrides", e); callback.setLoadingException(e); } Suggestion importance[1-10]: 5 __ Why: Explicitly handling the null provider case yields a clearer diagnostic message than a raw NPE, which is a reasonable minor improvement for debuggability.	Low
	Fail fast if credentials provider missing The new ROLE branch sets the custom callback handler but relies on mskCredentialsProvider being initialized earlier. If setAwsIamAuthProperties is ever invoked without the MSK credentials provider being set first, the callback handler will fail with NPE at authentication time. Add a defensive check or ensure initialization here to fail fast with a clearer error. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaSecurityConfigurer.java [252-253] properties.put(SASL_CLIENT_CALLBACK_HANDLER_CLASS, "software.amazon.msk.auth.iam.IAMClientCallbackHandler"); if (awsIamAuthConfig == AwsIamAuthConfig.ROLE) { if (Objects.isNull(awsConfig)) { throw new RuntimeException("AWS Config needs to be specified when sasl/aws_msk_iam is set to \"role\""); } + if (Objects.isNull(mskCredentialsProvider)) { + throw new RuntimeException("MSK credentials provider must be initialized before configuring AWS_MSK_IAM role auth"); + } properties.put(SASL_CLIENT_CALLBACK_HANDLER_CLASS, MskIamAuthCredentialsCallbackHandler.class.getName()); properties.put(SASL_JAAS_CONFIG, "software.amazon.msk.auth.iam.IAMLoginModule required;"); Suggestion importance[1-10]: 4 __ Why: Adding a defensive null check for mskCredentialsProvider improves error reporting, but this is a minor robustness improvement; the provider is expected to be set in normal flow.	Low

---

Previous suggestions

Suggestions up to commit 1e06231

Category	Suggestion	Impact
Possible issue	Initialize credentials provider before callback use The custom callback handler relies on KafkaSecurityConfigurer.getMskCredentialsProvider(), but mskCredentialsProvider is only initialized in the MSK bootstrap discovery path (getBootStrapServersForMsk). When users configure aws_msk_iam: role without MSK broker discovery, the provider will be null and authentication will fail with NPE. Ensure the provider is initialized here before registering the callback handler. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaSecurityConfigurer.java [252-253] properties.put(SECURITY_PROTOCOL, SASL_SSL_PROTOCOL); properties.put(SASL_MECHANISM, "AWS_MSK_IAM"); properties.put(SASL_CLIENT_CALLBACK_HANDLER_CLASS, "software.amazon.msk.auth.iam.IAMClientCallbackHandler"); if (awsIamAuthConfig == AwsIamAuthConfig.ROLE) { if (Objects.isNull(awsConfig)) { throw new RuntimeException("AWS Config needs to be specified when sasl/aws_msk_iam is set to \"role\""); } + if (mskCredentialsProvider == null) { + mskCredentialsProvider = createCredentialsProvider(awsConfig); + } properties.put(SASL_CLIENT_CALLBACK_HANDLER_CLASS, MskIamAuthCredentialsCallbackHandler.class.getName()); properties.put(SASL_JAAS_CONFIG, "software.amazon.msk.auth.iam.IAMLoginModule required;"); Suggestion importance[1-10]: 7 __ Why: This points out a potentially real issue: if aws_msk_iam: role is used without MSK broker discovery, mskCredentialsProvider may be null, causing authentication to fail. However, the suggestion assumes a createCredentialsProvider method exists without verifying the outer codebase.	Medium
General	Fail fast with clear error on null provider When provider is null, this currently relies on the thrown NPE being caught. Log a clearer error and raise an explicit exception so operators can diagnose misconfiguration quickly instead of seeing an opaque NPE in the loading exception. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/MskIamAuthCredentialsCallbackHandler.java [62-70] final AwsCredentialsProvider provider = KafkaSecurityConfigurer.getMskCredentialsProvider(); try { + if (provider == null) { + throw new IllegalStateException("MSK IAM credentials provider has not been initialized"); + } LOG.debug("Resolving MSK IAM credentials from header-aware provider {}", - provider == null ? "null" : provider.getClass().getSimpleName()); + provider.getClass().getSimpleName()); callback.setAwsCredentials(toV1Credentials(provider.resolveCredentials())); } catch (final Exception e) { LOG.debug("Failed to resolve MSK IAM credentials with STS header overrides", e); callback.setLoadingException(e); } Suggestion importance[1-10]: 4 __ Why: Improves diagnostics by throwing a clear IllegalStateException instead of relying on NPE, but this is a minor code quality improvement and the exception is still caught and set as loading exception.	Low

Suggestions up to commit 9f917fb

Category	Suggestion	Impact
Possible issue	Guard against uninitialized credentials provider The code relies on mskCredentialsProvider being initialized before the callback handler resolves credentials, but there is no guard here ensuring it was set for the ROLE path. If setAwsIamAuthProperties is invoked without a prior call that initializes mskCredentialsProvider, the callback handler will throw an NPE at authentication time. Validate/initialize the provider here (using awsConfig) so STS header overrides are guaranteed to apply. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaSecurityConfigurer.java [252-253] -properties.put(SASL_CLIENT_CALLBACK_HANDLER_CLASS, "software.amazon.msk.auth.iam.IAMClientCallbackHandler"); if (awsIamAuthConfig == AwsIamAuthConfig.ROLE) { if (Objects.isNull(awsConfig)) { throw new RuntimeException("AWS Config needs to be specified when sasl/aws_msk_iam is set to \"role\""); } + if (Objects.isNull(mskCredentialsProvider)) { + throw new RuntimeException("MSK credentials provider has not been initialized; cannot apply STS header overrides."); + } properties.put(SASL_CLIENT_CALLBACK_HANDLER_CLASS, MskIamAuthCredentialsCallbackHandler.class.getName()); properties.put(SASL_JAAS_CONFIG, "software.amazon.msk.auth.iam.IAMLoginModule required;"); Suggestion importance[1-10]: 5 __ Why: Adding a guard for an uninitialized mskCredentialsProvider provides earlier and clearer failure feedback than an eventual NPE at authentication time, improving diagnostics. However, the initialization flow isn't fully visible in this diff, so impact is moderate.	Low
General	Fail with a clear error for null provider When provider is null, the current code logs "null" and then dereferences it, producing an opaque NullPointerException passed to setLoadingException. Throw a descriptive exception explicitly so operators can diagnose the misconfiguration (static provider not initialized) rather than seeing a bare NPE. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/MskIamAuthCredentialsCallbackHandler.java [61-71] private void handleCallback(final AWSCredentialsCallback callback) { final AwsCredentialsProvider provider = KafkaSecurityConfigurer.getMskCredentialsProvider(); try { - LOG.debug("Resolving MSK IAM credentials from header-aware provider {}", - provider == null ? "null" : provider.getClass().getSimpleName()); + if (provider == null) { + throw new IllegalStateException("MSK IAM credentials provider is not initialized; STS header overrides cannot be applied."); + } + LOG.debug("Resolving MSK IAM credentials from header-aware provider {}", provider.getClass().getSimpleName()); callback.setAwsCredentials(toV1Credentials(provider.resolveCredentials())); } catch (final Exception e) { LOG.debug("Failed to resolve MSK IAM credentials with STS header overrides", e); callback.setLoadingException(e); } } Suggestion importance[1-10]: 4 __ Why: Throwing a descriptive IllegalStateException instead of relying on an implicit NPE improves error diagnosability, but it's a minor enhancement since the current code already handles the exception via setLoadingException.	Low

Suggestions up to commit a5c5abe

Category	Suggestion	Impact
Possible issue	Verify role config reaches credentials provider The JAAS config no longer carries awsRoleArn, awsStsRegion, or awsRoleSessionName, so the configured STS role, region, and session name from AwsConfig are silently ignored for ROLE mode. Ensure MskIamAuthCredentialsCallbackHandler / mskCredentialsProvider is actually initialized from awsConfig with these fields (including stsRoleArn, region, and session name), otherwise the role assumption will never happen and MSK auth will use the default credentials chain instead. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaSecurityConfigurer.java [249-255] properties.put(SECURITY_PROTOCOL, SASL_SSL_PROTOCOL); properties.put(SASL_MECHANISM, "AWS_MSK_IAM"); properties.put(SASL_CLIENT_CALLBACK_HANDLER_CLASS, "software.amazon.msk.auth.iam.IAMClientCallbackHandler"); if (awsIamAuthConfig == AwsIamAuthConfig.ROLE) { if (Objects.isNull(awsConfig)) { throw new RuntimeException("AWS Config needs to be specified when sasl/aws_msk_iam is set to \"role\""); } + // Ensure mskCredentialsProvider is built from awsConfig (stsRoleArn, region, stsRoleSessionName) + // so the callback handler assumes the correct role; otherwise role-based auth will not work. properties.put(SASL_CLIENT_CALLBACK_HANDLER_CLASS, MskIamAuthCredentialsCallbackHandler.class.getName()); properties.put(SASL_JAAS_CONFIG, "software.amazon.msk.auth.iam.IAMLoginModule required;"); } else if (awsIamAuthConfig == AwsIamAuthConfig.DEFAULT) { Suggestion importance[1-10]: 6 __ Why: The suggestion raises a valid concern that the JAAS config no longer carries role-specific parameters, so mskCredentialsProvider must be properly initialized with awsConfig values. However, it only adds a comment rather than a code fix, and the actual initialization likely happens elsewhere in the codebase.	Low
General	Handle null credentials provider explicitly When provider is null, the handler ends up throwing a bare NullPointerException which is confusing for operators and loses the root cause. Explicitly check for null and set a descriptive exception (e.g. IllegalStateException) so logs clearly indicate that the MSK credentials provider was never initialized rather than reporting a generic NPE. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/authenticator/MskIamAuthCredentialsCallbackHandler.java [62-72] private void handleCallback(final AWSCredentialsCallback callback) { final AwsCredentialsProvider provider = KafkaSecurityConfigurer.getMskCredentialsProvider(); try { + if (provider == null) { + throw new IllegalStateException("MSK IAM credentials provider has not been initialized"); + } LOG.debug("Resolving MSK IAM credentials from header-aware provider {}", - provider == null ? "null" : provider.getClass().getSimpleName()); + provider.getClass().getSimpleName()); callback.setAwsCredentials(toV1Credentials(provider.resolveCredentials())); } catch (final Exception e) { LOG.debug("Failed to resolve MSK IAM credentials with STS header overrides", e); callback.setLoadingException(e); } } Suggestion importance[1-10]: 4 __ Why: The suggestion improves error messaging when the provider is null, which aids operator debugging. It's a minor readability/maintainability improvement rather than a critical fix, as the NPE is still caught and reported via setLoadingException.	Low

Suggestions up to commit 89e429d

Category	Suggestion	Impact
Possible issue	Validate credentials provider before IAM role auth The callback handler ignores awsConfig.getStsRoleArn(), getRegion(), and getStsRoleSessionName(), so switching from the JAAS-based config silently drops the configured role ARN/region/session name used by the login module. Verify that the header-aware mskCredentialsProvider is initialized with these values (role ARN, region, session name) before relying on it for ROLE auth; otherwise the broker-side sts:AssumeRole identity will change. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaSecurityConfigurer.java [249-254] properties.put(SECURITY_PROTOCOL, SASL_SSL_PROTOCOL); properties.put(SASL_MECHANISM, "AWS_MSK_IAM"); properties.put(SASL_CLIENT_CALLBACK_HANDLER_CLASS, "software.amazon.msk.auth.iam.IAMClientCallbackHandler"); if (awsIamAuthConfig == AwsIamAuthConfig.ROLE) { if (Objects.isNull(awsConfig)) { throw new RuntimeException("AWS Config needs to be specified when sasl/aws_msk_iam is set to \"role\""); } + if (Objects.isNull(mskCredentialsProvider)) { + throw new RuntimeException("MSK credentials provider must be initialized with STS role settings before setting IAM auth properties"); + } properties.put(SASL_CLIENT_CALLBACK_HANDLER_CLASS, MskIamAuthCredentialsCallbackHandler.class.getName()); properties.put(SASL_JAAS_CONFIG, "software.amazon.msk.auth.iam.IAMLoginModule required;"); } else if (awsIamAuthConfig == AwsIamAuthConfig.DEFAULT) { Suggestion importance[1-10]: 5 __ Why: The suggestion raises a valid concern about ensuring mskCredentialsProvider is properly initialized with role ARN, region, and session name before being used, which could prevent silent misconfiguration. However, it's partly a verification suggestion and the fix only adds a null check without addressing the deeper concern about whether role settings propagate correctly.	Low
General	Fail explicitly when provider is null When provider is null, the current code relies on a NullPointerException to signal failure, which is confusing in logs and tightly couples the test to NPE. Throw an explicit exception describing the misconfiguration so operators understand the root cause. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/authenticator/MskIamAuthCredentialsCallbackHandler.java [62-72] private void handleCallback(final AWSCredentialsCallback callback) { final AwsCredentialsProvider provider = KafkaSecurityConfigurer.getMskCredentialsProvider(); try { + if (provider == null) { + throw new IllegalStateException("MSK IAM credentials provider is not initialized"); + } LOG.debug("Resolving MSK IAM credentials from header-aware provider {}", - provider == null ? "null" : provider.getClass().getSimpleName()); + provider.getClass().getSimpleName()); callback.setAwsCredentials(toV1Credentials(provider.resolveCredentials())); } catch (final Exception e) { LOG.debug("Failed to resolve MSK IAM credentials with STS header overrides", e); callback.setLoadingException(e); } } Suggestion importance[1-10]: 4 __ Why: Throwing an explicit IllegalStateException with a descriptive message is a minor readability/debuggability improvement over relying on an implicit NPE, but the functional behavior is largely the same since the exception is still caught and set as the loading exception.	Low

Suggestions up to commit 5ae10b5

Category	Suggestion	Impact
General	Fail fast when provider is uninitialized Relying on a static getter means the handler silently gets a null provider if initialization order changes, and the resulting NullPointerException surfaces only as a loading failure during authentication. Explicitly check for null and raise a clearer exception so misconfiguration is diagnosable from logs. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/authenticator/MskIamAuthCredentialsCallbackHandler.java [63-71] final AwsCredentialsProvider provider = KafkaSecurityConfigurer.getMskCredentialsProvider(); try { + if (provider == null) { + throw new IllegalStateException("MSK IAM credentials provider is not initialized; " + + "ensure Kafka security configuration ran before SASL authentication."); + } LOG.debug("Resolving MSK IAM credentials from header-aware provider {}", - provider == null ? "null" : provider.getClass().getSimpleName()); + provider.getClass().getSimpleName()); callback.setAwsCredentials(toV1Credentials(provider.resolveCredentials())); } catch (final Exception e) { LOG.debug("Failed to resolve MSK IAM credentials with STS header overrides", e); callback.setLoadingException(e); } Suggestion importance[1-10]: 4 __ Why: Adds clearer diagnostics for a misconfiguration case, but the current code already propagates the NPE via setLoadingException, so impact is minor.	Low
	Avoid overwriting the callback handler property The initial SASL_CLIENT_CALLBACK_HANDLER_CLASS assignment to IAMClientCallbackHandler is immediately overwritten only for the ROLE branch; the DEFAULT branch still uses it. This is functional but confusing — consider setting the default handler only in the DEFAULT branch to make intent explicit and avoid redundant puts. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaSecurityConfigurer.java [248-254] properties.put(SECURITY_PROTOCOL, SASL_SSL_PROTOCOL); properties.put(SASL_MECHANISM, "AWS_MSK_IAM"); -properties.put(SASL_CLIENT_CALLBACK_HANDLER_CLASS, "software.amazon.msk.auth.iam.IAMClientCallbackHandler"); if (awsIamAuthConfig == AwsIamAuthConfig.ROLE) { if (Objects.isNull(awsConfig)) { throw new RuntimeException("AWS Config needs to be specified when sasl/aws_msk_iam is set to \"role\""); } properties.put(SASL_CLIENT_CALLBACK_HANDLER_CLASS, MskIamAuthCredentialsCallbackHandler.class.getName()); properties.put(SASL_JAAS_CONFIG, "software.amazon.msk.auth.iam.IAMLoginModule required;"); Suggestion importance[1-10]: 3 __ Why: Minor code clarity improvement; the current code works correctly but has a redundant property assignment that is immediately overwritten in the ROLE branch.	Low

[github-actions]

Persistent review updated to latest commit 5ae10b5

[github-actions]

Persistent review updated to latest commit 89e429d

[github-actions]

Persistent review updated to latest commit a5c5abe

[dlvenable]

Can you move this to the util package and make it package private? If Kafka accept this it would be better to keep it isolated.

[hozkaya2000]

Done — moved the handler into the util package and made the class package-private. Unit tests still green (7/7).

[dlvenable]

This also could be package private if you move the other class.

[hozkaya2000]

Done — getMskCredentialsProvider() is now package-private since both classes live in util. (35/35 passing.)

[dlvenable]

I was curious how this even worked since I don't see this here.

We already have code to use the STS headers. They are just not used on communicating with the Kafka client directly.

data-prepper/data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaSecurityConfigurer.java

Lines 324 to 334 in e48a6ff

	Map<String, String> headers = awsConfig.getAwsStsHeaderOverrides();
	if (Objects.nonNull(headers)) {
	assumeRequestBuilder.overrideConfiguration(configuration -> {
	headers.forEach(configuration::putHeader);
	});
	}
	mskCredentialsProvider = StsAssumeRoleCredentialsProvider
	.builder()
	.stsClient(stsClient)
	.refreshRequest(assumeRequestBuilder.build())
	.build();

[hozkaya2000]

Yep, that STS-header code is only used for the initial GetBootstrapBrokers call. The Kafka client's own SASL handshake goes through the vendored IAMClientCallbackHandler, which builds its own provider from the JAAS role with no header overrides — so the data-plane AssumeRole was missing the headers. This callback handler is what routes that handshake through the header-aware provider too.

[github-actions]

Persistent review updated to latest commit 9f917fb

[github-actions]

✅ License Header Check Passed

All newly added files have proper license headers. Great work! 🎉

[github-actions]

Persistent review updated to latest commit 1e06231

[dlvenable]

Thank you @hozkaya2000 for this contribution!

[dinujoh]

Kafka plugin integ test are failing. Can you take a look ?

[github-actions]

Persistent review updated to latest commit f33698e