class ClientComputeTask implements ClientCloseableResource {
    /** No failover flag mask. */
    private static final byte NO_FAILOVER_FLAG_MASK = 0x01;

    /** No result cache flag mask. */
    private static final byte NO_RESULT_CACHE_FLAG_MASK = 0x02;

    /** Keep binary flag mask. */
    public static final byte KEEP_BINARY_FLAG_MASK = 0x04;

    /** Context. */
    private final ClientConnectionContext ctx;

    /** Logger. */
    private final IgniteLogger log;

    /** Task id. */
    private volatile long taskId;

    /** Task future. */
    private volatile ComputeTaskInternalFuture<Object> taskFut;

    /** Task closed flag. */
    private final AtomicBoolean closed = new AtomicBoolean();

    /**
     * Ctor.
     *
     * @param ctx Connection context.
     */
    ClientComputeTask(ClientConnectionContext ctx) {
        assert ctx != null;

        this.ctx = ctx;

        log = ctx.kernalContext().log(getClass());
    }

    /**
     * @param taskId Task ID.
     * @param taskName Task name.
     * @param arg Task arguments.
     * @param nodeIds Nodes to run task jobs.
     * @param flags Flags for task.
     * @param timeout Task timeout.
     */
    void execute(long taskId, String taskName, Object arg, Set<UUID> nodeIds, byte flags, long timeout) {
        assert taskName != null;

        this.taskId = taskId;

        GridTaskProcessor task = ctx.kernalContext().task();

        IgnitePredicate<ClusterNode> nodePredicate = F.isEmpty(nodeIds) ? node -> !node.isClient() :
            F.nodeForNodeIds(nodeIds);

        TaskExecutionOptions opts = options()
            .asPublicRequest()
            .withProjectionPredicate(nodePredicate)
            .withTimeout(timeout);

        if ((flags & NO_FAILOVER_FLAG_MASK) != 0)
            opts.withFailoverDisabled();

        if ((flags & NO_RESULT_CACHE_FLAG_MASK) != 0)
            opts.withResultCacheDisabled();

        taskFut = task.execute(taskName, arg, opts);

        // Fail fast.
        if (taskFut.isDone() && taskFut.error() != null)
            throw new IgniteClientException(ClientStatus.FAILED, taskFut.error().getMessage());
    }

    /**
     * Callback for response sent event.
     */
    void onResponseSent() {
        // Listener should be registered only after response for this task was sent, to ensure that client doesn't
        // receive notification before response for the task.
        taskFut.listen(() -> {
            try {
                ClientNotification notification;

                if (taskFut.error() != null) {
                    String msg = ctx.kernalContext().clientListener().sendServerExceptionStackTraceToClient()
                            ? taskFut.error().getMessage() + U.nl() + X.getFullStackTrace(taskFut.error())
                            : taskFut.error().getMessage();

                    notification = new ClientNotification(OP_COMPUTE_TASK_FINISHED, taskId, msg);
                }
                else if (taskFut.isCancelled())
                    notification = new ClientNotification(OP_COMPUTE_TASK_FINISHED, taskId, "Task was cancelled");
                else
                    notification = new ClientObjectNotification(OP_COMPUTE_TASK_FINISHED, taskId, taskFut.result());

                ctx.notifyClient(notification);
            }
            finally {
                // If task was explicitly closed before, resource is already released.
                if (closed.compareAndSet(false, true)) {
                    ctx.decrementActiveTasksCount();

                    ctx.resources().release(taskId);
                }
            }
        });
    }

    /**
     * Gets task ID.
     */
    long taskId() {
        return taskId;
    }

    /**
     * Closes the task resource.
     */
    @Override public void close() {<FILL_FUNCTION_BODY>}
}

        if (closed.compareAndSet(false, true)) {
            ctx.decrementActiveTasksCount();

            try {
                if (taskFut != null)
                    taskFut.cancel();
            }
            catch (IgniteCheckedException e) {
                log.warning("Failed to cancel task", e);
            }
        }
    
<no_super_class>