Optional<Integer> findOneStaleBroker() {
        BrokerHeartbeatStateIterator iterator = unfenced.iterator();
        if (iterator.hasNext()) {
            BrokerHeartbeatState broker = iterator.next();
            // The unfenced list is sorted on last contact time from each
            // broker. If the first broker is not stale, then none is.
            if (!hasValidSession(broker)) {
                return Optional.of(broker.id);
            }
        }
        return Optional.empty();
    }
@Test
    public void testFindOneStaleBroker() {
        BrokerHeartbeatManager manager = newBrokerHeartbeatManager();
        MockTime time = (MockTime)  manager.time();
        assertFalse(manager.hasValidSession(0));
        for (int brokerId = 0; brokerId < 3; brokerId++) {
            manager.register(brokerId, true);
        }
        manager.touch(0, false, 0);
        time.sleep(5);
        manager.touch(1, false, 0);
        time.sleep(1);
        manager.touch(2, false, 0);

        Iterator<BrokerHeartbeatState> iter = manager.unfenced().iterator();
        assertEquals(0, iter.next().id());
        assertEquals(1, iter.next().id());
        assertEquals(2, iter.next().id());
        assertFalse(iter.hasNext());
        assertEquals(Optional.empty(), manager.findOneStaleBroker());

        time.sleep(5);
        assertEquals(Optional.of(0), manager.findOneStaleBroker());
        manager.fence(0);
        assertEquals(Optional.empty(), manager.findOneStaleBroker());
        iter = manager.unfenced().iterator();
        assertEquals(1, iter.next().id());
        assertEquals(2, iter.next().id());
        assertFalse(iter.hasNext());

        time.sleep(20);
        assertEquals(Optional.of(1), manager.findOneStaleBroker());
        manager.fence(1);
        assertEquals(Optional.of(2), manager.findOneStaleBroker());
        manager.fence(2);

        assertEquals(Optional.empty(), manager.findOneStaleBroker());
        iter = manager.unfenced().iterator();
        assertFalse(iter.hasNext());
    }