class TransportCreateSubscriptionAction extends TransportMasterNodeAction<CreateSubscriptionRequest, AcknowledgedResponse> {

    public static final String ACTION_NAME = "internal:crate:replication/logical/subscription/create";

    private final String source;
    private final LogicalReplicationService logicalReplicationService;
    private final Roles roles;

    @Inject
    public TransportCreateSubscriptionAction(TransportService transportService,
                                             ClusterService clusterService,
                                             LogicalReplicationService logicalReplicationService,
                                             ThreadPool threadPool,
                                             Roles roles) {
        super(ACTION_NAME,
              transportService,
              clusterService,
              threadPool,
              CreateSubscriptionRequest::new);
        this.logicalReplicationService = logicalReplicationService;
        this.roles = roles;
        this.source = "create-subscription";
    }

    @Override
    protected String executor() {
        return ThreadPool.Names.SAME;
    }

    @Override
    protected AcknowledgedResponse read(StreamInput in) throws IOException {
        return new AcknowledgedResponse(in);
    }

    @Override
    protected void masterOperation(CreateSubscriptionRequest request,
                                   ClusterState state,
                                   ActionListener<AcknowledgedResponse> listener) throws Exception {

        // Ensure subscription owner exists
        if (roles.findUser(request.owner()) == null) {
            throw new IllegalStateException(
                String.format(
                    Locale.ENGLISH, "Subscription '%s' cannot be created as the user '%s' owning the subscription has been dropped.",
                    request.name(),
                    request.owner()
                )
            );
        }

        logicalReplicationService.getPublicationState(
                request.name(),
                request.publications(),
                request.connectionInfo()
            )
            .thenCompose(
                response -> {
                    if (response.unknownPublications().isEmpty() == false) {
                        throw new PublicationUnknownException(response.unknownPublications().get(0));
                    }

                    // Published tables can have metadata or documents which subscriber with a lower version might not process.
                    // We check published tables version and not publisher cluster's MinNodeVersion.
                    // Publisher cluster can have a higher version but contain old tables, restored from a snapshot,
                    // in this case subscription works fine.
                    for (RelationMetadata relationMetadata: response.relationsInPublications().values()) {
                        if (relationMetadata.template() != null) {
                            checkVersionCompatibility(
                                relationMetadata.name().fqn(),
                                state.nodes().getMinNodeVersion(),
                                relationMetadata.template().settings()
                            );
                        }
                        if (!relationMetadata.indices().isEmpty()) {
                            // All indices belong to the same table and has same metadata.
                            IndexMetadata indexMetadata = relationMetadata.indices().get(0);
                            checkVersionCompatibility(
                                relationMetadata.name().fqn(),
                                state.nodes().getMinNodeVersion(),
                                indexMetadata.getSettings()
                            );
                        }
                    }

                    logicalReplicationService.verifyTablesDoNotExist(request.name(), response);
                    return submitClusterStateTask(request, response);
                }
            )
            .whenComplete(
                (acknowledgedResponse, err) -> {
                    if (err == null) {
                        listener.onResponse(acknowledgedResponse);
                    } else {
                        listener.onFailure(Exceptions.toException(err));
                    }
                }
            );
    }

    private static void checkVersionCompatibility(String tableFqn, Version subscriberMinNodeVersion, Settings settings) {<FILL_FUNCTION_BODY>}

    private CompletableFuture<AcknowledgedResponse> submitClusterStateTask(CreateSubscriptionRequest request,
                                                                           PublicationsStateAction.Response publicationsStateResponse) {

        AckedClusterStateUpdateTask<AcknowledgedResponse> task = new AckedClusterStateUpdateTask<>(request) {
            @Override
            public ClusterState execute(ClusterState currentState) throws Exception {
                Metadata currentMetadata = currentState.metadata();
                Metadata.Builder mdBuilder = Metadata.builder(currentMetadata);

                var oldMetadata = (SubscriptionsMetadata) mdBuilder.getCustom(SubscriptionsMetadata.TYPE);
                if (oldMetadata != null && oldMetadata.subscription().containsKey(request.name())) {
                    throw new SubscriptionAlreadyExistsException(request.name());
                }

                HashMap<RelationName, Subscription.RelationState> relations = new HashMap<>();
                for (var relation : publicationsStateResponse.tables()) {
                    relations.put(
                        relation,
                        new Subscription.RelationState(Subscription.State.INITIALIZING, null)
                    );
                }

                Subscription subscription = new Subscription(
                    request.owner(),
                    request.connectionInfo(),
                    request.publications(),
                    request.settings(),
                    relations
                );


                var newMetadata = SubscriptionsMetadata.newInstance(oldMetadata);
                newMetadata.subscription().put(request.name(), subscription);
                assert !newMetadata.equals(oldMetadata) : "must not be equal to guarantee the cluster change action";
                mdBuilder.putCustom(SubscriptionsMetadata.TYPE, newMetadata);

                return ClusterState.builder(currentState).metadata(mdBuilder).build();
            }

            @Override
            protected AcknowledgedResponse newResponse(boolean acknowledged) {
                return new AcknowledgedResponse(acknowledged);
            }
        };

        clusterService.submitStateUpdateTask(source, task);
        return task.completionFuture();
    }

    @Override
    protected ClusterBlockException checkBlock(CreateSubscriptionRequest request, ClusterState state) {
        return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_WRITE);
    }
}

        Version publishedTableVersion = settings.getAsVersion(IndexMetadata.SETTING_VERSION_CREATED, null);
        assert publishedTableVersion != null : "All published tables must have version created setting";
        if (subscriberMinNodeVersion.beforeMajorMinor(publishedTableVersion)) {
            throw new IllegalStateException(String.format(
                Locale.ENGLISH,
                "One of the published tables has version higher than subscriber's minimal node version." +
                " Table=%s, Table-Version=%s, Local-Minimal-Version: %s",
                tableFqn,
                publishedTableVersion,
                subscriberMinNodeVersion
            ));
        }
    
<methods><variables>protected final non-sealed org.elasticsearch.cluster.service.ClusterService clusterService,private final non-sealed java.lang.String executor,protected final non-sealed org.elasticsearch.threadpool.ThreadPool threadPool,protected final non-sealed org.elasticsearch.transport.TransportService transportService