Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java (Java)

Code before:
            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 by its topic, partition and offset in the current batch.
     *
     * @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 acknowledgment 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();
            if (tip.topic().equals(topic) && (tip.partition() == partition)) {
                tipBatch.getValue().addAcknowledgement(offset, type);
                return;
            }
        }
        throw new IllegalStateException("The record cannot be acknowledged.");
    }

    /**
     * Acknowledge all records in the current batch. If any records in the batch already have
     * been acknowledged, those acknowledgements are not overwritten.
     *
     * @param type The acknowledge type which indicates whether it was processed successfully
     */
    public void acknowledgeAll(final AcknowledgeType type) {
        batches.forEach((tip, batch) -> batch.acknowledgeAll(type));
    }

    /**
     * Removes all acknowledged records from the in-flight records and returns the map of acknowledgements
     * to send. If some records were not acknowledged, the in-flight records will not be empty after this
     * method.
     *
     * @return The map of acknowledgements to send, along with node information
     */
    public Map<TopicIdPartition, NodeAcknowledgements> takeAcknowledgedRecords() {
        Map<TopicIdPartition, NodeAcknowledgements> acknowledgementMap = new LinkedHashMap<>();
        batches.forEach((tip, batch) -> {
            int nodeId = batch.nodeId();
            Acknowledgements acknowledgements = batch.takeAcknowledgedRecords();
            if (!acknowledgements.isEmpty())

Review comment (performance):
The checking here is not tight enough. If the user calls `acknowledge(ConsumerRecord, AcknowledgeType)`, the code makes sure that the in-flight records include the offset (see `ShareInFlightBatch.acknowledge(ConsumerRecord, AcknowledgeType)`). However, in this case, there is no validation of the offset. The handling of an exceptional batch needs to be a bit more sophisticated I think. It should not be possible to use `ShareConsumer.acknowledge(ConsumerRecord, AcknowledgeType)` with an exceptional batch. It should only be possible to use `ShareConsumer.acknowlege(String, int, long, AcknowledgeType)` and only for the specific parameters that were retrieved from the `RecordDeserializationException`.

Code after:
            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 by its topic, partition and offset in the current batch.
     *
     * @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 acknowledgment 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 &&
                exception.offsets().contains(offset)) {

                tipBatch.getValue().addAcknowledgement(offset, type);
                return;
            }
        }
        throw new IllegalStateException("The record cannot be acknowledged.");
    }

    /**
     * Acknowledge all records in the current batch. If any records in the batch already have
     * been acknowledged, those acknowledgements are not overwritten.
     *
     * @param type The acknowledge type which indicates whether it was processed successfully
     */
    public void acknowledgeAll(final AcknowledgeType type) {
        batches.forEach((tip, batch) -> batch.acknowledgeAll(type));
    }

    /**
     * Removes all acknowledged records from the in-flight records and returns the map of acknowledgements
     * to send. If some records were not acknowledged, the in-flight records will not be empty after this
     * method.
     *
     * @return The map of acknowledgements to send, along with node information
     */
    public Map<TopicIdPartition, NodeAcknowledgements> takeAcknowledgedRecords() {
        Map<TopicIdPartition, NodeAcknowledgements> acknowledgementMap = new LinkedHashMap<>();