apache/kafka · error · TopicAuthorizationException

Not authorized to access topics: ${unauthorizedTopics}

Error message

Not authorized to access topics: ${unauthorizedTopics}

What it means

Thrown as TopicAuthorizationException by the legacy/sync TopicMetadataFetcher.getTopicMetadata when the broker's MetadataResponse lists one or more topics in cluster.unauthorizedTopics(). The client surfaced it instead of silently dropping the topic because the principal authenticated correctly but lacks the Describe ACL on those topics. Most commonly raised from Consumer.partitionsFor() / listTopics() paths.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataFetcher.java:113

        // Save the round trip if no topics are requested.
        if (!request.isAllTopics() && request.emptyTopicList())
            return Collections.emptyMap();

        long attempts = 0L;
        do {
            RequestFuture<ClientResponse> future = sendMetadataRequest(request);
            client.poll(future, timer);

            if (future.failed() && !future.isRetriable())
                throw future.exception();

            if (future.succeeded()) {
                MetadataResponse response = (MetadataResponse) future.value().responseBody();
                Cluster cluster = response.buildCluster();

                Set<String> unauthorizedTopics = cluster.unauthorizedTopics();
                if (!unauthorizedTopics.isEmpty())
                    throw new TopicAuthorizationException(unauthorizedTopics);

                boolean shouldRetry = false;
                Map<String, Errors> errors = response.errors();
                if (!errors.isEmpty()) {
                    // if there were errors, we need to check whether they were fatal or whether
                    // we should just retry

                    log.debug("Topic metadata fetch included errors: {}", errors);

                    for (Map.Entry<String, Errors> errorEntry : errors.entrySet()) {
                        String topic = errorEntry.getKey();
                        Errors error = errorEntry.getValue();

                        if (error == Errors.INVALID_TOPIC_EXCEPTION)
                            throw new InvalidTopicException("Topic '" + topic + "' is invalid");
                        else if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION)
                            // if a requested topic is unknown, we just continue and let it be absent
                            // in the returned map

View on GitHub (pinned to c31c9215e1)

Solutions

  1. Grant the DESCRIBE (and READ for consume) ACL: kafka-acls --add --allow-principal User:bob --operation Describe --operation Read --topic orders.
  2. Verify the principal the consumer actually presents: enable DEBUG org.apache.kafka.clients.NetworkClient / check broker server.log for the PrincipalLoader output.
  3. Confirm sasl.mechanism, sasl.jaas.config, and security.protocol match what the broker expects (mismatch silently authenticates as ANONYMOUS).
  4. If using prefix ACLs, ensure the topic literal falls under the granted prefix (kafka-acls --resource-pattern-type prefixed).

Example fix

// before: consumer runs as wrong principal, no ACL
bin/kafka-acls.sh --bootstrap-server broker:9092 \
  --list --principal User:bob   # (empty)

// after: grant Describe + Read on the topic
bin/kafka-acls.sh --bootstrap-server broker:9092 --add \
  --allow-principal User:bob --operation Describe --operation Read \
  --topic orders
Defensive patterns

Strategy: try-catch

Validate before calling

// Client-side ACL preflight (optional, heavier): verify READ/DESCRIBE before subscribing.
try (var admin = org.apache.kafka.clients.admin.AdminClient.create(commonProps)) {
    java.util.Collection<org.apache.kafka.common.acl.AclBinding> have =
        admin.describeAcls(org.apache.kafka.common.acl.AccessControlEntryFilter.ANY).values().get();
    // intersect with required (topic, READ/DESCRIBE) before consumer.subscribe(...)
}

Type guard

import org.apache.kafka.common.errors.TopicAuthorizationException;

/** Narrows a thrown Throwable to a TopicAuthorizationException and extracts unauthorized topic names. */
static java.util.Optional<java.util.Set<String>> unauthorizedTopics(Throwable t) {
    if (t instanceof TopicAuthorizationException) {
        return java.util.Optional.of(((TopicAuthorizationException) t).unauthorizedTopics());
    }
    return java.util.Optional.empty();
}

Try / catch

try {
    consumer.subscribe(singleton(topic));
    consumer.poll(Duration.ofMillis(1000));
} catch (org.apache.kafka.common.errors.TopicAuthorizationException e) {
    java.util.Set<String> unauthorized = e.unauthorizedTopics();
    log.error("Principal lacks READ/DESCRIBE on {}; provision ACLs via kafka-acls", unauthorized);
    // surface to ops; do not blindly retry — this will not self-heal without ACL changes
    throw e;
}

Prevention

When it happens

Trigger: Calling consumer.partitionsFor("topic"), consumer.listTopics(), or any path that triggers a MetadataRequest where the authenticated principal lacks the DESCRIBE operation ACL on at least one requested topic. Also raised when a topic name is correct but the ACL was scoped to a different principal or prefixed differently.

Common situations: Service principals mismatched between producer and consumer (e.g. producer runs as user=alice, consumer as user=bob without a matching ACL); authorizer configured (SimpleAclAuthorizer / StandardAuthorizer in KRaft) but ACLs never granted; SASL/JAAS config wrong so the client authenticates as ANONYMOUS; cross-environment promotion without copying ACLs.

Related errors

AI-assisted analysis of apache/kafka@c31c9215e1 (2026-08-03). Data as JSON: /data/errors/72eb846138339b99.json. Report an issue: GitHub.