@Test
    public void testAutoCommitAwaitsInterval() {
        try (ConsumerCoordinator coordinator = buildCoordinator(rebalanceConfig, new Metrics(), assignors, true, subscriptions)) {
            subscriptions.subscribe(singleton(topic1), Optional.of(rebalanceListener));
            joinAsFollowerAndReceiveAssignment(coordinator, singletonList(t1p));

            subscriptions.seek(t1p, 100);
            time.sleep(autoCommitIntervalMs);

            // Send the offset commit request, but do not respond
            coordinator.poll(time.timer(Long.MAX_VALUE));
            assertEquals(1, client.inFlightRequestCount());

            time.sleep(autoCommitIntervalMs / 2);

            // Ensure that no additional offset commit is sent
            coordinator.poll(time.timer(Long.MAX_VALUE));
            assertEquals(1, client.inFlightRequestCount());

            respondToOffsetCommitRequest(singletonMap(t1p, 100L), Errors.NONE);
            coordinator.poll(time.timer(Long.MAX_VALUE));
            assertEquals(0, client.inFlightRequestCount());

            subscriptions.seek(t1p, 200);

            // If we poll again before the auto-commit interval, there should be no new sends
            coordinator.poll(time.timer(Long.MAX_VALUE));
            assertEquals(0, client.inFlightRequestCount());

            // After the remainder of the interval passes, we send a new offset commit
            time.sleep(autoCommitIntervalMs / 2);
            coordinator.poll(time.timer(Long.MAX_VALUE));
            assertEquals(1, client.inFlightRequestCount());
            respondToOffsetCommitRequest(singletonMap(t1p, 200L), Errors.NONE);
        }
    }
Verifies that subscription change updates SubscriptionState correctly even after JoinGroup failures
that don't re-invoke onJoinPrepare.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java