public void markCoordinatorUnknown(final String cause, final long currentTimeMs) {
        if (this.coordinator != null) {
            log.info("Group coordinator {} is unavailable or invalid due to cause: {}. "
                    + "Rediscovery will be attempted.", this.coordinator, cause);
            this.coordinator = null;
            timeMarkedUnknownMs = currentTimeMs;
            totalDisconnectedMin = 0;
        } else {
            long durationOfOngoingDisconnectMs = Math.max(0, currentTimeMs - timeMarkedUnknownMs);
            long currDisconnectMin = durationOfOngoingDisconnectMs / COORDINATOR_DISCONNECT_LOGGING_INTERVAL_MS;
            if (currDisconnectMin > this.totalDisconnectedMin) {
                log.debug("Consumer has been disconnected from the group coordinator for {}ms", durationOfOngoingDisconnectMs);
                totalDisconnectedMin = currDisconnectMin;
            }
        }
    }
@Test
    public void testMarkCoordinatorUnknown() {
        CoordinatorRequestManager coordinatorManager = setupCoordinatorManager(GROUP_ID);

        expectFindCoordinatorRequest(coordinatorManager, Errors.NONE);
        assertTrue(coordinatorManager.coordinator().isPresent());

        // It may take time for metadata to converge between after a coordinator has
        // been demoted. This can cause a tight loop in which FindCoordinator continues to
        // return node X while that node continues to reply with NOT_COORDINATOR. Hence we
        // still want to ensure a backoff after successfully finding the coordinator.
        coordinatorManager.markCoordinatorUnknown("coordinator changed", time.milliseconds());
        assertEquals(Collections.emptyList(), coordinatorManager.poll(time.milliseconds()).unsentRequests);

        time.sleep(RETRY_BACKOFF_MS - 1);
        assertEquals(Collections.emptyList(), coordinatorManager.poll(time.milliseconds()).unsentRequests);

        time.sleep(RETRY_BACKOFF_MS);
        expectFindCoordinatorRequest(coordinatorManager, Errors.NONE);
        assertTrue(coordinatorManager.coordinator().isPresent());
    }