long trySend(long now) {
        long pollDelayMs = maxPollTimeoutMs;

        // send any requests that can be sent now
        for (Node node : unsent.nodes()) {
            Iterator<ClientRequest> iterator = unsent.requestIterator(node);
            if (iterator.hasNext())
                pollDelayMs = Math.min(pollDelayMs, client.pollDelayMs(node, now));

            while (iterator.hasNext()) {
                ClientRequest request = iterator.next();
                if (client.ready(node, now)) {
                    client.send(request, now);
                    iterator.remove();
                } else {
                    // try next node when current node is not ready
                    break;
                }
            }
        }
        return pollDelayMs;
    }
@Test
    public void testTrySend() {
        final AtomicBoolean isReady = new AtomicBoolean();
        final AtomicInteger checkCount = new AtomicInteger();
        client = new MockClient(time, metadata) {
            @Override
            public boolean ready(Node node, long now) {
                checkCount.incrementAndGet();
                if (isReady.get())
                    return super.ready(node, now);
                else
                    return false;
            }
        };
        consumerClient = new ConsumerNetworkClient(new LogContext(), client, metadata, time, 100, 10, Integer.MAX_VALUE);
        consumerClient.send(node, heartbeat());
        consumerClient.send(node, heartbeat());
        assertEquals(2, consumerClient.pendingRequestCount(node));
        assertEquals(0, client.inFlightRequestCount(node.idString()));

        consumerClient.trySend(time.milliseconds());
        // only check one time when the node doesn't ready
        assertEquals(1, checkCount.getAndSet(0));
        assertEquals(2, consumerClient.pendingRequestCount(node));
        assertEquals(0, client.inFlightRequestCount(node.idString()));

        isReady.set(true);
        consumerClient.trySend(time.milliseconds());
        // check node ready or not for every request
        assertEquals(2, checkCount.getAndSet(0));
        assertEquals(2, consumerClient.pendingRequestCount(node));
        assertEquals(2, client.inFlightRequestCount(node.idString()));
    }