public void maybeAutoCommitOffsetsAsync(long now) {
        if (autoCommitEnabled) {
            nextAutoCommitTimer.update(now);
            if (nextAutoCommitTimer.isExpired()) {
                nextAutoCommitTimer.reset(autoCommitIntervalMs);
                autoCommitOffsetsAsync();
            }
        }
    }
@Test
    public void testAutoCommitAfterCoordinatorBackToService() {
        try (ConsumerCoordinator coordinator = buildCoordinator(rebalanceConfig, new Metrics(), assignors, true, subscriptions)) {
            subscriptions.assignFromUser(Collections.singleton(t1p));
            subscriptions.seek(t1p, 100L);

            coordinator.markCoordinatorUnknown("test cause");
            assertTrue(coordinator.coordinatorUnknown());
            client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
            prepareOffsetCommitRequest(singletonMap(t1p, 100L), Errors.NONE);

            // async commit offset should find coordinator
            time.sleep(autoCommitIntervalMs); // sleep for a while to ensure auto commit does happen
            coordinator.maybeAutoCommitOffsetsAsync(time.milliseconds());
            assertFalse(coordinator.coordinatorUnknown());
            assertEquals(100L, subscriptions.position(t1p).offset);
        }
    }