package io.apicurio.registry.storage.impl.kafkasql;

import io.apicurio.common.apps.config.DynamicConfigPropertyDto;
import io.apicurio.registry.content.ContentHandle;
import io.apicurio.registry.events.ArtifactCreated;
import io.apicurio.registry.events.ArtifactDeleted;
import io.apicurio.registry.events.ArtifactMetadataUpdated;
import io.apicurio.registry.events.ArtifactRuleConfigured;
import io.apicurio.registry.events.ArtifactVersionCreated;
import io.apicurio.registry.events.ArtifactVersionDeleted;
import io.apicurio.registry.events.ArtifactVersionMetadataUpdated;
import io.apicurio.registry.events.GlobalRuleConfigured;
import io.apicurio.registry.events.GroupCreated;
import io.apicurio.registry.events.GroupDeleted;
import io.apicurio.registry.events.GroupMetadataUpdated;
import io.apicurio.registry.events.GroupRuleConfigured;
import io.apicurio.registry.logging.Logged;
import io.apicurio.registry.metrics.StorageMetricsApply;
import io.apicurio.registry.metrics.health.liveness.PersistenceExceptionLivenessApply;
import io.apicurio.registry.metrics.health.readiness.PersistenceTimeoutReadinessApply;
import io.apicurio.registry.model.BranchId;
import io.apicurio.registry.model.GA;
import io.apicurio.registry.model.VersionId;
import io.apicurio.registry.rules.compatibility.CompatibilityLevel;
import io.apicurio.registry.rules.integrity.IntegrityLevel;
import io.apicurio.registry.rules.validity.ValidityLevel;
import io.apicurio.registry.storage.RegistryStorage;
import io.apicurio.registry.storage.StorageEvent;
import io.apicurio.registry.storage.StorageEventType;
import io.apicurio.registry.storage.decorator.ReadOnlyDelegatingStorage;
import io.apicurio.registry.storage.dto.*;
import io.apicurio.registry.storage.error.ArtifactNotFoundException;
import io.apicurio.registry.storage.error.GroupAlreadyExistsException;
import io.apicurio.registry.storage.error.GroupNotFoundException;
import io.apicurio.registry.storage.error.RegistryStorageException;
import io.apicurio.registry.storage.error.RuleAlreadyExistsException;
import io.apicurio.registry.storage.error.RuleNotFoundException;
import io.apicurio.registry.storage.error.VersionNotFoundException;
import io.apicurio.registry.storage.impl.kafkasql.KafkaSqlFactory.KafkaAdminClient;
import io.apicurio.registry.storage.impl.kafkasql.messages.*;
import io.apicurio.registry.storage.impl.kafkasql.sql.KafkaSqlSink;
import io.apicurio.registry.storage.impl.sql.RegistryStorageContentUtils;
import io.apicurio.registry.storage.impl.sql.SqlRegistryStorage;
import io.apicurio.registry.storage.impl.util.KafkaAdminUtil;
import io.apicurio.registry.storage.impl.util.ProducerActions;
import io.apicurio.registry.storage.importing.DataImporter;
import io.apicurio.registry.storage.importing.v2.SqlDataUpgrader;
import io.apicurio.registry.storage.importing.v3.SqlDataImporter;
import io.apicurio.registry.types.RuleType;
import io.apicurio.registry.types.VersionState;
import io.apicurio.registry.utils.impexp.EntityInputStream;
import io.apicurio.registry.utils.impexp.v3.ArtifactEntity;
import io.apicurio.registry.utils.impexp.v3.ArtifactRuleEntity;
import io.apicurio.registry.utils.impexp.v3.ArtifactVersionEntity;
import io.apicurio.registry.utils.impexp.v3.BranchEntity;
import io.apicurio.registry.utils.impexp.v3.CommentEntity;
import io.apicurio.registry.utils.impexp.v3.ContentEntity;
import io.apicurio.registry.utils.impexp.v3.GlobalRuleEntity;
import io.apicurio.registry.utils.impexp.v3.GroupEntity;
import io.apicurio.registry.utils.impexp.v3.GroupRuleEntity;
import io.quarkus.arc.lookup.LookupIfProperty;
import jakarta.annotation.PreDestroy;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.event.Event;
import jakarta.inject.Inject;
import jakarta.inject.Named;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;

import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;

import static io.apicurio.registry.storage.impl.kafkasql.KafkaSqlSubmitter.BOOTSTRAP_MESSAGE_TYPE;
import static io.apicurio.registry.utils.ConcurrentUtil.blockOnResult;

/**
 * An implementation of a registry artifactStore that extends the basic SQL artifactStore but federates
 * 'write' operations to other nodes in a cluster using a Kafka topic. As a result, all reads are performed
 * locally but all writes are published to a topic for consumption by all nodes.
 */
@ApplicationScoped
@PersistenceExceptionLivenessApply
@PersistenceTimeoutReadinessApply
@StorageMetricsApply
@Logged
@LookupIfProperty(name = "apicurio.storage.kind", stringValue = "kafkasql")
public class KafkaSqlRegistryStorage extends ReadOnlyDelegatingStorage implements RegistryStorage {

    @Inject
    Logger log;

    @Inject
    KafkaSqlConfiguration configuration;

    @Inject
    KafkaSqlCoordinator coordinator;

    @Inject
    KafkaAdminUtil kafkaAdmin;

    @Inject
    KafkaSqlSink kafkaSqlSink;

