What causes "User rebalance callback throws an error" in apache/kafka?

Inspect the wrapped cause (KafkaException#getCause) to identify which callback and which exception type actually failed.

apache/kafka · error · KafkaException

User rebalance callback throws an error

Error message

User rebalance callback throws an error

What it means

Thrown at the end of ConsumerCoordinator.onJoinComplete when one of the user-invoked rebalance callbacks (ConsumerRebalanceListener.onPartitionsRevoked / onPartitionsAssigned, or a custom ConsumerPartitionAssignor.onAssignment) raised an exception that is not a KafkaException. KafkaException subclasses are rethrown unchanged (line 472); any other Throwable is wrapped in this KafkaException with the original as the cause. The library does this so that a misbehaving callback surfaces to the caller of poll() rather than silently corrupting the assignment state.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java:474

        maybeUpdateJoinedSubscription(assignedPartitions);

        // Catch any exception here to make sure we could complete the user callback.
        firstException.compareAndSet(null, invokeOnAssignment(assignor, assignment));

        // Reschedule the auto commit starting from now
        if (autoCommitEnabled)
            this.nextAutoCommitTimer.updateAndReset(autoCommitIntervalMs);

        subscriptions.assignFromSubscribed(assignedPartitions);

        // Add partitions that were not previously owned but are now assigned
        firstException.compareAndSet(null, rebalanceListenerInvoker.invokePartitionsAssigned(addedPartitions));

        if (firstException.get() != null) {
            if (firstException.get() instanceof KafkaException) {
                throw (KafkaException) firstException.get();
            } else {
                throw new KafkaException("User rebalance callback throws an error", firstException.get());
            }
        }
    }

    void maybeUpdateSubscriptionMetadata() {
        int version = metadata.updateVersion();
        if (version > metadataSnapshot.version) {
            Cluster cluster = metadata.fetch();

            if (subscriptions.hasPatternSubscription())
                updatePatternSubscription(cluster);

            // Update the current snapshot, which will be used to check for subscription
            // changes that would require a rebalance (e.g. new partitions).
            metadataSnapshot = new MetadataSnapshot(rackId, subscriptions, cluster, version);
        }
    }

View on GitHub (pinned to c31c9215e1)

Solutions

Inspect the wrapped cause (KafkaException#getCause) to identify which callback and which exception type actually failed.
Wrap the body of onPartitionsRevoked / onPartitionsAssigned in try/catch, log the failure, and never let a non-Kafka exception escape the listener.
Keep rebalance callbacks short and side-effect free; move expensive/cleanup work off the rebalance thread.
If the failure is transient (network to cleanup service), catch and retry outside the callback rather than rethrowing.

Example fix

// before
consumer.subscribe(topics, new ConsumerRebalanceListener() {
    public void onPartitionsRevoked(Collection<TopicPartition> p) {
        Files.delete(cleanupPath); // throws IOException -> wrapped here
    }
    public void onPartitionsAssigned(Collection<TopicPartition> p) {}
});

// after
consumer.subscribe(topics, new ConsumerRebalanceListener() {
    public void onPartitionsRevoked(Collection<TopicPartition> p) {
        try { Files.deleteIfExists(cleanupPath); }
        catch (IOException e) { log.warn("cleanup failed for {}", p, e); }
    }
    public void onPartitionsAssigned(Collection<TopicPartition> p) {}
});

Defensive patterns

Strategy: try-catch

Type guard

static boolean isUserCallbackFailure(KafkaException e) {
    return e.getCause() != null
        && e.getMessage() != null
        && e.getMessage().startsWith("User rebalance callback throws an error");
}

Try / catch

try {
    consumer.poll(Duration.ofMillis(1000));
} catch (KafkaException e) {
    if (isUserCallbackFailure(e)) {
        // e.getCause() is the real exception your ConsumerRebalanceListener.onPartitionsAssigned threw
        log.error("Rebalance callback failed", e.getCause());
        // repair local state, optionally rethrow if fatal
    } else {
        throw e;
    }
}

Prevention

Wrap the body of every ConsumerRebalanceListener method in try/catch and never propagate
Keep onPartitionsAssigned fast, idempotent, and side-effect-free
Log inside callbacks but swallow non-fatal errors so a rebalance can complete
Avoid blocking I/O or long computation inside rebalance callbacks

When it happens

Trigger: A consumer using subscribe() with a ConsumerRebalanceListener completes a group join; during onJoinComplete the COOPERATIVE-path invokePartitionsRevoked (line 445), invokeOnAssignment (line 459), or invokePartitionsAssigned (line 468) throws a checked/non-Kafka exception (e.g. IOException, NullPointerException, IllegalStateException). The first such exception is captured in firstException and rethrown at line 474 on the next poll()/commit that triggers the join completion.

Common situations: Listener callbacks performing I/O (closing files, calling external cleanup HTTP APIs, DB deletes) that throw IOException; deserializers/assignors with NPEs on null userdata; callbacks that throw RuntimeExceptions from third-party libraries; stateful cleanup logic that fails because resources were already closed.

Related errors

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