@SuppressWarnings("checkstyle:javaNCSS")
    private void replay(ApiMessage message, Optional<OffsetAndEpoch> snapshotId, long offset) {
        if (log.isTraceEnabled()) {
            if (snapshotId.isPresent()) {
                log.trace("Replaying snapshot {} record {}",
                    Snapshots.filenameFromSnapshotId(snapshotId.get()),
                        recordRedactor.toLoggableString(message));
            } else {
                log.trace("Replaying log record {} with offset {}",
                        recordRedactor.toLoggableString(message), offset);
            }
        }
        logReplayTracker.replay(message);
        MetadataRecordType type = MetadataRecordType.fromId(message.apiKey());
        // AutoMQ for Kafka inject start
        boolean extensionMatch = extension.replay(type, message, snapshotId, offset);
        // AutoMQ for Kafka inject end
        switch (type) {
            case REGISTER_BROKER_RECORD:
                clusterControl.replay((RegisterBrokerRecord) message, offset);
                break;
            case UNREGISTER_BROKER_RECORD:
                clusterControl.replay((UnregisterBrokerRecord) message);
                break;
            case TOPIC_RECORD:
                replicationControl.replay((TopicRecord) message);
                break;
            case PARTITION_RECORD:
                replicationControl.replay((PartitionRecord) message);
                break;
            case CONFIG_RECORD:
                configurationControl.replay((ConfigRecord) message);
                break;
            case PARTITION_CHANGE_RECORD:
                replicationControl.replay((PartitionChangeRecord) message);
                break;
            case FENCE_BROKER_RECORD:
                clusterControl.replay((FenceBrokerRecord) message);
                break;
            case UNFENCE_BROKER_RECORD:
                clusterControl.replay((UnfenceBrokerRecord) message);
                break;
            case REMOVE_TOPIC_RECORD:
                replicationControl.replay((RemoveTopicRecord) message);
                break;
            case FEATURE_LEVEL_RECORD:
                featureControl.replay((FeatureLevelRecord) message);
                handleFeatureControlChange();
                break;
            case CLIENT_QUOTA_RECORD:
                clientQuotaControlManager.replay((ClientQuotaRecord) message);
                break;
            case PRODUCER_IDS_RECORD:
                producerIdControlManager.replay((ProducerIdsRecord) message);
                break;
            case BROKER_REGISTRATION_CHANGE_RECORD:
                clusterControl.replay((BrokerRegistrationChangeRecord) message);
                break;
            case ACCESS_CONTROL_ENTRY_RECORD:
                aclControlManager.replay((AccessControlEntryRecord) message);
                break;
            case REMOVE_ACCESS_CONTROL_ENTRY_RECORD:
                aclControlManager.replay((RemoveAccessControlEntryRecord) message);
                break;
            case USER_SCRAM_CREDENTIAL_RECORD:
                scramControlManager.replay((UserScramCredentialRecord) message);
                break;
            case REMOVE_USER_SCRAM_CREDENTIAL_RECORD:
                scramControlManager.replay((RemoveUserScramCredentialRecord) message);
                break;
            case DELEGATION_TOKEN_RECORD:
                delegationTokenControlManager.replay((DelegationTokenRecord) message);
                break;
            case REMOVE_DELEGATION_TOKEN_RECORD:
                delegationTokenControlManager.replay((RemoveDelegationTokenRecord) message);
                break;
            case NO_OP_RECORD:
                // NoOpRecord is an empty record and doesn't need to be replayed
                break;
            case ZK_MIGRATION_STATE_RECORD:
                featureControl.replay((ZkMigrationStateRecord) message);
                break;
            case BEGIN_TRANSACTION_RECORD:
                offsetControl.replay((BeginTransactionRecord) message, offset);
                break;
            case END_TRANSACTION_RECORD:
                offsetControl.replay((EndTransactionRecord) message, offset);
                break;
            case ABORT_TRANSACTION_RECORD:
                offsetControl.replay((AbortTransactionRecord) message, offset);
                break;
            case REGISTER_CONTROLLER_RECORD:
                clusterControl.replay((RegisterControllerRecord) message);
                break;

            // AutoMQ for Kafka inject start
            case S3_STREAM_RECORD:
                streamControlManager.replay((S3StreamRecord) message);
                break;
            case REMOVE_S3_STREAM_RECORD:
                streamControlManager.replay((RemoveS3StreamRecord) message);
                break;
            case RANGE_RECORD:
                streamControlManager.replay((RangeRecord) message);
                break;
            case REMOVE_RANGE_RECORD:
                streamControlManager.replay((RemoveRangeRecord) message);
                break;
            case S3_STREAM_OBJECT_RECORD:
                streamControlManager.replay((S3StreamObjectRecord) message);
                break;
            case REMOVE_S3_STREAM_OBJECT_RECORD:
                streamControlManager.replay((RemoveS3StreamObjectRecord) message);
                break;
            case S3_STREAM_SET_OBJECT_RECORD:
                streamControlManager.replay((S3StreamSetObjectRecord) message);
                break;
            case REMOVE_STREAM_SET_OBJECT_RECORD:
                streamControlManager.replay((RemoveStreamSetObjectRecord) message);
                break;
            case S3_OBJECT_RECORD:
                s3ObjectControlManager.replay((S3ObjectRecord) message);
                break;
            case REMOVE_S3_OBJECT_RECORD:
                s3ObjectControlManager.replay((RemoveS3ObjectRecord) message);
                break;
            case ASSIGNED_STREAM_ID_RECORD:
                streamControlManager.replay((AssignedStreamIdRecord) message);
                break;
            case ASSIGNED_S3_OBJECT_ID_RECORD:
                s3ObjectControlManager.replay((AssignedS3ObjectIdRecord) message);
                break;
            case NODE_WALMETADATA_RECORD:
                streamControlManager.replay((NodeWALMetadataRecord) message);
                break;
            case REMOVE_NODE_WALMETADATA_RECORD:
                streamControlManager.replay((RemoveNodeWALMetadataRecord) message);
                break;
            case KVRECORD: {
                KVRecord record = (KVRecord) message;
                kvControlManager.replay(record);
                topicDeletionManager.replay(record);
                nodeControlManager.replay(record);
                break;
            }
            case REMOVE_KVRECORD: {
                RemoveKVRecord record = (RemoveKVRecord) message;
                kvControlManager.replay(record);
                topicDeletionManager.replay(record);
                nodeControlManager.replay(record);
                break;
            }
            case UPDATE_NEXT_NODE_ID_RECORD:
                clusterControl.replay((UpdateNextNodeIdRecord) message);
                break;
            case S3_STREAM_END_OFFSETS_RECORD:
                streamControlManager.replay((S3StreamEndOffsetsRecord) message);
                break;
            default:
                if (!extensionMatch) {
                    throw new RuntimeException("Unhandled record type " + type);
                }
            // AutoMQ for Kafka inject end
        }
    }
Apply the metadata record to its corresponding in-memory state(s)

@param message           The metadata record
@param snapshotId        The snapshotId if this record is from a snapshot
@param offset            The offset of the record
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java