Java Multithreaded - Better way to cancel Future task with database and http connections?

Question

I am having difficulty trying to correctly program my application in the way I want it to behave.

Currently, my application (as a Java Servlet) will query the database for a list of items to process. For every item in the list, it will submit an HTTP Post request. I am trying to create a way where I can stop this processing (and even terminate the HTTP Post request in progress) if the user requests. There can be simultaneous threads that are separately processing different queries. Right now, I will stop processing in all threads.

My current attempt involves implementing the database query and HTTP Post in a Callable class. Then I submit the Callable class via the Executor Service to get a Future object.

However, in order properly to stop the processing, I need to abort the HTTP Post and close the database's Connection, Statement and ResultSet - because the Future.cancel() will not do this for me. How can I do this when I call cancel() on the Future object? Do I have to store a List of Arrays that contains the Future object, HttpPost, Connection, Statement, and ResultSet? This seems overkill - surely there must be a better way?

Here is some code I have right now that only aborts the HttpPost (and not any database objects).

private static final ExecutorService pool = Executors.newFixedThreadPool(10);
public static Future<HttpClient> upload(final String url) {
    CallableTask ctask = new CallableTask();
    ctask.setFile(largeFile);
    ctask.setUrl(url);
    Future<HttpClient> f = pool.submit(ctask); //This will create an HttpPost that posts 'largefile' to the 'url'
    linklist.add(new tuple<Future<HttpClient>, HttpPost>(f, ctask.getPost())); //storing the objects for when I cancel later
    return f;
}


//This method cancels all running Future tasks and aborts any POSTs in progress
public static void cancelAll() {
    System.out.println("Checking status...");

    for (tuple<Future<HttpClient>, HttpPost> t : linklist) {
        Future<HttpClient> f = t.getFuture();
        HttpPost post = t.getPost();
        if (f.isDone()) {
            System.out.println("Task is done!");
        } else {
            if (f.isCancelled()) {
                System.out.println("Task was cancelled!");
            } else {
                while (!f.isDone()) {
                        f.cancel(true);
                        try {
                            Thread.sleep(5000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("!Aborting Post!");
                        try {
                            post.abort();
                        } catch (Exception ex) {
                            System.out.println("Aborted Post, swallowing exception: ");
                            ex.printStackTrace();
                        }

                }
            }
        }
    }
}

Is there an easier way or a better design? Right now I terminate all processing threads - in the future, I would like to terminate individual threads.

Answer 1

I think keeping a list of all the resources to be closed is not the best approach. In your current code, it seems that the HTTP request is initiated by the CallableTask but the closing is done by somebody else. Closing resources is the responsibility of the one who opened it, in my opinion.

I would let CallableTask to initiate the HTTP request, connect to database and do it's stuff and, when it is finished or aborted, it should close everything it opened. This way you have to keep track only the Future instances representing your currently running tasks.

Answer 2

I think your approach is correct. You would need to handle the rollback yourself when you are canceling the thread

cancel() just calls interrupt() for already executing thread. Have a look here

http://docs.oracle.com/javase/tutorial/essential/concurrency/interrupt.html:

As it says

An interrupt is an indication to a thread that it should stop what it is doing and do something else. It's up to the programmer to decide exactly how a thread responds to an interrupt, but it is very common for the thread to terminate.

Interrupted thread would throw InterruptedException

when a thread is waiting, sleeping, or otherwise paused for a long time and another thread interrupts it using the interrupt() method in class Thread.

So you need to explicitly code for scenarios such as you mentioned in executing thread where there is a possible interruption.