AbstractCoordinator implements Closeable { public void ensureCoordinatorReady() { while (coordinatorUnknown()) { RequestFuture<Void> future = sendGroupCoordinatorRequest(); client.poll(future); if (future.failed()) { if (future.isRetriable()) { client.awaitMetadataUpdate(); } else { throw future.exception(); } } else if (coordinator != null && client.connectionFailed(coordinator)) { coordinatorDead(); time.sleep(retryBackoffMs); } } } AbstractCoordinator(ConsumerNetworkClient client,
        String groupId,
        int sessionTimeoutMs,
        int heartbeatIntervalMs,
        Metrics metrics,
        String metricGrpPrefix,
        Time time,
        long retryBackoffMs); void ensureCoordinatorReady(); void ensureActiveGroup(); boolean coordinatorUnknown(); @Override void close(); void maybeLeaveGroup(); RequestFuture<Void> sendHeartbeatRequest();  }
@Test public void testCoordinatorDiscoveryBackoff() { mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); mockClient.prepareResponse(groupCoordinatorResponse(node, Errors.NONE.code())); mockClient.blackout(coordinatorNode, 50L); long initialTime = mockTime.milliseconds(); coordinator.ensureCoordinatorReady(); long endTime = mockTime.milliseconds(); assertTrue(endTime - initialTime >= RETRY_BACKOFF_MS); }