class AbstractDDLTransportAction<Request extends AcknowledgedRequest<Request>, Response extends AcknowledgedResponse> extends TransportMasterNodeAction<Request, Response> {

    private final Writeable.Reader<Response> reader;
    private final Function<Boolean, Response> ackedResponseFunction;
    private final String source;

    public AbstractDDLTransportAction(String actionName,
                                      TransportService transportService,
                                      ClusterService clusterService,
                                      ThreadPool threadPool,
                                      Writeable.Reader<Request> requestReader,
                                      Writeable.Reader<Response> responseReader,
                                      Function<Boolean, Response> ackedResponseFunction,
                                      String source) {
        super(actionName, transportService, clusterService, threadPool, requestReader);
        this.reader = responseReader;
        this.ackedResponseFunction = ackedResponseFunction;
        this.source = source;
    }

    @Override
    protected String executor() {
        // no need to use a thread pool, we go async right away
        return ThreadPool.Names.SAME;
    }

    @Override
    protected Response read(StreamInput in) throws IOException {
        return reader.read(in);
    }

    @Override
    protected void masterOperation(Request request,
                                   ClusterState state,
                                   ActionListener<Response> listener) throws Exception {<FILL_FUNCTION_BODY>}

    public abstract ClusterStateTaskExecutor<Request> clusterStateTaskExecutor(Request request);
}

        clusterService.submitStateUpdateTask(source,
            request,
            ClusterStateTaskConfig.build(Priority.HIGH, request.masterNodeTimeout()),
            clusterStateTaskExecutor(request),
            new AckedClusterStateTaskListener() {
                @Override
                public void onFailure(String source, Exception e) {
                    listener.onFailure(e);
                }

                @Override
                public boolean mustAck(DiscoveryNode discoveryNode) {
                    return true;
                }

                @Override
                public void onAllNodesAcked(@Nullable Exception e) {
                    listener.onResponse(ackedResponseFunction.apply(true));
                }

                @Override
                public void onAckTimeout() {
                    listener.onResponse(ackedResponseFunction.apply(false));
                }

                @Override
                public TimeValue ackTimeout() {
                    return request.ackTimeout();
                }
            });

    
<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