apache/kafka · error · TopicAuthorizationException

Not authorized to access topics: ${Set.of(tp.topic())}

Error message

Not authorized to access topics: ${Set.of(tp.topic())}

What it means

Thrown by ShareFetchCollector.handleInitializeErrors when a fetch response carries Errors.TOPIC_AUTHORIZATION_FAILED for a topic-partition. The share consumer has no read ACL for that topic, so the collector logs the failing partition and raises TopicAuthorizationException naming the offending topic(s). It is the share group's enforcement of Kafka ACLs (cluster READ + topic DESCRIBE/READ) at fetch time.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetchCollector.java:175

                error == Errors.REPLICA_NOT_AVAILABLE ||
                error == Errors.KAFKA_STORAGE_ERROR ||
                error == Errors.FENCED_LEADER_EPOCH ||
                error == Errors.OFFSET_NOT_AVAILABLE) {
            log.debug("Error in fetch for partition {}: {}", tp, error.exceptionName());
            requestMetadataUpdate(metadata, subscriptions, tp.topicPartition());
        } else if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
            log.warn("Received unknown topic or partition error in fetch for partition {}.", tp);
            requestMetadataUpdate(metadata, subscriptions, tp.topicPartition());
        } else if (error == Errors.UNKNOWN_TOPIC_ID) {
            log.warn("Received unknown topic ID error in fetch for partition {}.", tp);
            requestMetadataUpdate(metadata, subscriptions, tp.topicPartition());
        } else if (error == Errors.INCONSISTENT_TOPIC_ID) {
            log.warn("Received inconsistent topic ID error in fetch for partition {}.", tp);
            requestMetadataUpdate(metadata, subscriptions, tp.topicPartition());
        } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
            // Log the actual partition and not just the topic to help with ACL propagation issues in large clusters
            log.warn("Not authorized to read from partition {}.", tp.topicPartition());
            throw new TopicAuthorizationException(Set.of(tp.topic()));
        } else if (error == Errors.UNKNOWN_LEADER_EPOCH) {
            log.debug("Received unknown leader epoch error in fetch for partition {}.", tp);
        } else if (error == Errors.UNKNOWN_SERVER_ERROR) {
            log.warn("Unknown server error while fetching topic-partition {}.",
                    tp.topicPartition());
        } else if (error == Errors.CORRUPT_MESSAGE) {
            throw new KafkaException("Encountered corrupt message when fetching topic-partition "
                    + tp.topicPartition());
        } else {
            throw new IllegalStateException("Unexpected error code " + error.code()
                    + " while fetching from topic-partition " + tp.topicPartition());
        }
    }
}

View on GitHub (pinned to c31c9215e1)

Solutions

  1. Grant READ ACL to the principal on the topic (and DESCRIBE on the cluster/topic): kafka-acls.sh --add --operation Read --principal ... --topic <name>.
  2. Verify the configured principal matches the one granted ACLs (sasl.jaas.config / client credentials).
  3. If using a pattern subscription, ensure ACLs cover all matched topics or narrow the pattern.
  4. Re-fetch metadata after ACL changes and retry; ACL propagation may lag briefly in large clusters.

Example fix

# before
# principal 'User:app' has no ACL on topic 'orders'

# after
bin/kafka-acls.sh --bootstrap-server broker:9092 \
  --add --allow-principal User:app --operation Read --topic orders
bin/kafka-acls.sh --bootstrap-server broker:9092 \
  --add --allow-principal User:app --operation Describe --topic orders
Defensive patterns

Strategy: try-catch

Validate before calling

// Validate ACLs before subscribing when possible using the AdminClient.
try (Admin admin = Admin.create(commonProps)) {
    DescribeAclsResult res = admin.describeAcls(
        AclBindingFilter.forResourceType(ResourceType.TOPIC));
    Set<String> allowedTopics = res.values().get().stream()
        .map(b -> b.pattern().name())
        .collect(Collectors.toSet());
    topics.retainAll(allowedTopics);
    if (topics.isEmpty()) throw new IllegalStateException("No topics authorized");
}
consumer.subscribe(topics);

Try / catch

try {
    consumer.subscribe(topics);
    consumer.poll(Duration.ofSeconds(5));
} catch (org.apache.kafka.common.errors.TopicAuthorizationException e) {
    // Missing READ/DESCRIBE ACL on the listed topics.
    log.error("Authorization failed for topics: {}", e.unauthorizedTopics(), e);
    // Surface to operator so they can grant ACLs:
    //   bin/kafka-acls.sh --add --allow-principal User:... --operation READ --topic <t>
    alertOps(e.unauthorizedTopics());
}

Prevention

When it happens

Trigger: The principal configured via SASL/KafkaClient lacks the READ operation ACL on the topic being fetched in share mode; an ACL was revoked between subscribe and poll; a topic-pattern subscription included a topic the client is not authorized to read.

Common situations: Service principal not granted READ on the topic; ACLs scoped per-topic but the client subscribes broadly; recent ACL cleanup/rotation removed access; cross-environment (dev->prod) config copied without adjusting principal; mTLS/SASL credentials point to the wrong user.

Related errors

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