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);
        }
    }
@Test
    public void testOffsetCommitsNumberMetricWithTransactionalOffsets() {
        OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();

        // Add pending transactional commit for producer id 4.
        verifyTransactionalReplay(context, 4L, "foo", "bar", 0, new OffsetAndMetadata(
            0L,
            100L,
            OptionalInt.empty(),
            "small",
            context.time.milliseconds(),
            OptionalLong.empty()
        ));

        // Add pending transactional commit for producer id 5.
        verifyTransactionalReplay(context, 5L, "foo", "bar", 0, new OffsetAndMetadata(
            1L,
            101L,
            OptionalInt.empty(),
            "small",
            context.time.milliseconds(),
            OptionalLong.empty()
        ));

        // Add pending transactional commit for producer id 6.
        verifyTransactionalReplay(context, 6L, "foo", "bar", 1, new OffsetAndMetadata(
            2L,
            200L,
            OptionalInt.empty(),
            "small",
            context.time.milliseconds(),
            OptionalLong.empty()
        ));

        // Commit all the transactions.
        context.replayEndTransactionMarker(4L, TransactionResult.COMMIT);
        context.replayEndTransactionMarker(5L, TransactionResult.COMMIT);
        context.replayEndTransactionMarker(6L, TransactionResult.COMMIT);

        // Verify the sensor is called twice as we have only
        // two partitions.
        verify(context.metrics, times(2)).incrementNumOffsets();
    }