Kafka Producer OAuth2 Authentication Fails After Internet Outage In Air‑gapped Cluster

Q: Solution

Implement an offline token cache with proactive refresh and configure the Kafka client to use a custom OAuthBearerTokenProvider that falls back to the cached token when the token endpoint is unreachable.

In this article

Table of Contents

Kafka producer OAuth2 authentication fails after internet outage in an air‑gapped cluster

Problem

In a production Kafka cluster that is completely air‑gapped, producers and consumers obtain OAuth2 access tokens from an external authorization server (e.g., Keycloak, Auth0). When the corporate internet link goes down, the token acquisition and refresh flow cannot reach the token endpoint, resulting in:

Repeated org.apache.kafka.common.errors.SaslAuthenticationException errors in producer logs.
Message production stalls – the producer blocks on send() until the request times out.
Consumer group rebalances triggered by authentication failures.

Typical log excerpt during the outage:


2024-10-18 09:12:34,567 ERROR [producer-1] org.apache.kafka.clients.producer.internals.Sender - Authentication failed for client-id=producer-1: org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed: token acquisition failed: java.net.UnknownHostException: auth.example.com
    at org.apache.kafka.common.security.oauthbearer.internals.secured.OAuthBearerLoginModule.login(OAuthBearerLoginModule.java:210)
    at javax.security.auth.login.LoginContext.login(LoginContext.java:124)
    ... 12 more
Caused by: java.net.UnknownHostException: auth.example.com
    at java.base/sun.net.spi.DefaultProxySelector.select(DefaultProxySelector.java:174)
    ... 5 more

Root Cause

The failure stems from a hard dependency on the external OAuth2 token endpoint for both initial token fetch and periodic refresh. In an air‑gapped environment:

Network isolation prevents any outbound TCP connection to the token server.
Token lifetimes (typically 5–15 minutes) are shorter than the duration of an internet outage, causing the client to attempt a refresh that inevitably fails.
Kafka client configuration does not include a fallback or cached token mechanism; the OAuthBearerLoginModule always invokes TokenRetriever which performs a live HTTP request.

Consequently, once the cached token expires, the producer cannot re‑authenticate and all subsequent produce requests are rejected.

Debug

Step‑by‑step investigation that reproduced the issue:

Confirm that the Kafka client can reach the broker when the internet is up.
Simulate an outage by blocking DNS resolution for auth.example.com on the producer host:

sudo iptables -A OUTPUT -p tcp -d auth.example.com -j REJECT

Observe the log pattern shown above – the UnknownHostException appears exactly when the token refresh timer fires.

Additional diagnostics:

journalctl -u kafka-producer.service -f – confirms no other errors besides authentication.
Enable debug logging for the OAuth module:

# application.properties
log4j.logger.org.apache.kafka.common.security.oauthbearer=DEBUG

Resulting debug snippet:


DEBUG OAuthBearerLoginModule - Starting token acquisition from https://auth.example.com/realms/prod/protocol/openid-connect/token
DEBUG OAuthBearerLoginModule - HTTP request failed: java.net.UnknownHostException: auth.example.com

Solution

Implement an offline token cache with proactive refresh and configure the Kafka client to use a custom OAuthBearerTokenProvider that falls back to the cached token when the token endpoint is unreachable.

1. Create a local token cache service

Deploy a lightweight HTTP endpoint inside the air‑gapped network that stores a long‑lived refresh token and can issue short‑lived access tokens on demand. The service uses the original external OAuth2 server when connectivity exists, otherwise serves the last known good token.

Example token-cache.yaml (Kubernetes deployment):


apiVersion: apps/v1
kind: Deployment
metadata:
  name: oauth-token-cache
spec:
  replicas: 1
  selector:
    matchLabels:
      app: oauth-token-cache
  template:
    metadata:
      labels:
        app: oauth-token-cache
    spec:
      containers:
      - name: token-cache
        image: ghcr.io/example/oauth-token-cache:1.2.0
        env:
        - name: EXTERNAL_TOKEN_URL
          value: "https://auth.example.com/realms/prod/protocol/openid-connect/token"
        - name: CLIENT_ID
          valueFrom:
            secretKeyRef:
              name: oauth-client
              key: client-id
        - name: CLIENT_SECRET
          valueFrom:
            secretKeyRef:
              name: oauth-client
              key: client-secret
        ports:
        - containerPort: 8080

2. Implement a custom `OAuthBearerTokenProvider`

Compile a small Java class that contacts the local cache instead of the external endpoint.

Before (default configuration):


# producer.properties
sasl.mechanism=OAUTHBEARER
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  oauth.token.endpoint.uri="https://auth.example.com/realms/prod/protocol/openid-connect/token" \
  oauth.client.id="kafka-producer" \
  oauth.client.secret="*****";

After (custom provider):


# producer.properties
sasl.mechanism=OAUTHBEARER
security.protocol=SASL_SSL
sasl.login.callback.handler.class=com.example.kafka.oauth.LocalCacheTokenProvider
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;

Java implementation (simplified):


package com.example.kafka.oauth;

import org.apache.kafka.common.security.oauthbearer.OAuthBearerToken;
import org.apache.kafka.common.security.oauthbearer.OAuthBearerTokenProvider;
import java.io.IOException;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.Base64;
import java.util.Collections;
import java.util.Map;