    @Inject
    SqlRegistryStorage sqlStore;

    @Inject
    RegistryStorageContentUtils utils;

    @Inject
    @Named("KafkaSqlJournalConsumer")
    KafkaConsumer<KafkaSqlMessageKey, KafkaSqlMessage> journalConsumer;

    @Inject
    @Named("KafkaSqlSnapshotsConsumer")
    KafkaConsumer<String, String> snapshotsConsumer;

    @Inject
    @Named("KafkaSqlSnapshotsProducer")
    ProducerActions<String, String> snapshotsProducer;

    @Inject
    KafkaAdminClient adminClient;

    @Inject
    KafkaSqlSubmitter submitter;

    @Inject
    Event<StorageEvent> storageEvent;

    @Inject
    Event<KafkaSqlOutboxEvent> outboxEvent;

    private volatile boolean bootstrapped = false;
    private volatile boolean stopped = true;
    private volatile boolean snapshotProcessed = false;

    // The snapshot id used to determine if this replica must process a snapshot message
    private volatile String lastTriggeredSnapshot = null;

    // Reference to the consumer thread for health checks
    private volatile Thread consumerThread = null;

    @Override
    public String storageName() {
        return "kafkasql";
    }

    @Override
    public void initialize() {
        log.info("Using Kafka-SQL artifactStore.");

        // First, if needed, create the Kafka topics.
        if (configuration.getTopicAutoCreate()) {
            // This is sequential instead of concurrent, but I think for 3 topics it's worth the simplicity.
            kafkaAdmin.createTopicIfDoesNotExist(configuration.getTopic(), configuration.getTopicProperties());
            kafkaAdmin.createTopicIfDoesNotExist(configuration.getSnapshotsTopic(), configuration.getSnapshotTopicProperties());
            kafkaAdmin.createTopicIfDoesNotExist(configuration.getEventsTopic(), configuration.getEventsTopicProperties());
        }

        kafkaAdmin.verifyTopicConfiguration(configuration.getTopic());
        kafkaAdmin.verifyTopicConfiguration(configuration.getSnapshotsTopic());
        kafkaAdmin.verifyTopicConfiguration(configuration.getEventsTopic());
        adminClient.close();
        kafkaAdmin.verifyJournalTopicContents();

        // Try to restore the internal database from a snapshot
        final long bootstrapStart = System.currentTimeMillis();
        String snapshotId = consumeSnapshotsTopic(snapshotsConsumer);

        // Once the topics are created, and the snapshots processed, initialize the internal SQL Storage.
        sqlStore.initialize();
        setDelegate(sqlStore);

        // Once the SQL storage has been initialized, start the Kafka consumer thread.
        log.info("SQL store initialized, starting consumer thread.");
        startConsumerThread(journalConsumer, snapshotId, bootstrapStart);
    }

    @Override
    public boolean isReady() {
        return bootstrapped;
    }

    @Override
    public boolean isAlive() {
        // Check that we've bootstrapped, haven't been stopped, and the consumer thread is still running.
        // If the consumer thread dies (e.g., due to Kafka connectivity issues), the storage is no longer alive.
        return bootstrapped && !stopped && consumerThread != null && consumerThread.isAlive();
    }

    @PreDestroy
    void onDestroy() {
        stopped = true;
        journalConsumer.close();
        snapshotsConsumer.close();
    }

    /**
     * Consume the snapshots topic, looking for the most recent snapshot in the topic. Once found, it
     * restores the internal database using the snapshot's content. Polls in a loop until all messages
     * are consumed from the topic.
     */
    private String consumeSnapshotsTopic(KafkaConsumer<String, String> snapshotsConsumer) {
        // Subscribe to the snapshots topic
        Collection<String> topics = Collections.singleton(configuration.getSnapshotsTopic());
        snapshotsConsumer.subscribe(topics);

        // Wait for partition assignment
        var assigned = snapshotsConsumer.assignment();
        while (assigned.isEmpty()) {
            snapshotsConsumer.poll(configuration.getPollTimeout());
            assigned = snapshotsConsumer.assignment();
        }

        // Record the current end offsets so we stop there, even if other pods produce new snapshots
        var endOffsets = snapshotsConsumer.endOffsets(assigned);

        List<ConsumerRecord<String, String>> snapshots = new ArrayList<>();
        String snapshotRecordKey = null;

        // Poll until we reach the original end offsets or get an empty result
        ConsumerRecords<String, String> records;
        do {
            records = snapshotsConsumer.poll(configuration.getPollTimeout());
            if (records != null && !records.isEmpty()) {
                records.forEach(snapshots::add);
                log.debug("Polled {} snapshot records, total collected: {}", records.count(), snapshots.size());
            }
        } while (records != null && !records.isEmpty()
                && endOffsets.entrySet().stream()
                        .anyMatch(e -> snapshotsConsumer.position(e.getKey()) < e.getValue()));

        if (!snapshots.isEmpty()) {
            log.info("Found {} total snapshots in the snapshots topic.", snapshots.size());

            // sort snapshots by timestamp
            snapshots.sort(Comparator.comparingLong(ConsumerRecord::timestamp));

            Path mostRecentSnapshotPath = null;
            for (ConsumerRecord<String, String> snapshotFound : snapshots) {
                // Restore database from snapshot
                try {
                    String path = snapshotFound.value();
                    if (null != path && !path.isBlank() && Files.exists(Path.of(snapshotFound.value()))) {
                        log.debug("Snapshot with path {} found.", snapshotFound.value());
                        snapshotRecordKey = snapshotFound.key();
                        mostRecentSnapshotPath = Path.of(snapshotFound.value());
                    }
                } catch (IllegalArgumentException ex) {
                    log.warn(
                            "Snapshot with path {} ignored, the snapshot is likely invalid or cannot be found",
                            snapshotFound.value());
                }
            }

            // Here we have the most recent snapshot that we can find, try to restore the internal database
            // from it.
            if (null != mostRecentSnapshotPath) {
                log.info("Restoring snapshot {} to the internal database...", mostRecentSnapshotPath);
                sqlStore.restoreFromSnapshot(mostRecentSnapshotPath.toString());
            }
        }

        return snapshotRecordKey;
    }

