@Test
    public void shouldLoseAllOwnedPartitionsBeforeRejoiningAfterDroppingOutOfTheGroup() {
        final List<TopicPartition> partitions = singletonList(t1p);
        try (ConsumerCoordinator coordinator = prepareCoordinatorForCloseTest(true, false, Optional.of("group-id"), true)) {
            final Time realTime = Time.SYSTEM;
            coordinator.ensureActiveGroup();

            prepareOffsetCommitRequest(singletonMap(t1p, 100L), Errors.REBALANCE_IN_PROGRESS);

            assertThrows(RebalanceInProgressException.class, () -> coordinator.commitOffsetsSync(
                singletonMap(t1p, new OffsetAndMetadata(100L)),
                time.timer(Long.MAX_VALUE)));

            int generationId = 42;
            String memberId = "consumer-42";

            client.prepareResponse(joinGroupFollowerResponse(generationId, memberId, "leader", Errors.NONE));
            client.prepareResponse(syncGroupResponse(Collections.emptyList(), Errors.UNKNOWN_MEMBER_ID));

            boolean res = coordinator.joinGroupIfNeeded(realTime.timer(1000));

            assertFalse(res);
            assertEquals(AbstractCoordinator.Generation.NO_GENERATION, coordinator.generation());
            assertEquals("", coordinator.generation().memberId);

            res = coordinator.joinGroupIfNeeded(realTime.timer(1000));
            assertFalse(res);
        }
        Collection<TopicPartition> lost = getLost(partitions);
        assertEquals(lost.isEmpty() ? 0 : 1, rebalanceListener.lostCount);
        assertEquals(lost.isEmpty() ? null : lost, rebalanceListener.lost);
    }
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