What causes "The record cannot be acknowledged." in apache/kafka?

Acknowledge records within the same poll() processing cycle, before the next call to poll() or commitAcknowledgements().

apache/kafka · error · IllegalStateException

The record cannot be acknowledged.

Error message

The record cannot be acknowledged.

What it means

Thrown by ShareFetch.acknowledge(ConsumerRecord, AcknowledgeType) when the supplied ConsumerRecord's topic and partition do not match any TopicIdPartition currently held in the in-flight batches. The share consumer only allows acknowledging records that belong to the active fetch batch; if the record is from a previous batch or never fetched, the lookup fails and this IllegalStateException is raised to signal a programming error.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java:166

            acquisitionLockTimeoutMs = acquisitionLockTimeoutMsRenewed;
        }
    }

    /**
     * Acknowledge a single record in the current batch.
     *
     * @param record The record to acknowledge
     * @param type The acknowledge type which indicates whether it was processed successfully
     */
    public void acknowledge(final ConsumerRecord<K, V> record, final AcknowledgeType type) {
        for (Map.Entry<TopicIdPartition, ShareInFlightBatch<K, V>> tipBatch : batches.entrySet()) {
            TopicIdPartition tip = tipBatch.getKey();
            if (tip.topic().equals(record.topic()) && (tip.partition() == record.partition())) {
                tipBatch.getValue().acknowledge(record, type);
                return;
            }
        }
        throw new IllegalStateException("The record cannot be acknowledged.");
    }

    /**
     * Acknowledge a single record which experienced an exception during its delivery by its topic, partition
     * and offset in the current batch. This method is specifically for overriding the default acknowledge
     * type for records whose delivery failed.
     *
     * @param topic     The topic of the record to acknowledge
     * @param partition The partition of the record
     * @param offset    The offset of the record
     * @param type      The acknowledge type which indicates whether it was processed successfully
     */
    public void acknowledge(final String topic, final int partition, final long offset, final AcknowledgeType type) {
        for (Map.Entry<TopicIdPartition, ShareInFlightBatch<K, V>> tipBatch : batches.entrySet()) {
            TopicIdPartition tip = tipBatch.getKey();
            ShareInFlightBatchException exception = tipBatch.getValue().getException();
            if (tip.topic().equals(topic) && (tip.partition() == partition) &&
                exception != null &&

View on GitHub (pinned to c31c9215e1)

Solutions

Acknowledge records within the same poll() processing cycle, before the next call to poll() or commitAcknowledgements().
Increase share.group.acquire.timeout.ms so the acquisition lock survives your processing latency.
Do not retain or reuse ConsumerRecord references across polls; track only the offsets you actually need to acknowledge late.
If processing asynchronously, ensure acknowledgements are flushed before the acquisition lock expires or before the next poll drains the batch.

Example fix

// before
List<ConsumerRecord> buffered = new ArrayList<>();
for (ConsumerRecord r : records) buffered.add(r);  // acknowledged later, batch may be gone
consumer.acknowledge(buffered.get(0), AcknowledgeType.ACCEPT);

// after
for (ConsumerRecord r : records) {
    process(r);
    consumer.acknowledge(r, AcknowledgeType.ACCEPT);
}
consumer.commitAcknowledgements();

Defensive patterns

Strategy: validation

Validate before calling

// Only acknowledge records that came from the most recent poll() batch,
// exactly once, before the next poll() clears them.
private final Set<String> pendingKeys = ConcurrentHashMap.newKeySet();

ConsumerRecords<K,V> records = consumer.poll(Duration.ofSeconds(5));
for (ConsumerRecord<K,V> r : records) {
    String key = r.topic() + ":" + r.partition() + ":" + r.offset();
    pendingKeys.add(key);
}

for (ConsumerRecord<K,V> r : records) {
    String key = r.topic() + ":" + r.partition() + ":" + r.offset();
    if (pendingKeys.remove(key)) {            // already-acked records are filtered out
        consumer.acknowledge(r, AcknowledgeType.ACCEPT);
    }
}

Try / catch

try {
    consumer.acknowledge(record, AcknowledgeType.ACCEPT);
} catch (IllegalStateException e) {
    // Record is not in the current in-flight batch: it was already acknowledged,
    // or it belonged to a previous batch that has been released. Treat as idempotent no-op.
    log.debug("Skipping duplicate/stale acknowledgement for {}-{}@{}",
            record.topic(), record.partition(), record.offset(), e);
}

Prevention

Never hold ConsumerRecord references across poll() calls; the in-flight batch is replaced each poll.
Acknowledge every record from a batch exactly once; track offsets you have already acked.
Avoid calling acknowledge() after an explicit acknowledgeAll() on the same batch.
Call acknowledge() inside the same thread that called poll().

When it happens

Trigger: Calling consumer.acknowledge(record) with a record that came from an earlier poll() invocation whose batch was already completed/cleared; acknowledging a record after a rebalance or acquisition-lock expiry that drained the in-flight batches; acknowledging a record that was reconstructed or fabricated by application code rather than returned by poll().

Common situations: Application caches ConsumerRecords across poll() calls and acknowledges them later; share-group acquisition lock times out (record no longer in-flight) before the app calls acknowledge; processing records in a separate thread after the main thread already issued a new poll; mixing records from multiple consumer instances.

Related errors

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