NiFi 2.9 Kafka3ConnectionService with Google Cloud Managed Kafka using SASL_SSL/OAUTHBEARER

Hi,

I am using Apache NiFi 2.9 running in Kubernetes and need to connect a ConsumeKafka processor to Google Cloud Managed Service for Apache Kafka using Kafka3ConnectionService.

Due to a security constraint, SASL/PLAIN cannot be used. I need to use:

Security Protocol: SASL_SSL

SASL Mechanism: OAUTHBEARER

For Google Cloud Managed Kafka authentication, I am trying to use Google's Kafka authentication callback handler with Application Default Credentials (ADC).

The authentication setup requires the Google/Kafka authentication dependencies, including:

google-auth-library-oauth2-http-1.47.0.jar

kafka-oauth-client-0.17.1.jar

managed-kafka-auth-login-handler-1.0.5.jar

I have made these dependencies available to the NiFi environment.

I have also configured NiFi to use a JAAS configuration file through bootstrap.conf:

.properties

java.arg.18=-Djava.security.auth.login.config=/opt/nifi/nifi-current/conf/kafka_client_jaas.conf

The JAAS configuration uses:

org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;

For the Google Managed Kafka authentication mechanism, I also need the Kafka client to use:

.properties

sasl.login.callback.handler.class=com.google.cloud.hosted.kafka.auth.GcpLoginCallbackHandler

Issue

When I configure Kafka3ConnectionService with:

Security Protocol = SASL_SSL

SASL Mechanism = OAUTHBEARER

NiFi requires me to configure an OAuth2 Access Token Provider Service.

My understanding is that the Google authentication library can handle token acquisition through GcpLoginCallbackHandler using Google credentials/ADC.

Therefore, I am trying to understand how this authentication model should be integrated with Kafka3ConnectionService.

I also don't see a way to add the following Kafka client property as a dynamic property in Kafka3ConnectionService:

.properties

sasl.login.callback.handler.class=com.google.cloud.hosted.kafka.auth.GcpLoginCallbackHandler

Questions

1. Does Kafka3ConnectionService in NiFi 2.9 support using a custom Kafka sasl.login.callback.handler.class with the OAUTHBEARER mechanism?

2. Is there a supported way to use Google's:

com.google.cloud.hosted.kafka.auth.GcpLoginCallbackHandler

with Kafka3ConnectionService without configuring NiFi's OAuth2 Access Token Provider Service?

3. If Kafka3ConnectionService requires an OAuth2AccessTokenProvider for OAUTHBEARER, would implementing a custom NiFi OAuth2AccessTokenProvider Controller Service as a separate NAR be the recommended extension point for Google Cloud Managed Kafka?

4. Is there another recommended approach for passing the required Google Kafka authentication configuration to the underlying Kafka client?

Upgradeability requirement

I would prefer a solution that does not modify or rebuild NiFi's standard Kafka NARs.

Since NiFi is deployed as a Kubernetes pod and will be upgraded to newer NiFi versions over time, ideally the Google authentication integration would be packaged/configured separately so that NiFi upgrades do not require rebuilding the standard NiFi Kafka components.

Has anyone successfully configured NiFi 2.9 + Kafka3ConnectionService + Google Cloud Managed Kafka using SASL_SSL/OAUTHBEARER?

Any guidance on the supported extension/configuration approach would be greatly appreciated.

Thanks.