apache/kafka · error · IllegalStateException

The record cannot be acknowledged.

Error message

The record cannot be acknowledged.

What it means

Thrown by ShareInFlightBatch.acknowledge when the record's offset is not present in inFlightRecords. The in-flight batch is a TreeMap keyed by offset; acknowledging an offset that was never added (via addRecord) or that has already been drained by a prior commit/clear is a programming error and raises this IllegalStateException. It is the lowest-level guard beneath ShareFetch.acknowledge.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareInFlightBatch.java:70

    }

    public void addAcknowledgement(long offset, AcknowledgeType type) {
        acknowledgements.add(offset, type);
        if (type == AcknowledgeType.RENEW) {
            checkForRenewAcknowledgements = true;
        }
    }

    public void acknowledge(ConsumerRecord<K, V> record, AcknowledgeType type) {
        if (inFlightRecords.get(record.offset()) != null) {
            acknowledgements.add(record.offset(), type);
            acknowledgedRecords.add(record.offset());
            if (type == AcknowledgeType.RENEW) {
                checkForRenewAcknowledgements = true;
            }
            return;
        }
        throw new IllegalStateException("The record cannot be acknowledged.");
    }

    public void acknowledgeAll(AcknowledgeType type) {
        for (Map.Entry<Long, ConsumerRecord<K, V>> entry : inFlightRecords.entrySet()) {
            if (acknowledgements.addIfAbsent(entry.getKey(), type)) {
                acknowledgedRecords.add(entry.getKey());
            }
        }
        if (type == AcknowledgeType.RENEW) {
            checkForRenewAcknowledgements = true;
        }
    }

    public boolean checkAllInFlightAreAcknowledged() {
        return inFlightRecords.size() == acknowledgedRecords.size();
    }

    public void addRecord(ConsumerRecord<K, V> record) {

View on GitHub (pinned to c31c9215e1)

Solutions

1. Track which offsets you have already acknowledged (dedupe set) and skip duplicates.
2. Only acknowledge offsets returned by the most recent poll() against the current in-flight batch.
3. Avoid concurrent acknowledge calls from multiple threads; serialize acknowledgements with poll().
4. Increase share.group.acquire.timeout.ms if records are leaving the in-flight set before processing completes.

Example fix

// before
records.forEach(r -> {
    exec.submit(() -> consumer.acknowledge(r, AcknowledgeType.ACCEPT));
});

// after
records.forEach(r -> {
    process(r);
    consumer.acknowledge(r, AcknowledgeType.ACCEPT);
});
consumer.commitAcknowledgements();

// Track acknowledged offsets per (topic, partition) so we never call acknowledge()
// twice for the same offset, and never call it after the in-flight window has moved on.
private final Map<String, Set<Long>> acked = new ConcurrentHashMap<>();

void safeAcknowledge(ConsumerRecord<K,V> r, AcknowledgeType type) {
    Set<Long> s = acked.computeIfAbsent(r.topic() + ":" + r.partition(), k -> ConcurrentHashMap.newKeySet());
    if (s.add(r.offset())) {            // false => already acknowledged
        consumer.acknowledge(r, type);
    }
}

try {
    consumer.acknowledge(record, type);
} catch (IllegalStateException e) {
    // record.offset() is not in inFlightRecords: it was already acknowledged,
    // or the in-flight batch has expired/released. Idempotent no-op.
    log.debug("Record {}-{}@{} not in-flight; skipping acknowledge",
            record.topic(), record.partition(), record.offset(), e);
}

When it happens

Trigger: Calling acknowledge(record) twice for the same record after the second call has the record removed; acknowledging a record whose offset was added as a gap (addGap) rather than a record; acknowledging an offset that belongs to a different batch than the one being mutated; acknowledging after commitAcknowledgements has reset the batch.

Common situations: Duplicate processing pipelines acknowledging the same offset; test code constructing synthetic ConsumerRecords; race between an async acknowledgement thread and the main poll loop that replaces the batch; acquisition-lock expiry having moved records out of in-flight state.

Related errors

• The record cannot be acknowledged.
• Consumer is not subscribed to any topics.
• Telemetry is not enabled. Set config `${ConsumerConfig.ENABL
• Failed to close Kafka share consumer
• This consumer has already been closed.

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