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());
        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;
            default:
                throw new RuntimeException("Unhandled record type " + type);
        }
    }
@Test
    public void testActivationRecordsPartialTransactionNoSupport() {
        SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
        FeatureControlManager featureControlManager = new FeatureControlManager.Builder()
            .setSnapshotRegistry(snapshotRegistry)
            .setMetadataVersion(MetadataVersion.IBP_3_6_IV0)
            .build();

        OffsetControlManager offsetControlManager = new OffsetControlManager.Builder().build();
        offsetControlManager.replay(new BeginTransactionRecord(), 10);
        offsetControlManager.handleCommitBatch(Batch.data(20, 1, 1L, 0,
            Collections.singletonList(new ApiMessageAndVersion(new BeginTransactionRecord(), (short) 0))));

        assertThrows(RuntimeException.class, () ->
            ActivationRecordsGenerator.generate(
                msg -> { },
                false,
                offsetControlManager.transactionStartOffset(),
                false,
                BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_6_IV0, "test"),
                featureControlManager)
        );
    }