@SuppressWarnings("checkstyle:javaNCSS")
    public void replay(ApiMessage record) {
        MetadataRecordType type = MetadataRecordType.fromId(record.apiKey());
        switch (type) {
            case REGISTER_BROKER_RECORD:
                replay((RegisterBrokerRecord) record);
                break;
            case UNREGISTER_BROKER_RECORD:
                replay((UnregisterBrokerRecord) record);
                break;
            case TOPIC_RECORD:
                replay((TopicRecord) record);
                break;
            case PARTITION_RECORD:
                replay((PartitionRecord) record);
                break;
            case CONFIG_RECORD:
                replay((ConfigRecord) record);
                break;
            case PARTITION_CHANGE_RECORD:
                replay((PartitionChangeRecord) record);
                break;
            case FENCE_BROKER_RECORD:
                replay((FenceBrokerRecord) record);
                break;
            case UNFENCE_BROKER_RECORD:
                replay((UnfenceBrokerRecord) record);
                break;
            case REMOVE_TOPIC_RECORD:
                replay((RemoveTopicRecord) record);
                break;
            case DELEGATION_TOKEN_RECORD:
                replay((DelegationTokenRecord) record);
                break;
            case USER_SCRAM_CREDENTIAL_RECORD:
                replay((UserScramCredentialRecord) record);
                break;
            case FEATURE_LEVEL_RECORD:
                replay((FeatureLevelRecord) record);
                break;
            case CLIENT_QUOTA_RECORD:
                replay((ClientQuotaRecord) record);
                break;
            case PRODUCER_IDS_RECORD:
                replay((ProducerIdsRecord) record);
                break;
            case BROKER_REGISTRATION_CHANGE_RECORD:
                replay((BrokerRegistrationChangeRecord) record);
                break;
            case ACCESS_CONTROL_ENTRY_RECORD:
                replay((AccessControlEntryRecord) record);
                break;
            case REMOVE_ACCESS_CONTROL_ENTRY_RECORD:
                replay((RemoveAccessControlEntryRecord) record);
                break;
            case REMOVE_USER_SCRAM_CREDENTIAL_RECORD:
                replay((RemoveUserScramCredentialRecord) record);
                break;
            case REMOVE_DELEGATION_TOKEN_RECORD:
                replay((RemoveDelegationTokenRecord) record);
                break;
            case NO_OP_RECORD:
                /* NoOpRecord is an empty record and doesn't need to be replayed beyond
                 * updating the highest offset and epoch.
                 */
                break;
            case ZK_MIGRATION_STATE_RECORD:
                replay((ZkMigrationStateRecord) record);
                break;
            case REGISTER_CONTROLLER_RECORD:
                replay((RegisterControllerRecord) record);
                break;
            // AutoMQ for Kafka inject start
            case UPDATE_NEXT_NODE_ID_RECORD:
                replay((UpdateNextNodeIdRecord) record);
                break;
            case S3_STREAM_RECORD:
                replay((S3StreamRecord) record);
                break;
            case REMOVE_S3_STREAM_RECORD:
                replay((RemoveS3StreamRecord) record);
                break;
            case RANGE_RECORD:
                replay((RangeRecord) record);
                break;
            case REMOVE_RANGE_RECORD:
                replay((RemoveRangeRecord) record);
                break;
            case S3_STREAM_OBJECT_RECORD:
                replay((S3StreamObjectRecord) record);
                break;
            case REMOVE_S3_STREAM_OBJECT_RECORD:
                replay((RemoveS3StreamObjectRecord) record);
                break;
            case S3_STREAM_SET_OBJECT_RECORD:
                replay((S3StreamSetObjectRecord) record);
                break;
            case REMOVE_STREAM_SET_OBJECT_RECORD:
                replay((RemoveStreamSetObjectRecord) record);
                break;
            case S3_OBJECT_RECORD:
                replay((S3ObjectRecord) record);
                break;
            case REMOVE_S3_OBJECT_RECORD:
                replay((RemoveS3ObjectRecord) record);
                break;
            case ASSIGNED_S3_OBJECT_ID_RECORD:
                replay((AssignedS3ObjectIdRecord) record);
                break;
            case ASSIGNED_STREAM_ID_RECORD:
                replay((AssignedStreamIdRecord) record);
                break;
            case NODE_WALMETADATA_RECORD:
                replay((NodeWALMetadataRecord) record);
                break;
            case REMOVE_NODE_WALMETADATA_RECORD:
                replay((RemoveNodeWALMetadataRecord) record);
                break;
            case KVRECORD:
                replay((KVRecord) record);
                break;
            case REMOVE_KVRECORD:
                replay((RemoveKVRecord) record);
                break;
            case S3_STREAM_END_OFFSETS_RECORD:
                replay((S3StreamEndOffsetsRecord) record);
                break;
                // AutoMQ for Kafka inject end
            default:
                throw new RuntimeException("Unknown metadata record type " + type);
        }
    }
A change to the broker metadata image.
metadata/src/main/java/org/apache/kafka/image/MetadataDelta.java