apache/kafka · error · TopicAuthorizationException

Not authorized to access topics: ${unauthorizedTopics}

Error message

Not authorized to access topics: ${unauthorizedTopics}

What it means

Async-consumer counterpart of error 232. Thrown as TopicAuthorizationException by TopicMetadataRequestManager.TopicMetadataRequestHandler.handleTopicMetadataResponse when cluster.unauthorizedTopics() from the MetadataResponse is non-empty. The new consumer (broker-side network thread / RequestManager architecture) surfaces the same authorization failure as the legacy TopicMetadataFetcher, but from the async request pipeline.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/TopicMetadataRequestManager.java:246

                Map<String, List<PartitionInfo>> res = handleTopicMetadataResponse((MetadataResponse) response.responseBody());
                future.complete(res);
                inflightRequests.remove(this);
            } catch (Exception e) {
                handleError(e, response.receivedTimeMs());
            }
        }

        private void completeFutureAndRemoveRequest(final Throwable throwable) {
            future.completeExceptionally(throwable);
            inflightRequests.remove(this);
        }

        private Map<String, List<PartitionInfo>> handleTopicMetadataResponse(final MetadataResponse response) {
            Cluster cluster = response.buildCluster();

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

            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
                        continue;

View on GitHub (pinned to c31c9215e1)

Solutions

1. Grant DESCRIBE (and READ for consume) on the topic to the principal the consumer presents: kafka-acls --add --allow-principal User:<p> --operation Describe --operation Read --topic <t>.
2. Double-check sasl.mechanism / sasl.jaas.config / security.protocol — a silent ANONYMOUS auth is the most common cause.
3. Use kafka-acls --list --principal <p> to confirm the ACL is actually attached; if using prefix ACLs verify the topic literal falls under the prefix.
4. Enable DEBUG org.apache.kafka.clients.consumer.internals.TopicMetadataRequestManager to see the unauthorized topic set.

Example fix

// before
props.put("sasl.mechanism", "PLAIN");
props.put("sasl.jaas.config",
    "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"alice\" password=\"***\";");
// broker expects principal User:alice but ACL was granted to User:bob

// after: align principal with the granted ACL
bin/kafka-acls.sh --bootstrap-server broker:9092 --add \
  --allow-principal User:alice --operation Describe --operation Read --topic orders

// Optional ACL preflight via AdminClient before subscribe (heavier than trusting the error path).
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();
    // verify required (topic, READ/DESCRIBE) bindings are present
}

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

/** Narrows a throwable to TopicAuthorizationException and returns its unauthorized topic set. */
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 {
    consumer.poll(Duration.ofMillis(1000));
} catch (org.apache.kafka.common.errors.TopicAuthorizationException e) {
    java.util.Set<String> unauthorized = e.unauthorizedTopics();
    log.error("Consumer principal not authorized for {}; run kafka-acls to grant READ/DESCRIBE", unauthorized);
    // Not self-healing: surface to ops, halt, or fall back to an authorized topic subset.
    throw e;
}

When it happens

Trigger: Any async-consumer metadata operation (partitionsFor, listTopics, subscribe that triggers metadata refresh) where the authenticated principal lacks the DESCRIBE ACL on at least one requested topic. Raised on the network thread and propagated via the request's CompletableFuture to the application thread.

Common situations: Same root causes as 232 — principal mismatch (ANONYMOUS via misconfigured SASL/JAAS), ACLs granted for a different user or prefix, principal mapping (user:alice vs User:alice) configured differently on the broker, cross-cluster migration without copying ACLs.

Related errors

• Not authorized to access topics: ${unauthorizedTopics}
• Topic '${topic}' is invalid
• Unexpected error fetching metadata for topic ${topic}
• Failed to get offsets by times in {}ms
• Topic '${topic}' is invalid

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