protected void handleTaskSubmittedRequest(Runnable runnable, Address source, 
                                              long requestId, long threadId) {
        // We store in our map so that when that task is
        // finished so that we can send back to the owner
        // with the results
        _running.put(runnable, new Owner(source, requestId));
        // We give the task to the thread that is now waiting for it to be returned
        // If we can't offer then we have to respond back to
        // caller that we can't handle it.  They must have
        // gotten our address when we had a consumer, but
        // they went away between then and now.
        boolean received;
        try {
            _tasks.put(threadId, runnable);
            
            CyclicBarrier barrier = _taskBarriers.remove(threadId);
            if (received = (barrier != null)) {
                // Only wait 10 milliseconds, in case if the consumer was
                // stopped between when we were told it was available and now
                barrier.await(10, TimeUnit.MILLISECONDS);
            }
        }
        catch (InterruptedException e) {
            if (log.isDebugEnabled())
                log.debug("Interrupted while handing off task");
            Thread.currentThread().interrupt();
            received = false;
        }
        catch (BrokenBarrierException e) {
            if (log.isDebugEnabled())
                log.debug("Consumer " + threadId + " has been interrupted, " +
                        "must retry to submit elsewhere");
            received = false;
        }
        catch (TimeoutException e) {
            if (log.isDebugEnabled())
                log.debug("Timeout waiting to hand off to barrier, consumer " + 
                        threadId + " must be slow");
            // This should only happen if the consumer put the latch then got
            // interrupted but hadn't yet removed the latch, should almost never
            // happen
            received = false;
        }
        
        if (!received) {
            // Clean up the tasks request
            _tasks.remove(threadId);
            if (log.isDebugEnabled())
                log.debug("Run rejected not able to pass off to consumer");
            // If we couldn't hand off the task we have to tell the client
            // and also reupdate the coordinator that our consumer is ready
            sendRequest(source, Type.RUN_REJECTED, requestId, null);
            _running.remove(runnable);
        }
    }