apache/kafka · error · KafkaException

Failed to close kafka consumer

Error message

Failed to close kafka consumer

What it means

Thrown from KafkaConsumer.close when one of the components being closed (coordinator, fetcher, network client, interceptors, metrics, etc.) throws an exception that is not an InterruptException and is not being swallowed. The message wraps the first captured exception in a KafkaException so callers see that shutdown failed, with the cause holding the real error. It surfaces resource cleanup failures during teardown.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java:1190

            // This is a blocking call bound by the time remaining in closeTimer
            swallow(log, Level.ERROR, "Failed to close fetcher with a timeout(ms)=" + closeTimer.timeoutMs(), () -> fetcher.close(closeTimer), firstException);
        }

        closeQuietly(interceptors, "consumer interceptors", firstException);
        closeQuietly(kafkaConsumerMetrics, "kafka consumer metrics", firstException);
        closeQuietly(fetchMetricsManager, "kafka fetch metrics", firstException);
        closeQuietly(metrics, "consumer metrics", firstException);
        closeQuietly(client, "consumer network client", firstException);
        closeQuietly(deserializers, "consumer deserializers", firstException);
        clientTelemetryReporter.ifPresent(reporter -> closeQuietly(reporter, "consumer telemetry reporter", firstException));
        AppInfoParser.unregisterAppInfo(CONSUMER_JMX_PREFIX, clientId, metrics);
        log.debug("Kafka consumer has been closed");
        Throwable exception = firstException.get();
        if (exception != null && !swallowException) {
            if (exception instanceof InterruptException) {
                throw (InterruptException) exception;
            }
            throw new KafkaException("Failed to close kafka consumer", exception);
        }
    }

    /**
     * Set the fetch position to the committed position (if there is one)
     * or reset it using the offset reset policy the user has configured.
     *
     * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
     * @throws NoOffsetForPartitionException If no offset is stored for a given partition and no offset reset policy is
     *             defined
     * @return true iff the operation completed without timing out
     */
    private boolean updateFetchPositions(final Timer timer) {
        // If any partitions have been truncated due to a leader change, we need to validate the offsets
        offsetFetcher.validatePositionsIfNeeded();

        cachedSubscriptionHasAllFetchPositions = subscriptions.hasAllFetchPositions();
        if (cachedSubscriptionHasAllFetchPositions) return true;

View on GitHub (pinned to c31c9215e1)

Solutions

1. Inspect the cause of the KafkaException to find which component failed and address that (broker reachability, interceptor bug, deserializer).
2. Use close(CloseOptions) with an adequate timeout and group membership operation appropriate to your shutdown path.
3. Ensure the consumer thread is not interrupted during close; catch InterruptException separately and re-interrupt cleanly.
4. Log and continue on shutdown paths where full cleanup is best-effort, after fixing the root cause.

Example fix

// before
consumer.close(); // surfaces KafkaException("Failed to close kafka consumer", cause)

// after
try {
    consumer.close(new CloseOptions()
        .timeout(Duration.ofSeconds(10))
        .groupMembershipOperation(GroupMembershipOperation.LEAVE_GROUP));
} catch (KafkaException e) {
    log.warn("Consumer close failed; cause={}", e.getCause());
}

// No pre-call validation prevents this; the failure is during close itself.
// Mitigate by passing a generous, explicit close timeout:
consumer.close(CloseOptions.timeout(Duration.ofSeconds(30))
    .groupMembershipOperation(GroupMembershipOperation.LEAVE));

// Closing must always succeed from the caller's perspective; capture, log, swallow:
try {
    consumer.close(Duration.ofSeconds(30));
} catch (KafkaException e) {
    if (e.getMessage().contains("Failed to close kafka consumer")) {
        log.error("Consumer close failed (suppressed); cause: {}", e.getCause());
        // Optionally call wakeup() and re-attempt once, then give up — never let close abort shutdown.
    } else throw e;
} finally {
    consumer = null;
}

When it happens

Trigger: Coordinator.close() failing because a LeaveGroup request could not complete; fetcher.close() failing on a network error; metrics/network client close throwing an IOException; deserializer or interceptor close throwing.

Common situations: Closing the consumer during broker outage or network partition; close() invoked from a shutdown hook racing with JVM teardown; partial outage leaving sockets half-open; interceptor bugs surfacing only on close; calling close() after an InterruptException interrupted the consumer thread.

Related errors

• This consumer has already been closed.
• Timeout of {}ms expired before the position for partition {}
• Timeout of {}ms expired before the last committed offset for
• Invalid value null for configuration key.deserializer: must
• Invalid value null for configuration value.deserializer: mus

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