private void replay(ApiMessageAndVersion record) {
        MetadataRecordType type = MetadataRecordType.fromId(record.message().apiKey());
        switch (type) {
            case BEGIN_TRANSACTION_RECORD:
                if (transactionState == TransactionState.STARTED_TRANSACTION ||
                    transactionState == TransactionState.CONTINUED_TRANSACTION) {
                    throw new RuntimeException("Encountered BeginTransactionRecord while already in a transaction");
                } else {
                    transactionState = TransactionState.STARTED_TRANSACTION;
                }
                break;
            case END_TRANSACTION_RECORD:
                if (transactionState == TransactionState.CONTINUED_TRANSACTION ||
                    transactionState == TransactionState.STARTED_TRANSACTION) {
                    transactionState = TransactionState.ENDED_TRANSACTION;
                } else {
                    throw new RuntimeException("Encountered EndTransactionRecord without having seen a BeginTransactionRecord");
                }
                break;
            case ABORT_TRANSACTION_RECORD:
                if (transactionState == TransactionState.CONTINUED_TRANSACTION ||
                    transactionState == TransactionState.STARTED_TRANSACTION) {
                    transactionState = TransactionState.ABORTED_TRANSACTION;
                } else {
                    throw new RuntimeException("Encountered AbortTransactionRecord without having seen a BeginTransactionRecord");
                }
                break;
            default:
                switch (transactionState) {
                    case STARTED_TRANSACTION:
                        // If we see a non-transaction record after starting a transaction, transition to CONTINUED_TRANSACTION
                        transactionState = TransactionState.CONTINUED_TRANSACTION;
                        break;
                    case ENDED_TRANSACTION:
                    case ABORTED_TRANSACTION:
                        // If we see a non-transaction record after ending a transaction, transition back to NO_TRANSACTION
                        transactionState = TransactionState.NO_TRANSACTION;
                        break;
                    case CONTINUED_TRANSACTION:
                    case NO_TRANSACTION:
                    default:
                        break;
                }
                hasSeenRecord = true;
                delta.replay(record.message());
        }
    }
Flush the metadata accumulated in this batch loader if not in the middle of a transaction. The
flushed metadata will be passed to the {@link MetadataUpdater} configured for this class.
metadata/src/main/java/org/apache/kafka/image/loader/MetadataBatchLoader.java