    /**
     * Start the KSQL Kafka consumer thread which is responsible for subscribing to the kafka topic, consuming
     * JournalRecord entries found on that topic, and applying those journal entries to the internal data
     * model.
     */
    private void startConsumerThread(final KafkaConsumer<KafkaSqlMessageKey, KafkaSqlMessage> consumer,
                                     String snapshotId, long bootstrapStart) {
        log.info("Starting KSQL consumer thread on topic: {}", configuration.getTopic());
        log.info("Bootstrap servers: {}", configuration.getBootstrapServers());

        final String bootstrapId = UUID.randomUUID().toString();
        submitter.submitBootstrap(bootstrapId);

        Runnable runner = () -> {
            try (consumer) {
                log.info("Subscribing to {}", configuration.getTopic());
                // Subscribe to the journal topic
                Collection<String> topics = Collections.singleton(configuration.getTopic());
                consumer.subscribe(topics);

                // Main consumer loop
                while (!stopped) {
                    final ConsumerRecords<KafkaSqlMessageKey, KafkaSqlMessage> records = consumer.poll(configuration.getPollTimeout());
                    if (records != null && !records.isEmpty()) {
                        log.debug("Consuming {} journal records.", records.count());

                        if (null != snapshotId && !snapshotProcessed) {
                            // If there is a snapshot key present, we process (and discard) all the messages
                            // until we find the snapshot marker that corresponds to the snapshot key.
                            Iterator<ConsumerRecord<KafkaSqlMessageKey, KafkaSqlMessage>> it = records
                                    .iterator();
                            while (it.hasNext() && !snapshotProcessed) {
                                ConsumerRecord<KafkaSqlMessageKey, KafkaSqlMessage> record = it.next();
                                if (processSnapshot(snapshotId, record)) {
                                    log.debug(
                                            "Snapshot marker found {} the new messages will be applied on top of the snapshot data.",
                                            record.key());
                                    snapshotProcessed = true;
                                    break;
                                } else {
                                    log.debug(
                                            "Discarding message with key {} as it was sent before a newer snapshot was created.",
                                            record.key());
                                }
                            }

                            // If the snapshot marker has not been found, continue with message skipping until
                            // we find it.
                            if (snapshotProcessed) {
                                // Once the snapshot marker message has been found, we can process the rest of
                                // the messages as usual, applying the new changes on top of the existing ones
                                // in the snapshot.
                                while (it.hasNext()) {
                                    ConsumerRecord<KafkaSqlMessageKey, KafkaSqlMessage> record = it.next();
                                    processRecord(record, bootstrapId, bootstrapStart);
                                }
                            }
                        } else {
                            // If there is no snapshot, simply process the existing messages in the kafka
                            // topic as usual.
                            records.forEach(record -> processRecord(record, bootstrapId, bootstrapStart));
                        }
                    }
                }
            }
        };
        stopped = false;
        Thread thread = new Thread(runner);
        thread.setDaemon(true);
        thread.setName("KSQL Kafka Consumer Thread");
        consumerThread = thread;
        thread.start();
    }

    private boolean processSnapshot(String snapshotId,
            ConsumerRecord<KafkaSqlMessageKey, KafkaSqlMessage> record) {
        return record.value() instanceof CreateSnapshot1Message
                && snapshotId.equals(((CreateSnapshot1Message) record.value()).getSnapshotId());
    }

    private void processRecord(ConsumerRecord<KafkaSqlMessageKey, KafkaSqlMessage> record, String bootstrapId,
            long bootstrapStart) {
        // If the key is null, we couldn't deserialize the message
        if (record.key() == null) {
            log.warn("Discarded an unreadable/unrecognized Kafka message.");
            return;
        }

        // If the key is a Bootstrap key, then we have processed all messages and can set bootstrapped to
        // 'true'
        if (BOOTSTRAP_MESSAGE_TYPE.equals(record.key().getMessageType())) {
            KafkaSqlMessageKey bkey = record.key();
            if (bkey.getUuid().equals(bootstrapId)) {
                this.bootstrapped = true;
                storageEvent.fireAsync(StorageEvent.builder().type(StorageEventType.READY).build());
                log.info("KafkaSQL storage bootstrapped in {} ms.",
                        System.currentTimeMillis() - bootstrapStart);
            }
            return;
        }

        // If the key is a CreateSnapshotMessage key, but this replica does not have the snapshotId, it means
        // that it wasn't triggered here, so just skip the message.
        if (record.value() instanceof CreateSnapshot1Message
                && !((CreateSnapshot1Message) record.value()).getSnapshotId().equals(lastTriggeredSnapshot)) {
            log.debug(
                    "Snapshot trigger message with id {} being skipped since this replica did not trigger the creation.",
                    ((CreateSnapshot1Message) record.value()).getSnapshotId());
            return;
        }

        // If the value is null, then this is a tombstone (or unrecognized) message and should not
        // be processed.
        if (record.value() == null) {
            log.info("Discarded a (presumed) tombstone message with key: {}", record.key());
            return;
        }

        // Note: We process journal records directly on the consumer thread. Since all messages in KafkaSQL
        // are in a single partition, ordering is guaranteed and processing must be sequential. Introducing
        // a separate processing thread with a queue would add complexity without significant benefit, as
        // we'd still need to ensure sequential processing for correctness. The coordinator mechanism already
        // handles response synchronization for write operations.
        kafkaSqlSink.processMessage(record);
    }

