Interrupt Thread (Java) does not kill execution in Oracle database

Question

Guys, I’m working on a script runner PL/SQL, using the JDBC Statement to make my connections with the bank. My problem is that the project works with Threads and when the user wants to interrupt the execution of some script, I use the interrupt() which ends up interrupting Thread in fact, but the script is still running and when I view the SID situation in the bank session, it’s still like active. Performing other tests, where this script writes something on disk, it keeps increasing even after the interrupt() in Thread. I’ve tried to force the statement.close() and the statement.cancel(), but without success.

Here is my class that runs the script:

package com.dxc.scheduler.repository;

import static com.dxc.scheduler.exception.JobInterruptedException.checkInterruptedThread;
import static java.lang.String.format;
import static java.util.Optional.empty;
import static java.util.Optional.ofNullable;

import java.sql.Array;
import java.sql.CallableStatement;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import javax.sql.DataSource;

import com.dxc.commons.exception.FailureException;
import com.dxc.commons.slf4j.Logger;
import com.dxc.commons.slf4j.LoggerFactory;
import com.dxc.scheduler.domain.Step;
import com.dxc.scheduler.exception.JobInterruptedException;
import com.dxc.scheduler.exception.StepException;

public class OracleRepository implements DatabaseRepository {

    private static final String JOBID_KEY = "job_id";
    private static final String SEQUENCE_KEY = "sequence";

    private static final Logger LOGGER = LoggerFactory.getLogger(OracleRepository.class);

    private static final String DML_SESSION_ID = "SELECT sid FROM V$SESSION WHERE audsid = userenv('sessionid')";

    private static final String DBMSOUTPUT_ENABLE = "begin dbms_output.enable(); end;";
    private static final String DBMSOUTPUT_DISABLE = "begin dbms_output.disable(); end;";
    private static final Integer DBMSOUTPUT_BUFFER = 100_000_000;
    private static final int DBMSOUTPUT_LINES = 100_000_000;

    private final DataSource dataSource;

    public OracleRepository(final DataSource ds) {
        if (ds == null) {
            throw new FailureException(
                    "invalid datasource, please use the OracleRepositoryHandler.getDataSource() method to get a valid datasource.");
        }

        dataSource = ds;
    }

    private void enableOrDisableOutput(final Statement st, final boolean enable) throws SQLException {
        st.executeUpdate(enable ? DBMSOUTPUT_ENABLE : DBMSOUTPUT_DISABLE);
    }

    @Override
    public String execute(final Step step) throws SQLException, JobInterruptedException {
        try (Connection conn = dataSource.getConnection(); Statement st = conn.createStatement()) {
            try {
                enableOrDisableOutput(st, true);

                LOGGER.info().message("running script...")
                        .extra("script_name", step.getScriptName(), JOBID_KEY, step.getJob().getId(), SEQUENCE_KEY, step.getSequence()).log();

                final String script = wrapScriptInTheTemplate(step);
                LOGGER.trace().message(script).log();
                st.executeUpdate(script);

                LOGGER.info().message("getting output script...")
                        .extra("script_name", step.getScriptName(), JOBID_KEY, step.getJob().getId(), SEQUENCE_KEY, step.getSequence()).log();

                return getOutput(conn);
            } catch (final Exception e) {
                checkInterruptedThread();

                String output = "";
                try {
                    output = getOutput(conn);
                } catch (final Exception ex) {
                    LOGGER.warn().message("could not get the script output")
                            .extra("script_name", step.getScriptName(), JOBID_KEY, step.getJob().getId(), SEQUENCE_KEY, step.getSequence())
                            .log();
                }
                throw new StepException(e.getMessage(), output, e);
            } finally {
                try {
                    enableOrDisableOutput(st, false);
                } catch (final SQLException e) {
                    LOGGER.error().message("could not disable the output...")
                            .extra(JOBID_KEY, step.getJob().getId(), SEQUENCE_KEY, step.getScriptName(), "error", e.getMessage()).log();
                }
            }
        }
    }

    private String getOutput(final Connection conn) throws SQLException {
        Array array = null;

        try (CallableStatement call = conn.prepareCall(getScriptOuputLines(DBMSOUTPUT_LINES))) {
            call.registerOutParameter(1, Types.ARRAY, "DBMSOUTPUT_LINESARRAY");
            call.execute();
            array = call.getArray(1);

            final String output = Stream.of((Object[]) array.getArray()).filter(v -> v != null).map(Object::toString)
                    .collect(Collectors.joining("\n"));

            LOGGER.debug().message("the script output was retrieved").extra("lenght", output.length()).log();

            return output;
        } finally {
            if (array != null) {
                array.free();
            }
        }
    }

    private String getScriptOuputLines(final int number) {
        return format("declare num integer := %d; begin dbms_output.get_lines(?, num); end;", number);
    }

    @Override
    @SuppressWarnings("PMD.AvoidBranchingStatementAsLastInLoop")
    public Optional<Integer> getSessionId() throws SQLException {
        try (Connection conn = dataSource.getConnection(); PreparedStatement ps = conn.prepareStatement(DML_SESSION_ID)) {
            try (ResultSet rs = ps.executeQuery()) {
                while (rs.next()) {
                    final int sessionId = rs.getInt(1);
                    LOGGER.debug().message("the session id was retrieved").extra("session_id", sessionId).log();
                    return ofNullable(sessionId);
                }
            }
        }
        return empty();
    }

    private String wrapScriptInTheTemplate(final Step step) {
        return format("begin dbms_output.enable(%d); dbms_application_info.set_action('%s_%s_%s'); %s end;", DBMSOUTPUT_BUFFER,
                step.getJob().getType(), step.getJob().getSource(), step.getSequence(), step.getSource());
    }

}

Here’s the endpoint I use to stop the execution:

@ResponseBody
@PutMapping("jobs/{id}/interrupt")
public ResponseEntity<?> interrupt(@PathVariable final Long id, @RequestHeader("X-SCHEDULER2-APIKEY") final Optional<String> token) {

    final Optional<ResponseEntity<?>> res = checkAPIToken(token);
    if (res.isPresent()) {
        return res.get();
    }

    final Optional<Job> job = jobRepository.findByIdAndStatus(id, JobStatus.RUNNING);
    if (job.isPresent()) {
        jobLifecycleManager.interrupt(id);
        return new ResponseEntity<>(ACCEPTED);
    }
    final ErrorDataResponse response = new ErrorDataResponse(of("job not found or job is not in running"), NOT_FOUND);
    return new ResponseEntity<>(response, response.getHttpStatus());
}

Here the called method redirects to the class that executes the script and terminates it in the "Finally" of Exception:

public void interrupt(final Long jobId) {
    final Optional<JobMetadata> jm = get(jobId);
    if (jm.isPresent()) {
        final long threadId = jm.get().getThreadId();
        getAllStackTraces().keySet().stream().filter(t -> t.getId() == threadId).forEach(t -> t.interrupt());
    }
}

And after that, is updated the status of the script in the bank, as interrupted. Why even after interrupting the thread the script still runs?

Answer 1

From the moment you have the script run I believe that stopping Thread will no longer help, you will have to send a

ALTER SYSTEM KILL SESSION 'sid,serial#';

after interrupting Thread, creates a Statement and passes the command...