public void acknowledge(final ConsumerRecord<K, V> record, 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 in the current batch.

@param record The record to acknowledge
@param type The acknowledge type which indicates whether it was processed successfully
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java