apache/kafka · error · IllegalStateException

All records must be acknowledged in explicit acknowledgement

Error message

All records must be acknowledged in explicit acknowledgement mode.

What it means

Thrown by ensureInFlightAcknowledgedIfExplicitAcknowledgement() (line 1218) at the start of poll() when the acknowledgement mode is EXPLICIT and currentFetch.checkAllInFlightAreAcknowledged() returns false. In explicit mode every record fetched in the previous batch must be acknowledged (ACCEPT/REJECT/RELEASE) before the next poll, otherwise the client refuses to fetch more records and surfaces the gap as an IllegalStateException.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:1218

    /**
     * If the acknowledgement mode is IMPLICIT, acknowledges all records in the current batch.
     */
    private void acknowledgeBatchIfImplicitAcknowledgement() {
        // If IMPLICIT, acknowledge all records
        if (acknowledgementMode == ShareAcknowledgementMode.IMPLICIT) {
            currentFetch.acknowledgeAll(AcknowledgeType.ACCEPT);
        }
    }

    /**
     * If the acknowledgement mode is EXPLICIT, ensure that all in-flight records have been acknowledged.
     */
    private void ensureInFlightAcknowledgedIfExplicitAcknowledgement() {
        if (acknowledgementMode == ShareAcknowledgementMode.EXPLICIT) {
            if (!currentFetch.checkAllInFlightAreAcknowledged()) {
                // We cannot leave unacknowledged records in EXPLICIT acknowledgement mode, so we throw an exception to the application.
                throw new IllegalStateException("All records must be acknowledged in explicit acknowledgement mode.");
            }
        }
    }

    /**
     * Returns any ready acknowledgements to be sent to the cluster.
     */
    private Map<TopicIdPartition, NodeAcknowledgements> acknowledgementsToSend() {
        return currentFetch.takeAcknowledgedRecords();
    }

    /**
     * Called to verify if the acknowledgement mode is EXPLICIT, else throws an exception.
     */
    private void ensureExplicitAcknowledgement() {
        if (acknowledgementMode == ShareAcknowledgementMode.IMPLICIT) {
            throw new IllegalStateException("Implicit acknowledgement of delivery is being used.");
        }

View on GitHub (pinned to c31c9215e1)

Solutions

1. In explicit mode, acknowledge every record returned by each poll() — wrap processing in try/finally and call acknowledge() for each ConsumerRecord.
2. If you cannot ack per-record, switch the consumer to implicit acknowledgement mode by setting share.acknowledgement.mode=implicit, which auto-acks the whole batch.
3. Use consumer.acknowledgeAll(AcknowledgeType.ACCEPT) (if exposed via your API surface) to ack the remaining records before the next poll when you truly mean to accept all.
4. Audit the processing pipeline for early returns/continues/exceptions that skip the acknowledge call.

Example fix

// before
for (ConsumerRecord<String,String> r : records) {
    if (r.value() == null) continue; // never acknowledged -> next poll throws
    process(r);
    consumer.acknowledge(r);
}

// after
for (ConsumerRecord<String,String> r : records) {
    try {
        process(r);
        consumer.acknowledge(r, AcknowledgeType.ACCEPT);
    } catch (Exception e) {
        consumer.acknowledge(r, AcknowledgeType.REJECT);
    }
}

// In EXPLICIT acknowledgement mode, ensure every in-flight record is acknowledged before the next poll().
// Track unacked records locally so the precondition is never violated.
java.util.Set<org.apache.kafka.clients.consumer.ShareRecord> inFlight =
    java.util.Collections.newSetFromMap(new java.util.concurrent.ConcurrentHashMap<>());

for (org.apache.kafka.clients.consumer.ShareRecord<K,V> r : records) {
    inFlight.add(r);
    // ... process ...
    consumer.acknowledge(r, org.apache.kafka.clients.consumer.AcknowledgeType.ACCEPT);
    inFlight.remove(r);
}
if (!inFlight.isEmpty()) {
    throw new IllegalStateException("Cannot poll: " + inFlight.size() + " records still unacknowledged");
}
consumer.poll(java.time.Duration.ofMillis(100));

null

try {
    consumer.poll(java.time.Duration.ofMillis(100));
} catch (IllegalStateException e) {
    if (e.getMessage() != null && e.getMessage().contains("must be acknowledged")) {
        // drain unacked records (acknowledge them) before retrying the poll
        drainAndAcknowledgeInFlight(consumer);
    } else {
        throw e;
    }
}

When it happens

Trigger: Calling consumer.poll(...) in explicit acknowledgement mode without having called consumer.acknowledge(...) on every record returned by the previous poll(). Also triggered when acknowledgements are partial (only some records of a batch were acknowledged) or when an earlier acknowledge() call targeted a record not in the current fetch.

Common situations: Application filters records and skips acknowledge() for the filtered-out ones; exception in the processing loop that bypasses the finally-block acknowledge; partial batch retries where only failing records are re-acked; migrating from implicit mode without updating the processing loop to ack every record.

Related errors

• Consumer is not subscribed to any topics.
• KafkaShareConsumer methods are not accessible from user-defi
• Implicit acknowledgement of delivery is being used.
• The record cannot be acknowledged.
• The record cannot be acknowledged.

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