@ParameterizedTest
    @ValueSource(booleans = {true, false})
    public void testIdUsageWithAllForgottenPartitions(boolean useTopicIds) {
        // We want to test when all topics are removed from the session
        TopicPartition foo0 = new TopicPartition("foo", 0);
        Uuid topicId = useTopicIds ? Uuid.randomUuid() : Uuid.ZERO_UUID;
        short responseVersion = useTopicIds ? ApiKeys.FETCH.latestVersion() : 12;
        FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1);

        // Add topic foo to the session
        FetchSessionHandler.Builder builder = handler.newBuilder();
        builder.add(foo0, new FetchRequest.PartitionData(topicId, 0, 100, 200, Optional.empty()));
        FetchSessionHandler.FetchRequestData data = builder.build();
        assertMapsEqual(reqMap(new ReqEntry("foo", topicId, 0, 0, 100, 200)),
                data.toSend(), data.sessionPartitions());
        assertTrue(data.metadata().isFull());
        assertEquals(useTopicIds, data.canUseTopicIds());

        FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123,
                respMap(new RespEntry("foo", 0, topicId, 10, 20)));
        handler.handleResponse(resp, responseVersion);

        // Remove the topic from the session
        FetchSessionHandler.Builder builder2 = handler.newBuilder();
        FetchSessionHandler.FetchRequestData data2 = builder2.build();
        assertEquals(Collections.singletonList(new TopicIdPartition(topicId, foo0)), data2.toForget());
        // Should have the same session ID, next epoch, and same ID usage.
        assertEquals(123, data2.metadata().sessionId(), "Did not use same session when useTopicIds was " + useTopicIds);
        assertEquals(1, data2.metadata().epoch(), "Did not have correct epoch when useTopicIds was " + useTopicIds);
        assertEquals(useTopicIds, data2.canUseTopicIds());
    }
Test that calling FetchSessionHandler#Builder#build twice fails.
clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java