Q:

Timeout the thread if any bundle is taking lot of time

I am working on a project in which I will have different Bundles. Let's take an example, Suppose I have 5 bundles and each of those bundles will have a method name process.
Now currently, I am calling the process method of all those 5 bundles sequentially, one by one and then I am writing to the database. But that's what I don't want. 

I need to call all those 5 Bundles process method in parallel using multithread and then write to the database.
And I also want to have some timeout feature for those threads. I will be having a default timeout settings for all the threads for the bundles. If any bundle is taking some higher time than the timeout settings I have, then I want to timeout those threads and then log back saying this bundle got timeout bcoz it was taking lot of time.

I hope question is clear enough...
Below is the code I have so far which is calling process method sequentially one by one.
public void processEvents(final Map<String, Object> eventData) {

    final Map<String, String> outputs = (Map<String, String>)eventData.get(BConstants.EVENT_HOLDER);

    for (final BundleRegistration.BundlesHolderEntry entry : BundleRegistration.getInstance()) {

        final Map<String, String> response = entry.getPlugin().process(outputs);

        // write to the database.
        System.out.println(response);
    }
}

I am not sure what is the best and efficient way to do this? Because, in future, it might be possible that I will have more than 5 bundles.
Can anyone provide me an example of how can I achieve this? Any help will be appreciated on this. Thanks.

A:

It is not too difficult to achieve what you want, but you should be aware that both with concurrency and timeouts you get added complexity, especially when it comes to error handling. 
For instance, threads that are running when a timeout occurs may keep running after the timeout. Only well behaved threads that cooperate by handling an interrupt signal will be able to stop succefully in the middle of processing. 
You must also make sure that individual bundle entries may be processed in parallel, i.e. that the are thread safe. If they modify some shared resource while processing, then you might get strange errors as a result.
I was also wondering whether you wanted to include the database writing to each of these threads. If so, you will need to handle interruptions while writing to the database; e.g. by rolling back a transaction.
Anyways, to get thread pooling and a total timeout for all threads, you can use ExecutorService with (for instance) a fixed pool size and invoke all threads using the invokeAll method.
The following attempt is most probably flawed and error handling is by no means complete, but it should give you a starting point. 
First, an implementation of Callable for your threads:
public class ProcessBundleHolderEntry implements Callable {
    private BundleRegistration.BundlesHolderEntry entry;
    private Map<String, String> outputs;

    public ProcessBundleHolderEntry(BundleRegistration.BundlesHolderEntry entry, Map<String, String> outputs) {
        this.entry = entry;
        this.outputs = outputs;
    }

    public Object call() throws Exception {
        final Map<String, String> response = entry.getPlugin().process(outputs);
        // write to the database.
        System.out.println(response);
        return response;
    }
}

and now, the modified processEvents method:
public void processEvents(final Map<String, Object> eventData) {
    ExecutorService pool = Executors.newFixedThreadPool(5);
    List<ProcessBundleHolderEntry> entries = new ArrayList<ProcessBundleHolderEntry>();

    Map<String, String> outputs = (Map<String, String>)eventData.get(BConstants.EVENT_HOLDER);
    for (BundleRegistration.BundlesHolderEntry entry : BundleRegistration.getInstance()) {
        ProcessBundleHolderEntry processBundleHolderEntry = new ProcessBundleHolderEntry(entry, outputs);
        entries.add(processBundleHolderEntry);
    }

    try {
        List<Future<Object>> futures = pool.invokeAll(entries, 30, TimeUnit.SECONDS);
        for (int i = 0; i < futures.size(); i++) {
            // This works since the list of future objects are in the
            // same sequential order as the list of entries
            Future<Object> future = futures.get(i);
            ProcessBundleHolderEntry entry = entries.get(i);
            if (!future.isDone()) {
                // log error for this entry
            }
        }
    } catch (InterruptedException e) {
        // handle this exception!
    }
}