    /**
     * @see io.apicurio.common.apps.config.DynamicConfigStorage#setConfigProperty(io.apicurio.common.apps.config.DynamicConfigPropertyDto)
     */
    @Override
    public void setConfigProperty(DynamicConfigPropertyDto propertyDto) {
        var message = new SetConfigProperty1Message(propertyDto);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.common.apps.config.DynamicConfigStorage#deleteConfigProperty(java.lang.String)
     */
    @Override
    public void deleteConfigProperty(String propertyName) {
        var message = new DeleteConfigProperty1Message(propertyName);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    @Override
    public Pair<ArtifactMetaDataDto, ArtifactVersionMetaDataDto> createArtifact(String groupId,
            String artifactId, String artifactType, EditableArtifactMetaDataDto artifactMetaData,
            String version, ContentWrapperDto versionContent, EditableVersionMetaDataDto versionMetaData,
            List<String> versionBranches, boolean versionIsDraft, boolean dryRun, String owner)
            throws RegistryStorageException {
        String content = versionContent != null ? versionContent.getContent().content() : null;
        String contentType = versionContent != null ? versionContent.getContentType() : null;
        List<ArtifactReferenceDto> references = versionContent != null ? versionContent.getReferences()
                : null;
        var message = new CreateArtifact11Message(groupId, artifactId, artifactType, artifactMetaData,
                version, contentType, content, references, versionMetaData, versionBranches, versionIsDraft,
                dryRun, owner);
        var uuid = blockOnResult(submitter.submitMessage(message));

        Pair<ArtifactMetaDataDto, ArtifactVersionMetaDataDto> createdArtifact = (Pair<ArtifactMetaDataDto, ArtifactVersionMetaDataDto>) coordinator
                .waitForResponse(uuid);

        outboxEvent.fire(KafkaSqlOutboxEvent.of(ArtifactCreated.of(createdArtifact.getLeft())));

        if (createdArtifact.getRight() != null) {
            outboxEvent.fire(KafkaSqlOutboxEvent.of(ArtifactVersionCreated.of(createdArtifact.getRight())));
        }

        return createdArtifact;
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#deleteArtifact(java.lang.String, java.lang.String)
     */
    @SuppressWarnings("unchecked")
    @Override
    public List<String> deleteArtifact(String groupId, String artifactId)
            throws ArtifactNotFoundException, RegistryStorageException {
        var message = new DeleteArtifact2Message(groupId, artifactId);
        var uuid = blockOnResult(submitter.submitMessage(message));
        List<String> versions = (List<String>) coordinator.waitForResponse(uuid);
        outboxEvent.fire(KafkaSqlOutboxEvent.of(ArtifactDeleted.of(groupId, artifactId)));
        return versions;
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#deleteArtifacts(java.lang.String)
     */
    @Override
    public void deleteArtifacts(String groupId) throws RegistryStorageException {
        var message = new DeleteArtifacts1Message(groupId);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    @Override
    public ArtifactVersionMetaDataDto createArtifactVersion(String groupId, String artifactId, String version,
            String artifactType, ContentWrapperDto contentDto, EditableVersionMetaDataDto metaData,
            List<String> branches, boolean isDraft, boolean dryRun, String owner)
            throws RegistryStorageException {
        String content = contentDto != null ? contentDto.getContent().content() : null;
        String contentType = contentDto != null ? contentDto.getContentType() : null;
        List<ArtifactReferenceDto> references = contentDto != null ? contentDto.getReferences() : null;
        var message = new CreateArtifactVersion10Message(groupId, artifactId, version, artifactType,
                contentType, content, references, metaData, branches, isDraft, dryRun, owner);
        var uuid = blockOnResult(submitter.submitMessage(message));
        ArtifactVersionMetaDataDto versionMetaDataDto = (ArtifactVersionMetaDataDto) coordinator
                .waitForResponse(uuid);
        outboxEvent.fire(KafkaSqlOutboxEvent.of(ArtifactVersionCreated.of(versionMetaDataDto)));
        return versionMetaDataDto;
    }

    @Override
    public ArtifactVersionMetaDataDto createArtifactVersionIfLatest(String groupId, String artifactId,
            String version, String artifactType, ContentWrapperDto contentDto,
            EditableVersionMetaDataDto metaData, List<String> branches, boolean isDraft, String owner,
            int expectedBaseVersionOrder, EditableArtifactMetaDataDto artifactMetaData)
            throws RegistryStorageException {
        String content = contentDto != null ? contentDto.getContent().content() : null;
        String contentType = contentDto != null ? contentDto.getContentType() : null;
        List<ArtifactReferenceDto> references = contentDto != null ? contentDto.getReferences() : null;
        var message = new CreateArtifactVersionIfLatest11Message(groupId, artifactId, version, artifactType,
                contentType, content, references, metaData, branches, isDraft, owner,
                expectedBaseVersionOrder, artifactMetaData);
        var uuid = blockOnResult(submitter.submitMessage(message));
        ArtifactVersionMetaDataDto versionMetaDataDto = (ArtifactVersionMetaDataDto) coordinator
                .waitForResponse(uuid);
        outboxEvent.fire(KafkaSqlOutboxEvent.of(ArtifactVersionCreated.of(versionMetaDataDto)));
        return versionMetaDataDto;
    }

    @Override
    public void updateArtifactVersionContent(String groupId, String artifactId, String version,
            String artifactType, ContentWrapperDto contentDto) throws RegistryStorageException {
        String content = contentDto != null ? contentDto.getContent().content() : null;
        String contentType = contentDto != null ? contentDto.getContentType() : null;
        List<ArtifactReferenceDto> references = contentDto != null ? contentDto.getReferences() : null;
        var message = new UpdateArtifactVersionContent5Message(groupId, artifactId, version, artifactType,
                contentType, content, references);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#updateArtifactMetaData(java.lang.String,
     *      java.lang.String, io.apicurio.registry.storage.dto.EditableArtifactMetaDataDto)
     */
    @Override
    public void updateArtifactMetaData(String groupId, String artifactId,
            EditableArtifactMetaDataDto metaData) throws ArtifactNotFoundException, RegistryStorageException {
        var message = new UpdateArtifactMetaData3Message(groupId, artifactId, metaData);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
        outboxEvent.fire(KafkaSqlOutboxEvent.of(ArtifactMetadataUpdated.of(groupId, artifactId, metaData)));
    }

    @Override
    public void createArtifactRule(String groupId, String artifactId, RuleType rule,
            RuleConfigurationDto config) throws RegistryStorageException {
        var message = new CreateArtifactRule4Message(groupId, artifactId, rule, config);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
        outboxEvent
                .fire(KafkaSqlOutboxEvent.of(ArtifactRuleConfigured.of(groupId, artifactId, rule, config)));
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#deleteArtifactRules(java.lang.String,
     *      java.lang.String)
     */
    @Override
    public void deleteArtifactRules(String groupId, String artifactId)
            throws ArtifactNotFoundException, RegistryStorageException {
        var message = new DeleteArtifactRules2Message(groupId, artifactId);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#updateArtifactRule(java.lang.String,
     *      java.lang.String, io.apicurio.registry.types.RuleType,
     *      io.apicurio.registry.storage.dto.RuleConfigurationDto)
     */
    @Override
    public void updateArtifactRule(String groupId, String artifactId, RuleType rule,
            RuleConfigurationDto config)
            throws ArtifactNotFoundException, RuleNotFoundException, RegistryStorageException {
        var message = new UpdateArtifactRule4Message(groupId, artifactId, rule, config);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
        outboxEvent
                .fire(KafkaSqlOutboxEvent.of(ArtifactRuleConfigured.of(groupId, artifactId, rule, config)));
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#deleteArtifactRule(java.lang.String,
     *      java.lang.String, io.apicurio.registry.types.RuleType)
     */
    @Override
    public void deleteArtifactRule(String groupId, String artifactId, RuleType rule)
            throws ArtifactNotFoundException, RuleNotFoundException, RegistryStorageException {
        var message = new DeleteArtifactRule3Message(groupId, artifactId, rule);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);

        switch (rule) {
            case VALIDITY ->
                    outboxEvent.fire(KafkaSqlOutboxEvent.of(ArtifactRuleConfigured.of(groupId, artifactId, rule,
                            RuleConfigurationDto.builder().configuration(ValidityLevel.NONE.name()).build())));
            case COMPATIBILITY -> outboxEvent.fire(KafkaSqlOutboxEvent.of(ArtifactRuleConfigured.of(groupId,
                    artifactId, rule,
                    RuleConfigurationDto.builder().configuration(CompatibilityLevel.NONE.name()).build())));
            case INTEGRITY ->
                    outboxEvent.fire(KafkaSqlOutboxEvent.of(ArtifactRuleConfigured.of(groupId, artifactId, rule,
                            RuleConfigurationDto.builder().configuration(IntegrityLevel.NONE.name()).build())));
        }
    }

    @Override
    public void createGroupRule(String groupId, RuleType rule, RuleConfigurationDto config)
            throws RegistryStorageException {
        var message = new CreateGroupRule3Message(groupId, rule, config);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
        outboxEvent.fire(KafkaSqlOutboxEvent.of(GroupRuleConfigured.of(groupId, rule, config)));
    }

    @Override
    public void updateGroupRule(String groupId, RuleType rule, RuleConfigurationDto config)
            throws RegistryStorageException {
        var message = new UpdateGroupRule3Message(groupId, rule, config);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
        outboxEvent.fire(KafkaSqlOutboxEvent.of(GroupRuleConfigured.of(groupId, rule, config)));
    }

    @Override
    public void deleteGroupRule(String groupId, RuleType rule) throws RegistryStorageException {
        var message = new DeleteGroupRule2Message(groupId, rule);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
        switch (rule) {
            case VALIDITY -> outboxEvent.fire(KafkaSqlOutboxEvent.of(GroupRuleConfigured.of(groupId, rule,
                    RuleConfigurationDto.builder().configuration(ValidityLevel.NONE.name()).build())));
            case COMPATIBILITY -> outboxEvent.fire(KafkaSqlOutboxEvent.of(GroupRuleConfigured.of(groupId,
                    rule,
                    RuleConfigurationDto.builder().configuration(CompatibilityLevel.NONE.name()).build())));
            case INTEGRITY -> outboxEvent.fire(KafkaSqlOutboxEvent.of(GroupRuleConfigured.of(groupId, rule,
                    RuleConfigurationDto.builder().configuration(IntegrityLevel.NONE.name()).build())));
        }
    }

    @Override
    public void deleteGroupRules(String groupId) throws RegistryStorageException {
        var message = new DeleteGroupRules1Message(groupId);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#deleteArtifactVersion(java.lang.String,
     *      java.lang.String, java.lang.String)
     */
    @Override
    public void deleteArtifactVersion(String groupId, String artifactId, String version)
            throws ArtifactNotFoundException, VersionNotFoundException, RegistryStorageException {
        var message = new DeleteArtifactVersion3Message(groupId, artifactId, version);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
        outboxEvent.fire(KafkaSqlOutboxEvent.of(ArtifactVersionDeleted.of(groupId, artifactId, version)));
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#updateArtifactVersionMetaData(java.lang.String,
     *      java.lang.String, java.lang.String, io.apicurio.registry.storage.dto.EditableVersionMetaDataDto)
     */
    @Override
    public void updateArtifactVersionMetaData(String groupId, String artifactId, String version,
            EditableVersionMetaDataDto metaData)
            throws ArtifactNotFoundException, VersionNotFoundException, RegistryStorageException {
        var message = new UpdateArtifactVersionMetaData4Message(groupId, artifactId, version, metaData);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
        outboxEvent.fire(KafkaSqlOutboxEvent
                .of(ArtifactVersionMetadataUpdated.of(groupId, artifactId, version, metaData)));
    }

    @Override
    public void updateArtifactVersionState(String groupId, String artifactId, String version,
            VersionState newState, boolean dryRun) {
        var message = new UpdateArtifactVersionState5Message(groupId, artifactId, version, newState, dryRun);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#createGlobalRule(io.apicurio.registry.types.RuleType,
     *      io.apicurio.registry.storage.dto.RuleConfigurationDto)
     */
    @Override
    public void createGlobalRule(RuleType rule, RuleConfigurationDto config)
            throws RuleAlreadyExistsException, RegistryStorageException {
        var message = new CreateGlobalRule2Message(rule, config);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
        outboxEvent.fire(KafkaSqlOutboxEvent.of(GlobalRuleConfigured.of(rule, config)));
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#deleteGlobalRules()
     */
    @Override
    public void deleteGlobalRules() throws RegistryStorageException {
        var message = new DeleteGlobalRules0Message();
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#updateGlobalRule(io.apicurio.registry.types.RuleType,
     *      io.apicurio.registry.storage.dto.RuleConfigurationDto)
     */
    @Override
    public void updateGlobalRule(RuleType rule, RuleConfigurationDto config)
            throws RuleNotFoundException, RegistryStorageException {
        var message = new UpdateGlobalRule2Message(rule, config);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
        outboxEvent.fire(KafkaSqlOutboxEvent.of(GlobalRuleConfigured.of(rule, config)));
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#deleteGlobalRule(io.apicurio.registry.types.RuleType)
     */
    @Override
    public void deleteGlobalRule(RuleType rule) throws RuleNotFoundException, RegistryStorageException {
        var message = new DeleteGlobalRule1Message(rule);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);

        switch (rule) {
            case VALIDITY -> outboxEvent.fire(KafkaSqlOutboxEvent.of(GlobalRuleConfigured.of(rule,
                    RuleConfigurationDto.builder().configuration(ValidityLevel.NONE.name()).build())));
            case COMPATIBILITY -> outboxEvent.fire(KafkaSqlOutboxEvent.of(GlobalRuleConfigured.of(rule,
                    RuleConfigurationDto.builder().configuration(CompatibilityLevel.NONE.name()).build())));
            case INTEGRITY -> outboxEvent.fire(KafkaSqlOutboxEvent.of(GlobalRuleConfigured.of(rule,
                    RuleConfigurationDto.builder().configuration(IntegrityLevel.NONE.name()).build())));
        }
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#createGroup(io.apicurio.registry.storage.dto.GroupMetaDataDto)
     */
    @Override
    public void createGroup(GroupMetaDataDto group)
            throws GroupAlreadyExistsException, RegistryStorageException {
        var message = new CreateGroup1Message(group);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);

        outboxEvent.fire(KafkaSqlOutboxEvent.of(GroupCreated.of(group)));
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#deleteGroup(java.lang.String)
     */
    @Override
    public void deleteGroup(String groupId) throws GroupNotFoundException, RegistryStorageException {
        var message = new DeleteGroup1Message(groupId);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);

        outboxEvent.fire(KafkaSqlOutboxEvent.of(GroupDeleted.of(groupId)));
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#updateGroupMetaData(java.lang.String,
     *      io.apicurio.registry.storage.dto.EditableGroupMetaDataDto)
     */
    @Override
    public void updateGroupMetaData(String groupId, EditableGroupMetaDataDto dto) {
        var message = new UpdateGroupMetaData2Message(groupId, dto);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
        outboxEvent.fire(KafkaSqlOutboxEvent.of(GroupMetadataUpdated.of(groupId, dto)));
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#importData(EntityInputStream, boolean, boolean)
     */
    @Override
    public void importData(EntityInputStream entities, boolean preserveGlobalId, boolean preserveContentId)
            throws RegistryStorageException {
        DataImporter dataImporter = new SqlDataImporter(log, utils, this, preserveGlobalId,
                preserveContentId);
        // All messages use the same partition key (__GLOBAL_PARTITION__), so Kafka guarantees
        // ordering within the partition. Reset messages sent after import messages will be
        // consumed in the correct order without needing any sleep/wait mechanism.
        dataImporter.importData(entities, () -> {});
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#upgradeData(EntityInputStream, boolean, boolean)
     */
    @Override
    public void upgradeData(EntityInputStream entities, boolean preserveGlobalId, boolean preserveContentId)
            throws RegistryStorageException {
        DataImporter dataImporter = new SqlDataUpgrader(log, utils, this, preserveGlobalId,
                preserveContentId);
        // All messages use the same partition key (__GLOBAL_PARTITION__), so Kafka guarantees
        // ordering within the partition. Reset messages sent after import messages will be
        // consumed in the correct order without needing any sleep/wait mechanism.
        dataImporter.importData(entities, () -> {});
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#createRoleMapping(java.lang.String, java.lang.String,
     *      java.lang.String)
     */
    @Override
    public void createRoleMapping(String principalId, String role, String principalName)
            throws RegistryStorageException {
        var message = new CreateRoleMapping3Message(principalId, role, principalName);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#updateRoleMapping(java.lang.String, java.lang.String)
     */
    @Override
    public void updateRoleMapping(String principalId, String role) throws RegistryStorageException {
        var message = new UpdateRoleMapping2Message(principalId, role);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#deleteRoleMapping(java.lang.String)
     */
    @Override
    public void deleteRoleMapping(String principalId) throws RegistryStorageException {
        var message = new DeleteRoleMapping1Message(principalId);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#deleteAllUserData()
     */
    @Override
    public void deleteAllUserData() {
        var message = new DeleteAllUserData0Message();
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#createDownload(io.apicurio.registry.storage.dto.DownloadContextDto)
     */
    @Override
    public String createDownload(DownloadContextDto context) throws RegistryStorageException {
        var message = new CreateDownload1Message(context);
        var uuid = blockOnResult(submitter.submitMessage(message));
        return (String) coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#consumeDownload(java.lang.String)
     */
    @Override
    public DownloadContextDto consumeDownload(String downloadId) throws RegistryStorageException {
        var message = new ConsumeDownload1Message(downloadId);
        var uuid = blockOnResult(submitter.submitMessage(message));
        return (DownloadContextDto) coordinator.waitForResponse(uuid);
    }

    /**
     * Bypasses the Kafka journal and writes directly to SQL. This operation is triggered by a
     * scheduled job that runs independently on every pod, so coordinating via the journal is
     * unnecessary and would add bloat that slows down new-pod bootstrap.
     *
     * @see io.apicurio.registry.storage.RegistryStorage#deleteAllExpiredDownloads()
     */
    @Override
    public void deleteAllExpiredDownloads() throws RegistryStorageException {
        sqlStore.deleteAllExpiredDownloads();
    }

    /**
     * Bypasses the Kafka journal and writes directly to SQL. This operation is triggered by a
     * scheduled job that runs independently on every pod, so coordinating via the journal is
     * unnecessary and would add bloat that slows down new-pod bootstrap.
     *
     * @see io.apicurio.registry.storage.RegistryStorage#deleteAllOrphanedContent()
     */
    @Override
    public void deleteAllOrphanedContent() throws RegistryStorageException {
        sqlStore.deleteAllOrphanedContent();
    }

    @Override
    public ContentWrapperDto getContentByReference(ArtifactReferenceDto reference) {
        return sqlStore.getContentByReference(reference);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#createArtifactVersionComment(java.lang.String,
     *      java.lang.String, java.lang.String, java.lang.String)
     */
    @Override
    public CommentDto createArtifactVersionComment(String groupId, String artifactId, String version,
            String value) {
        var message = new CreateArtifactVersionComment4Message(groupId, artifactId, version, value);
        var uuid = blockOnResult(submitter.submitMessage(message));
        return (CommentDto) coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#deleteArtifactVersionComment(java.lang.String,
     *      java.lang.String, java.lang.String, java.lang.String)
     */
    @Override
    public void deleteArtifactVersionComment(String groupId, String artifactId, String version,
            String commentId) {
        var message = new DeleteArtifactVersionComment4Message(groupId, artifactId, version, commentId);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#updateArtifactVersionComment(java.lang.String,
     *      java.lang.String, java.lang.String, java.lang.String, java.lang.String)
     */
    @Override
    public void updateArtifactVersionComment(String groupId, String artifactId, String version,
            String commentId, String value) {
        var message = new UpdateArtifactVersionComment5Message(groupId, artifactId, version, commentId,
                value);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#resetGlobalId()
     */
    @Override
    public void resetGlobalId() {
        var message = new ResetGlobalId0Message();
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#resetContentId()
     */
    @Override
    public void resetContentId() {
        var message = new ResetContentId0Message();
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#resetCommentId()
     */
    @Override
    public void resetCommentId() {
        var message = new ResetCommentId0Message();
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#nextContentId()
     */
    @Override
    public long nextContentId() {
        var message = new NextContentId0Message();
        var uuid = blockOnResult(submitter.submitMessage(message));
        return (long) coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#nextGlobalId()
     */
    @Override
    public long nextGlobalId() {
        var message = new NextGlobalId0Message();
        var uuid = blockOnResult(submitter.submitMessage(message));
        return (long) coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#nextCommentId()
     */
    @Override
    public long nextCommentId() {
        var message = new NextCommentId0Message();
        var uuid = blockOnResult(submitter.submitMessage(message));
        return (long) coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#importComment(CommentEntity)
     */
    @Override
    public void importComment(CommentEntity entity) {
        var message = new ImportComment1Message(entity);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#importGroup(GroupEntity)
     */
    @Override
    public void importGroup(GroupEntity entity) {
        var message = new ImportGroup1Message(entity);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#importGlobalRule(GlobalRuleEntity)
     */
    @Override
    public void importGlobalRule(GlobalRuleEntity entity) {
        var message = new ImportGlobalRule1Message(entity);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#importContent(ContentEntity)
     */
    @Override
    public void importContent(ContentEntity entity) {
        String content = ContentHandle.create(entity.contentBytes).content();
        var message = new ImportContent1Message(entity, content);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#importArtifactVersion(ArtifactVersionEntity)
     */
    @Override
    public void importArtifactVersion(ArtifactVersionEntity entity) {
        var message = new ImportArtifactVersion1Message(entity);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    @Override
    public void importArtifact(ArtifactEntity entity) {
        var message = new ImportArtifact1Message(entity);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#importArtifactRule(ArtifactRuleEntity)
     */
    @Override
    public void importArtifactRule(ArtifactRuleEntity entity) {
        var message = new ImportArtifactRule1Message(entity);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    @Override
    public void importGroupRule(GroupRuleEntity entity) {
        var message = new ImportGroupRule1Message(entity);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#importBranch(BranchEntity)
     */
    @Override
    public void importBranch(BranchEntity entity) {
        var message = new ImportBranch1Message(entity);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#updateContentCanonicalHash(java.lang.String, long,
     *      java.lang.String)
     */
    @Override
    public void updateContentCanonicalHash(String newCanonicalHash, long contentId, String contentHash) {
        var message = new UpdateContentCanonicalHash3Message(newCanonicalHash, contentId, contentHash);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    @Override
    public void appendVersionToBranch(GA ga, BranchId branchId, VersionId version) {
        var message = new AppendVersionToBranch3Message(ga.getRawGroupIdWithNull(), ga.getRawArtifactId(),
                branchId.getRawBranchId(), version.getRawVersionId());
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    @Override
    public void updateBranchMetaData(GA ga, BranchId branchId, EditableBranchMetaDataDto dto) {
        var message = new UpdateBranchMetaData3Message(ga.getRawGroupIdWithNull(), ga.getRawArtifactId(),
                branchId.getRawBranchId(), dto);
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    @Override
    public void replaceBranchVersions(GA ga, BranchId branchId, List<VersionId> versions) {
        var message = new ReplaceBranchVersions3Message(ga.getRawGroupIdWithNull(), ga.getRawArtifactId(),
                branchId.getRawBranchId(), versions.stream().map(VersionId::getRawVersionId).toList());
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    @Override
    public BranchMetaDataDto createBranch(GA ga, BranchId branchId, String description,
            List<String> versions) {
        var message = new CreateBranch4Message(ga.getRawGroupIdWithNull(), ga.getRawArtifactId(),
                branchId.getRawBranchId(), description, versions);
        var uuid = blockOnResult(submitter.submitMessage(message));
        return (BranchMetaDataDto) coordinator.waitForResponse(uuid);
    }

    /**
     * @see io.apicurio.registry.storage.RegistryStorage#deleteBranch(io.apicurio.registry.model.GA,
     *      io.apicurio.registry.model.BranchId)
     */
    @Override
    public void deleteBranch(GA ga, BranchId branchId) {
        var message = new DeleteBranch2Message(ga.getRawGroupIdWithNull(), ga.getRawArtifactId(),
                branchId.getRawBranchId());
        var uuid = blockOnResult(submitter.submitMessage(message));
        coordinator.waitForResponse(uuid);
    }

    @Override
    public String triggerSnapshotCreation() throws RegistryStorageException {
        // First we generate an identifier for the snapshot, then we send a snapshot marker to the journal
        // topic.
        String snapshotId = UUID.randomUUID().toString();
        Path path = Path.of(configuration.getSnapshotStoreLocation(), snapshotId + ".sql");
        var message = new CreateSnapshot1Message(path.toString(), snapshotId);
        this.lastTriggeredSnapshot = snapshotId;
        log.debug("Snapshot with id {} triggered.", snapshotId);
        var uuid = blockOnResult(submitter.submitMessage(message));
        String snapshotLocation = (String) coordinator.waitForResponse(uuid);
        // Then we send a new message to the snapshots topic, using the snapshot id as the key of the snapshot
        // message.
        ProducerRecord<String, String> record = new ProducerRecord<>(configuration.getSnapshotsTopic(), 0,
                snapshotId, snapshotLocation, Collections.emptyList());
        RecordMetadata recordMetadata = blockOnResult(snapshotsProducer.apply(record));
        return snapshotLocation;
    }

    @Override
    public String createSnapshot(String snapshotLocation) throws RegistryStorageException {
        throw new IllegalStateException("Directly creating a snapshot is not supported in Kafkasql");
    }

    @Override
    public String createEvent(OutboxEvent event) {
        // No op, the event is created by the event processor.
        return event.getId();
    }

}
