@Override
    public void replay(
        long offset,
        long producerId,
        short producerEpoch,
        CoordinatorRecord record
    ) throws RuntimeException {
        ApiMessageAndVersion key = record.key();
        ApiMessageAndVersion value = record.value();

        switch (key.version()) {
            case 0:
            case 1:
                offsetMetadataManager.replay(
                    offset,
                    producerId,
                    (OffsetCommitKey) key.message(),
                    (OffsetCommitValue) Utils.messageOrNull(value)
                );
                break;

            case 2:
                groupMetadataManager.replay(
                    (GroupMetadataKey) key.message(),
                    (GroupMetadataValue) Utils.messageOrNull(value)
                );
                break;

            case 3:
                groupMetadataManager.replay(
                    (ConsumerGroupMetadataKey) key.message(),
                    (ConsumerGroupMetadataValue) Utils.messageOrNull(value)
                );
                break;

            case 4:
                groupMetadataManager.replay(
                    (ConsumerGroupPartitionMetadataKey) key.message(),
                    (ConsumerGroupPartitionMetadataValue) Utils.messageOrNull(value)
                );
                break;

            case 5:
                groupMetadataManager.replay(
                    (ConsumerGroupMemberMetadataKey) key.message(),
                    (ConsumerGroupMemberMetadataValue) Utils.messageOrNull(value)
                );
                break;

            case 6:
                groupMetadataManager.replay(
                    (ConsumerGroupTargetAssignmentMetadataKey) key.message(),
                    (ConsumerGroupTargetAssignmentMetadataValue) Utils.messageOrNull(value)
                );
                break;

            case 7:
                groupMetadataManager.replay(
                    (ConsumerGroupTargetAssignmentMemberKey) key.message(),
                    (ConsumerGroupTargetAssignmentMemberValue) Utils.messageOrNull(value)
                );
                break;

            case 8:
                groupMetadataManager.replay(
                    (ConsumerGroupCurrentMemberAssignmentKey) key.message(),
                    (ConsumerGroupCurrentMemberAssignmentValue) Utils.messageOrNull(value)
                );
                break;

            case 9:
                groupMetadataManager.replay(
                    (ShareGroupPartitionMetadataKey) key.message(),
                    (ShareGroupPartitionMetadataValue) Utils.messageOrNull(value)
                );
                break;

            case 10:
                groupMetadataManager.replay(
                    (ShareGroupMemberMetadataKey) key.message(),
                    (ShareGroupMemberMetadataValue) Utils.messageOrNull(value)
                );
                break;

            case 11:
                groupMetadataManager.replay(
                    (ShareGroupMetadataKey) key.message(),
                    (ShareGroupMetadataValue) Utils.messageOrNull(value)
                );
                break;

            case 12:
                groupMetadataManager.replay(
                    (ShareGroupTargetAssignmentMetadataKey) key.message(),
                    (ShareGroupTargetAssignmentMetadataValue) Utils.messageOrNull(value)
                );
                break;

            case 13:
                groupMetadataManager.replay(
                    (ShareGroupTargetAssignmentMemberKey) key.message(),
                    (ShareGroupTargetAssignmentMemberValue) Utils.messageOrNull(value)
                );
                break;

            case 14:
                groupMetadataManager.replay(
                    (ShareGroupCurrentMemberAssignmentKey) key.message(),
                    (ShareGroupCurrentMemberAssignmentValue) Utils.messageOrNull(value)
                );
                break;

            default:
                throw new IllegalStateException("Received an unknown record type " + key.version()
                    + " in " + record);
        }
    }
@Test
    public void testReplayConsumerGroupMetadataWithNullValue() {
        GroupMetadataManager groupMetadataManager = mock(GroupMetadataManager.class);
        OffsetMetadataManager offsetMetadataManager = mock(OffsetMetadataManager.class);
        CoordinatorMetrics coordinatorMetrics = mock(CoordinatorMetrics.class);
        CoordinatorMetricsShard metricsShard = mock(CoordinatorMetricsShard.class);
        GroupCoordinatorShard coordinator = new GroupCoordinatorShard(
            new LogContext(),
            groupMetadataManager,
            offsetMetadataManager,
            Time.SYSTEM,
            new MockCoordinatorTimer<>(Time.SYSTEM),
            mock(GroupCoordinatorConfig.class),
            coordinatorMetrics,
            metricsShard
        );

        ConsumerGroupMetadataKey key = new ConsumerGroupMetadataKey();

        coordinator.replay(0L, RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, new CoordinatorRecord(
            new ApiMessageAndVersion(key, (short) 3),
            null
        ));

        verify(groupMetadataManager, times(1)).replay(key, null);
    }