Fix racy use of ConcurrentHashMap

[DaveCTurner]

ConcurrentHashMap does not always behave correctly if removing elements and
concurrently checking for its emptyiness. Work around this in PeerFinder by
protecting all usages with a mutex (there was only one usage unprotected by the
mutex anyway) and then we don't even need a ConcurrentHashMap at all.

[elasticmachine]

Pinging @elastic/es-distributed

[DaveCTurner]

The racy use of ConcurrentHashMap is demonstrated by the failure of the following test:

public void testConcurrentMapIsEmptyOnConcurrentRemovals() throws InterruptedException {
    final Map<String, String> map = newConcurrentMap();
    final Thread[] threads = new Thread[2];
    final CyclicBarrier cyclicBarrier = new CyclicBarrier(threads.length, () -> map.put("foo", "bar"));

    for (int threadIndex = 0; threadIndex < threads.length; threadIndex++) {
        threads[threadIndex] = new Thread(() -> {
            for (int iteration = 0; iteration < 100000; iteration++) {
                try {
                    cyclicBarrier.await(1, TimeUnit.SECONDS);
                } catch (InterruptedException | BrokenBarrierException | TimeoutException e) {
                    throw new AssertionError("unexpected exception at " + iteration, e);
                }
                map.remove("foo");
                assertTrue("expected empty at " + iteration, map.isEmpty());
            }
        }, "remover-" + threadIndex);
    }

    for (Thread thread : threads) {
        thread.start();
    }

    for (Thread thread : threads) {
        thread.join();
    }
}

[ywelsch]

maybe simpler to just write

final boolean peersRemoved = peersByAddress.values().removeIf(Peer::handleWakeUp);

and then further below just

return peersRemoved;

[DaveCTurner]

Very slick. I pushed 96b02b0.

[ywelsch]

LGTM