RANGER-5827: Add shared Kafka client authentication module with support for both SASL/OAUTHBEARER and SASL/GSSAPI

[kumaab]

What changes were proposed in this pull request?

Ranger's Kafka clients (audit-ingestor, audit-dispatchers, tagsync) each assemble their own SASL settings and only support Kerberos keytabs. Kafka's bundled OAUTHBEARER handler reads a file:// token once and caches it, so rotated platform tokens, such as Kubernetes projected service-account tokens, leave the client presenting an expired token.

• New module ranger-kafka-client-auth:
  • RangerKafkaClientSecurityConfig builds security.protocol / sasl.* client properties from .kafka.* for producers, consumers and admin clients. GSSAPI remains the default mechanism; OAUTHBEARER is selected with kafka.sasl.mechanism. JAAS is built from kafka.sasl.jaas.loginModuleName, kafka.sasl.jaas.loginModuleControlFlag and any kafka.sasl.jaas.option.<name>, so Krb5LoginModule options such as useTicketCache are configuration, not code. <prefix>.kafka.sasl.* and <prefix>.kafka.ssl.* pass through and override.
  • RangerKafkaOAuthBearerLoginCallbackHandler presents a JWT read from a file:// URL. It re-reads the file on every login and refresh, so the client always presents the current token after rotation. http(s):// endpoints use Kafka’s own client-credentials handler.
  • RangerOAuthBearerToken, a reusable OAuthBearerToken built from parsed claims.
• audit-ingestor, all audit-dispatcher backends and the audit topic bootstrap call the helper instead of building a Kerberos-only JAAS string each; audit-ingestor also loads the site XML given by -Daudit.config after the packaged default.
• tagsync: TAG_SOURCE_ATLAS_KAFKA_SASL_MECHANISM and TAG_SOURCE_ATLAS_KAFKA_OAUTH_TOKEN_URL rendered into atlas-application.properties; module added to the tagsync distribution.
• ranger-docker: broker accepts GSSAPI and OAUTHBEARER; ranger-kafka mints and rotates dev tokens into a shared volume;

Upgrade note for GSSAPI deployments: the Krb5LoginModule options useKeyTab, storeKey and useTicketCache are no longer hardcoded. The shipped site XML templates set them as <prefix>.kafka.sasl.jaas.option.*; an existing ranger-audit-ingestor-site.xml or ranger-audit-dispatcher-*-site.xml needs those three properties added.

How was this patch tested?

• Added unit tests
• Live in ranger-docker (KERBEROS_ENABLED=true, OpenSearch profile), fresh build and cold start, with the broker accepting GSSAPI and OAUTHBEARER:
  • OAUTHBEARER (shipped default): audit-ingestor, audit-dispatcher-opensearch and tagsync logged in through RangerKafkaOAuthBearerLoginCallbackHandler; denied Kafka accesses by a test user were audited end to end into OpenSearch; all three re-logged in with the rotated token at Kafka’s scheduled refresh.
  • GSSAPI (mechanism set to GSSAPI in the ingestor YAML, dispatcher XML and tagsync install properties): all three authenticated with their keytabs and audits flowed end to end.

Ingestor Logs

# run 1: OAUTHBEARER (shipped default)
00:10:24 RangerKafkaClientSecurityConfig  sasl.mechanism=OAUTHBEARER sasl.login.callback.handler.class=...RangerKafkaOAuthBearerLoginCallbackHandler
00:10:24 RangerKafkaOAuthBearerLoginCallbackHandler  token loaded from /etc/kafka-oauth/rangerauditserver.token, expiresAtMs=1791850775000
00:10:24 ExpiringCredentialRefreshingLogin  Successfully logged in.
00:14:34 ExpiringCredentialRefreshingLogin  Initiating re-login for rangerauditserver
00:14:35 RangerKafkaOAuthBearerLoginCallbackHandler  token loaded from /etc/kafka-oauth/rangerauditserver.token, expiresAtMs=1791851015000   <- rotated token picked up

# run 2: GSSAPI (kafka.sasl.mechanism: GSSAPI in the ingestor YAML)
00:15:18 RangerKafkaClientSecurityConfig  sasl.mechanism=GSSAPI sasl.kerberos.service.name=kafka
00:15:18 AbstractLogin  Successfully logged in.
00:15:18 KerberosLogin  [Principal=rangerauditserver/ranger-audit-ingestor.rangernw@EXAMPLE.COM]: TGT refresh thread started.

# run 3: back to OAUTHBEARER
00:16:11 RangerKafkaClientSecurityConfig  sasl.mechanism=OAUTHBEARER sasl.login.callback.handler.class=...RangerKafkaOAuthBearerLoginCallbackHandler
00:16:11 RangerKafkaOAuthBearerLoginCallbackHandler  token loaded from /etc/kafka-oauth/rangerauditserver.token, expiresAtMs=1791851135000
00:16:11 ExpiringCredentialRefreshingLogin  Successfully logged in.
00:20:35 ExpiringCredentialRefreshingLogin  Initiating re-login for rangerauditserver
00:20:35 RangerKafkaOAuthBearerLoginCallbackHandler  token loaded from /etc/kafka-oauth/rangerauditserver.token, expiresAtMs=1791851375000
00:24:35 ExpiringCredentialRefreshingLogin  Initiating re-login for rangerauditserver
00:24:35 RangerKafkaOAuthBearerLoginCallbackHandler  token loaded from /etc/kafka-oauth/rangerauditserver.token, expiresAtMs=1791851615000

[ramackri]

LGTM