public void replay(
        long recordOffset,
        long producerId,
        OffsetCommitKey key,
        OffsetCommitValue value
    ) {
        final String groupId = key.group();
        final String topic = key.topic();
        final int partition = key.partition();

        if (value != null) {
            // The classic or consumer group should exist when offsets are committed or
            // replayed. However, it won't if the consumer commits offsets but does not
            // use the membership functionality. In this case, we automatically create
            // a so-called "simple consumer group". This is an empty classic group
            // without a protocol type.
            try {
                groupMetadataManager.group(groupId);
            } catch (GroupIdNotFoundException ex) {
                groupMetadataManager.getOrMaybeCreateClassicGroup(groupId, true);
            }

            if (producerId == RecordBatch.NO_PRODUCER_ID) {
                log.debug("Replaying offset commit with key {}, value {}", key, value);
                // If the offset is not part of a transaction, it is directly stored
                // in the offsets store.
                OffsetAndMetadata previousValue = offsets.put(
                    groupId,
                    topic,
                    partition,
                    OffsetAndMetadata.fromRecord(recordOffset, value)
                );
                if (previousValue == null) {
                    metrics.incrementNumOffsets();
                }
            } else {
                log.debug("Replaying transactional offset commit with producer id {}, key {}, value {}", producerId, key, value);
                // Otherwise, the transaction offset is stored in the pending transactional
                // offsets store. Pending offsets there are moved to the main store when
                // the transaction is committed; or removed when the transaction is aborted.
                pendingTransactionalOffsets
                    .computeIfAbsent(producerId, __ -> new Offsets())
                    .put(
                        groupId,
                        topic,
                        partition,
                        OffsetAndMetadata.fromRecord(recordOffset, value)
                    );
                openTransactionsByGroup
                    .computeIfAbsent(groupId, __ -> new TimelineHashSet<>(snapshotRegistry, 1))
                    .add(producerId);
            }
        } else {
            if (offsets.remove(groupId, topic, partition) != null) {
                metrics.decrementNumOffsets();
            }

            // Remove all the pending offset commits related to the tombstone.
            TimelineHashSet<Long> openTransactions = openTransactionsByGroup.get(groupId);
            if (openTransactions != null) {
                openTransactions.forEach(openProducerId -> {
                    Offsets pendingOffsets = pendingTransactionalOffsets.get(openProducerId);
                    if (pendingOffsets != null) {
                        pendingOffsets.remove(groupId, topic, partition);
                    }
                });
            }
        }
    }
@Test
    public void testReplay() {
        OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();

        verifyReplay(context, "foo", "bar", 0, new OffsetAndMetadata(
            0L,
            100L,
            OptionalInt.empty(),
            "small",
            context.time.milliseconds(),
            OptionalLong.empty()
        ));

        verifyReplay(context, "foo", "bar", 0, new OffsetAndMetadata(
            1L,
            200L,
            OptionalInt.of(10),
            "small",
            context.time.milliseconds(),
            OptionalLong.empty()
        ));

        verifyReplay(context, "foo", "bar", 1, new OffsetAndMetadata(
            2L,
            200L,
            OptionalInt.of(10),
            "small",
            context.time.milliseconds(),
            OptionalLong.empty()
        ));

        verifyReplay(context, "foo", "bar", 1, new OffsetAndMetadata(
            3L,
            300L,
            OptionalInt.of(10),
            "small",
            context.time.milliseconds(),
            OptionalLong.of(12345L)
        ));
    }