Untitled

Producer Consumer pattern - handling producer failure
// helper class that does the following
//
// if any thread has an exception then interrupt all the others with an eye to cancelling them
// if the thread calling execute() is interrupted then interrupt all the child threads
public class LinkedExecutor
{
    private final Collection<Runnable> runnables;
    private final String name;

    public LinkedExecutor( String name, Collection<Runnable> runnables )
    {
        this.runnables = runnables;
        this.name = name;
    }

    public void execute()
    {
        ExecutorService executorService = Executors.newCachedThreadPool( ConfigurableThreadFactory.newWithPrefix( name ) );

        // use a completion service to poll the results
        CompletionService<Object> completionService = new ExecutorCompletionService<Object>( executorService );

        for ( Runnable runnable : runnables )
        {
            completionService.submit( runnable, null );
        }

        try
        {
            for ( int i = 0; i < runnables.size(); i++ )
            {
                Future<?> future = completionService.take();
                future.get();
            }
        }
        catch ( InterruptedException e )
        {
            // on an interruption of this thread interrupt all sub-threads in the executor
            executorService.shutdownNow();

            throw new RuntimeException( "Executor '" + name + "' interrupted", e );
        }
        catch ( ExecutionException e )
        {
            // on a failure of any of the sub-threads interrupt all the threads
            executorService.shutdownNow();

            throw new RuntimeException( "Execution execution in executor '" + name + "'", e );
        }
    }
}