public class LocalCacheTokenProvider implements OAuthBearerTokenProvider {

    private static final String CACHE_ENDPOINT = "http://oauth-token-cache:8080/token";

    @Override
    public void configure(Map<String, ?> configs, String saslMechanism, Map<String, ?> jaasConfig) {
        // No additional configuration required
    }

    @Override
    public OAuthBearerToken token() throws IOException {
        HttpURLConnection conn = (HttpURLConnection) new URL(CACHE_ENDPOINT).openConnection();
        conn.setRequestMethod("GET");
        int rc = conn.getResponseCode();
        if (rc != 200) {
            throw new IOException("Failed to obtain token from cache, HTTP " + rc);
        }
        String json = new String(conn.getInputStream().readAllBytes());
        // Assume JSON: {"access_token":"...","expires_in":300}
        Map<String, Object> payload = parseJson(json);
        String token = (String) payload.get("access_token");
        long expiresIn = ((Number) payload.get("expires_in")).longValue();
        long expiryMs = System.currentTimeMillis() + expiresIn * 1000;

        return new OAuthBearerToken() {
            @Override public String value() { return token; }
            @Override public Set<String> scope() { return Collections.emptySet(); }
            @Override public long lifetimeMs() { return expiryMs; }
            @Override public String principalName() { return "kafka-producer"; }
            @Override public Long startTimeMs() { return System.currentTimeMillis(); }
            @Override public String tokenType() { return "Bearer"; }
        };
    }

    @Override public void close() { /* no resources */ }

    private Map<String, Object> parseJson(String json) {
        // Use any JSON library; omitted for brevity
        return Collections.emptyMap();
    }
}

Key points:

The provider contacts the internal cache (oauth-token-cache) which always returns a valid token, even when the external OAuth server is unreachable.
The cache service refreshes its token in the background whenever the internet link is restored, ensuring the token never expires for the Kafka client.
Because the provider implements OAuthBearerTokenProvider, the Kafka client no longer performs outbound HTTP calls during token refresh.

Verify

After deploying the cache service and updating the producer/consumer configuration, run the following checks:

Confirm the producer can start without authentication errors:

kafka-producer --bootstrap-server broker.internal:9093 --topic test --property security.protocol=SASL_SSL --property sasl.mechanism=OAUTHBEARER

Expected log snippet:


2024-10-18 09:45:12,003 INFO  [producer-1] org.apache.kafka.clients.producer.KafkaProducer - Producer started, using token expires in 299 seconds

Simulate a prolonged internet outage (block external DNS) and verify that production continues for at least two token lifetimes (≈10 minutes):

sudo iptables -A OUTPUT -p tcp -d auth.example.com -j REJECT
# Run a load test for 12 minutes
kafka-producer-perf-test --topic test --num-records 1000000 --record-size 500 --throughput 5000

Observe that no SaslAuthenticationException appears and the throughput remains stable.

Check the cache service logs to ensure it fell back to the last known token rather than attempting an external request:


2024-10-18 09:46:00,112 INFO  TokenCache - Serving cached token, expires in 250 seconds
2024-10-18 09:48:00,115 INFO  TokenCache - Serving cached token, expires in 230 seconds

Prevent

To avoid recurrence, adopt the following guardrails:

Token lifetime alignment: Configure the external OAuth server to issue tokens with a minimum lifetime (e.g., 1 hour) for air‑gapped consumers, reducing the frequency of refresh attempts.
Health check for token cache: Deploy a Prometheus exporter on the cache service exposing token_cache_success_total and token_cache_failure_total. Alert if failures exceed a threshold.
Deployment validation: Include an integration test that starts a producer with the custom provider, disables outbound internet, and asserts successful message production for at least two token lifetimes.
Configuration immutability: Store the sasl.login.callback.handler.class setting in a ConfigMap that is version‑controlled and audited.
Graceful fallback: Extend the provider to attempt a direct external request only if the cache is unavailable, and log the attempt at WARN level.

FAQ

Why does the producer still attempt to reach the external token endpoint after adding the custom provider?
The default OAuthBearerLoginModule is still loaded unless sasl.login.callback.handler.class is set. Ensure the property is present in producer.properties and that no other JAAS configuration overrides it.
Can I use the same cache service for both producers and consumers?
Yes. The cache is stateless and returns a token scoped to the client principal. Just configure each client’s principalName in the token request payload if you need distinct identities.
What happens if the cache container crashes?
The provider will throw an IOException. Configure the producer’s reconnect.backoff.max.ms and delivery.timeout.ms to tolerate temporary token retrieval failures, and set up a Kubernetes livenessProbe to restart the cache automatically.
Is it safe to store a long‑lived refresh token inside the air‑gapped network?
Store it in a Kubernetes secret with type: Opaque and enable secret encryption at rest. Rotate the refresh token regularly (e.g., weekly) via a secure out‑of‑band process.
How can I monitor token expiry to avoid silent expiration?
Expose the token’s expires_at timestamp from the cache service and create a Prometheus alert when time() - expires_at > 300 seconds, indicating the cache is serving an expired token.

Related Topic Hub: Data Infrastructure Troubleshooting Hub