private void testInFlightRequestsFailedAfterCoordinatorMarkedDead(Errors error) {
        client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
        coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));

        // Send two async commits and fail the first one with an error.
        // This should cause a coordinator disconnect which will cancel the second request.

        MockCommitCallback firstCommitCallback = new MockCommitCallback();
        MockCommitCallback secondCommitCallback = new MockCommitCallback();
        coordinator.commitOffsetsAsync(singletonMap(t1p, new OffsetAndMetadata(100L)), firstCommitCallback);
        coordinator.commitOffsetsAsync(singletonMap(t1p, new OffsetAndMetadata(100L)), secondCommitCallback);
        assertEquals(coordinator.inFlightAsyncCommits.get(), 2);

        respondToOffsetCommitRequest(singletonMap(t1p, 100L), error);
        consumerClient.pollNoWakeup();
        consumerClient.pollNoWakeup(); // second poll since coordinator disconnect is async
        coordinator.invokeCompletedOffsetCommitCallbacks();

        assertTrue(coordinator.coordinatorUnknown());
        assertInstanceOf(RetriableCommitFailedException.class, firstCommitCallback.exception);
        assertInstanceOf(RetriableCommitFailedException.class, secondCommitCallback.exception);
        assertEquals(coordinator.inFlightAsyncCommits.get(), 0);
    }
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