### Question:
ConsumerNetworkClient implements Closeable { public void poll(RequestFuture<?> future) { while (!future.isDone()) poll(MAX_POLL_TIMEOUT_MS, time.milliseconds(), future); } ConsumerNetworkClient(KafkaClient client,
                                 Metadata metadata,
                                 Time time,
                                 long retryBackoffMs,
                                 long requestTimeoutMs); RequestFuture<ClientResponse> send(Node node, AbstractRequest.Builder<?> requestBuilder); synchronized Node leastLoadedNode(); synchronized boolean hasReadyNodes(); void awaitMetadataUpdate(); boolean awaitMetadataUpdate(long timeout); void ensureFreshMetadata(); void wakeup(); void poll(RequestFuture<?> future); boolean poll(RequestFuture<?> future, long timeout); void poll(long timeout); void poll(long timeout, long now, PollCondition pollCondition); void poll(long timeout, long now, PollCondition pollCondition, boolean disableWakeup); void pollNoWakeup(); boolean awaitPendingRequests(Node node, long timeoutMs); int pendingRequestCount(Node node); boolean hasPendingRequests(Node node); int pendingRequestCount(); boolean hasPendingRequests(); void failUnsentRequests(Node node, RuntimeException e); void maybeTriggerWakeup(); void disableWakeups(); @Override void close(); boolean connectionFailed(Node node); void tryConnect(Node node);  }### Answer:
@Test public void doNotBlockIfPollConditionIsSatisfied() { NetworkClient mockNetworkClient = EasyMock.mock(NetworkClient.class); ConsumerNetworkClient consumerClient = new ConsumerNetworkClient(mockNetworkClient, metadata, time, 100, 1000); EasyMock.expect(mockNetworkClient.poll(EasyMock.eq(0L), EasyMock.anyLong())).andReturn(Collections.<ClientResponse>emptyList()); EasyMock.replay(mockNetworkClient); consumerClient.poll(Long.MAX_VALUE, time.milliseconds(), new ConsumerNetworkClient.PollCondition() { @Override public boolean shouldBlock() { return false; } }); EasyMock.verify(mockNetworkClient); }
@Test public void blockWhenPollConditionNotSatisfied() { long timeout = 4000L; NetworkClient mockNetworkClient = EasyMock.mock(NetworkClient.class); ConsumerNetworkClient consumerClient = new ConsumerNetworkClient(mockNetworkClient, metadata, time, 100, 1000); EasyMock.expect(mockNetworkClient.inFlightRequestCount()).andReturn(1); EasyMock.expect(mockNetworkClient.poll(EasyMock.eq(timeout), EasyMock.anyLong())).andReturn(Collections.<ClientResponse>emptyList()); EasyMock.replay(mockNetworkClient); consumerClient.poll(timeout, time.milliseconds(), new ConsumerNetworkClient.PollCondition() { @Override public boolean shouldBlock() { return true; } }); EasyMock.verify(mockNetworkClient); }
@Test public void blockOnlyForRetryBackoffIfNoInflightRequests() { long retryBackoffMs = 100L; NetworkClient mockNetworkClient = EasyMock.mock(NetworkClient.class); ConsumerNetworkClient consumerClient = new ConsumerNetworkClient(mockNetworkClient, metadata, time, retryBackoffMs, 1000L); EasyMock.expect(mockNetworkClient.inFlightRequestCount()).andReturn(0); EasyMock.expect(mockNetworkClient.poll(EasyMock.eq(retryBackoffMs), EasyMock.anyLong())).andReturn(Collections.<ClientResponse>emptyList()); EasyMock.replay(mockNetworkClient); consumerClient.poll(Long.MAX_VALUE, time.milliseconds(), new ConsumerNetworkClient.PollCondition() { @Override public boolean shouldBlock() { return true; } }); EasyMock.verify(mockNetworkClient); }