public MetadataImage apply(MetadataProvenance provenance) {
        FeaturesImage newFeatures;
        if (featuresDelta == null) {
            newFeatures = image.features();
        } else {
            newFeatures = featuresDelta.apply();
        }
        ClusterImage newCluster;
        if (clusterDelta == null) {
            newCluster = image.cluster();
        } else {
            newCluster = clusterDelta.apply();
        }
        TopicsImage newTopics;
        if (topicsDelta == null) {
            newTopics = image.topics();
        } else {
            newTopics = topicsDelta.apply();
        }
        ConfigurationsImage newConfigs;
        if (configsDelta == null) {
            newConfigs = image.configs();
        } else {
            newConfigs = configsDelta.apply();
        }
        ClientQuotasImage newClientQuotas;
        if (clientQuotasDelta == null) {
            newClientQuotas = image.clientQuotas();
        } else {
            newClientQuotas = clientQuotasDelta.apply();
        }
        ProducerIdsImage newProducerIds;
        if (producerIdsDelta == null) {
            newProducerIds = image.producerIds();
        } else {
            newProducerIds = producerIdsDelta.apply();
        }
        AclsImage newAcls;
        if (aclsDelta == null) {
            newAcls = image.acls();
        } else {
            newAcls = aclsDelta.apply();
        }

        // AutoMQ for Kafka inject start
        S3StreamsMetadataImage newStreamMetadata = getNewS3StreamsMetadataImage();
        S3ObjectsImage newS3ObjectsMetadata = getNewS3ObjectsMetadataImage();
        KVImage newKVImage = getNewKVImage();
        // AutoMQ for Kafka inject end

        ScramImage newScram;
        if (scramDelta == null) {
            newScram = image.scram();
        } else {
            newScram = scramDelta.apply();
        }
        DelegationTokenImage newDelegationTokens;
        if (delegationTokenDelta == null) {
            newDelegationTokens = image.delegationTokens();
        } else {
            newDelegationTokens = delegationTokenDelta.apply();
        }
        return new MetadataImage(
            provenance,
            newFeatures,
            newCluster,
            newTopics,
            newConfigs,
            newClientQuotas,
            newProducerIds,
            newAcls,
            newScram,
            newDelegationTokens,
            newStreamMetadata,
            newS3ObjectsMetadata,
            newKVImage
        );
    }
Create removal deltas for anything which was in the base image, but which was not
referenced in the snapshot records we just applied.
metadata/src/main/java/org/apache/kafka/image/MetadataDelta.java