public void replayEndTransactionMarker(
        long producerId,
        TransactionResult result
    ) throws RuntimeException {
        Offsets pendingOffsets = pendingTransactionalOffsets.remove(producerId);

        if (pendingOffsets == null) {
            log.debug("Replayed end transaction marker with result {} for producer id {} but " +
                "no pending offsets are present. Ignoring it.", result, producerId);
            return;
        }

        pendingOffsets.offsetsByGroup.keySet().forEach(groupId -> {
            TimelineHashSet<Long> openTransactions = openTransactionsByGroup.get(groupId);
            if (openTransactions != null) {
                openTransactions.remove(producerId);
                if (openTransactions.isEmpty()) {
                    openTransactionsByGroup.remove(groupId);
                }
            }
        });

        if (result == TransactionResult.COMMIT) {
            log.debug("Committed transactional offset commits for producer id {}.", producerId);

            pendingOffsets.offsetsByGroup.forEach((groupId, topicOffsets) -> {
                topicOffsets.forEach((topicName, partitionOffsets) -> {
                    partitionOffsets.forEach((partitionId, offsetAndMetadata) -> {
                        OffsetAndMetadata existingOffsetAndMetadata = offsets.get(
                            groupId,
                            topicName,
                            partitionId
                        );

                        // We always keep the most recent committed offset when we have a mix of transactional and regular
                        // offset commits. Without preserving information of the commit record offset, compaction of the
                        // __consumer_offsets topic itself may result in the wrong offset commit being materialized.
                        if (existingOffsetAndMetadata == null || offsetAndMetadata.recordOffset > existingOffsetAndMetadata.recordOffset) {
                            log.debug("Committed transactional offset commit {} for producer id {} in group {} " +
                                "with topic {} and partition {}.",
                                offsetAndMetadata, producerId, groupId, topicName, partitionId);
                            OffsetAndMetadata previousValue = offsets.put(
                                groupId,
                                topicName,
                                partitionId,
                                offsetAndMetadata
                            );
                            if (previousValue == null) {
                                metrics.incrementNumOffsets();
                            }
                        } else {
                            log.info("Skipped the materialization of transactional offset commit {} for producer id {} in group {} with topic {}, " +
                                "partition {} since its record offset {} is smaller than the record offset {} of the last committed offset.",
                                offsetAndMetadata, producerId, groupId, topicName, partitionId, offsetAndMetadata.recordOffset, existingOffsetAndMetadata.recordOffset);
                        }
                    });
                });
            });
        } else {
            log.debug("Aborted transactional offset commits for producer id {}.", producerId);
        }
    }
Applies the given transaction marker.

@param producerId    The producer id.
@param result        The result of the transaction.
@throws RuntimeException if the transaction can not be completed.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java