@Test
    public void testRefreshOffsetWithValidation() {
        client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
        coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));

        subscriptions.assignFromUser(singleton(t1p));

        // Initial leader epoch of 4
        MetadataResponse metadataResponse = RequestTestUtils.metadataUpdateWith("kafka-cluster", 1,
                Collections.emptyMap(), singletonMap(topic1, 1), tp -> 4);
        client.updateMetadata(metadataResponse);

        // Load offsets from previous epoch
        client.prepareResponse(offsetFetchResponse(t1p, Errors.NONE, "", 100L, Optional.of(3)));
        coordinator.initWithCommittedOffsetsIfNeeded(time.timer(Long.MAX_VALUE));

        // Offset gets loaded, but requires validation
        assertEquals(Collections.emptySet(), subscriptions.initializingPartitions());
        assertFalse(subscriptions.hasAllFetchPositions());
        assertTrue(subscriptions.awaitingValidation(t1p));
        assertEquals(subscriptions.position(t1p).offset, 100L);
        assertNull(subscriptions.validPosition(t1p));
    }
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