class KillPlan implements Plan {

    @Nullable
    private final Symbol jobId;

    public KillPlan(@Nullable Symbol jobId) {
        this.jobId = jobId;
    }

    @Override
    public StatementType type() {
        return StatementType.MANAGEMENT;
    }

    @Override
    public void executeOrFail(DependencyCarrier dependencies,
                              PlannerContext plannerContext,
                              RowConsumer consumer,
                              Row params,
                              SubQueryResults subQueryResults) {
        execute(
            boundJobId(
                jobId,
                plannerContext.transactionContext(),
                dependencies.nodeContext(),
                params,
                subQueryResults),
            plannerContext.transactionContext().sessionSettings().userName(),
            req -> dependencies.client().execute(KillJobsNodeAction.INSTANCE, req),
            req -> dependencies.client().execute(KillAllNodeAction.INSTANCE, req),
            consumer
        );
    }

    @VisibleForTesting
    @Nullable
    public static UUID boundJobId(@Nullable Symbol jobId,
                                  CoordinatorTxnCtx txnCtx,
                                  NodeContext nodeCtx,
                                  Row parameters,
                                  SubQueryResults subQueryResults) {
        if (jobId != null) {
            try {
                return UUID.fromString(
                    DataTypes.STRING.sanitizeValue(
                        SymbolEvaluator.evaluate(
                            txnCtx,
                            nodeCtx,
                            jobId,
                            parameters,
                            subQueryResults
                        )));
            } catch (Exception e) {
                throw new IllegalArgumentException("Can not parse job ID: " + jobId, e);
            }
        }
        return null;
    }

    @VisibleForTesting
    void execute(@Nullable UUID jobId,
                 String userName,
                 ActionExecutor<KillJobsNodeRequest, KillResponse> killJobsNodeAction,
                 ActionExecutor<KillAllRequest, KillResponse> killAllNodeAction,
                 RowConsumer consumer) {<FILL_FUNCTION_BODY>}
}

        if (jobId != null) {
            killJobsNodeAction
                .execute(
                    new KillJobsNodeRequest(
                        List.of(),
                        List.of(jobId),
                        userName,
                        "KILL invoked by user: " + userName))
                .whenComplete(
                    new OneRowActionListener<>(
                        consumer,
                        killResponse -> new Row1(killResponse.numKilled()))
                );
        } else {
            killAllNodeAction
                .execute(
                    new KillAllRequest(userName))
                .whenComplete(
                    new OneRowActionListener<>(
                        consumer,
                        killResponse -> new Row1(killResponse.numKilled()))
                );
        }
    
<no_super_class>