@ParameterizedTest
    @ValueSource(booleans = {true, false})
    public void testSessionEpochWhenMixedUsageOfTopicIDs(boolean startsWithTopicIds) {
        Uuid fooId = startsWithTopicIds ? Uuid.randomUuid() : Uuid.ZERO_UUID;
        Uuid barId = startsWithTopicIds ? Uuid.ZERO_UUID : Uuid.randomUuid();
        short responseVersion = startsWithTopicIds ? ApiKeys.FETCH.latestVersion() : 12;

        TopicPartition tp0 = new TopicPartition("foo", 0);
        TopicPartition tp1 = new TopicPartition("bar", 1);

        FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1);
        FetchSessionHandler.Builder builder = handler.newBuilder();
        builder.add(tp0, new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty()));
        FetchSessionHandler.FetchRequestData data = builder.build();
        assertMapsEqual(reqMap(new ReqEntry("foo", fooId, 0, 0, 100, 200)),
                data.toSend(), data.sessionPartitions());
        assertTrue(data.metadata().isFull());
        assertEquals(startsWithTopicIds, data.canUseTopicIds());

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

        // Re-add the first partition. Then add a partition with opposite ID usage.
        FetchSessionHandler.Builder builder2 = handler.newBuilder();
        builder2.add(tp0, new FetchRequest.PartitionData(fooId, 10, 110, 210, Optional.empty()));
        builder2.add(tp1, new FetchRequest.PartitionData(barId, 0, 100, 200, Optional.empty()));
        FetchSessionHandler.FetchRequestData data2 = builder2.build();
        // Should have the same session ID, and the next epoch and can not use topic IDs.
        // The receiving broker will handle closing the session.
        assertEquals(123, data2.metadata().sessionId(), "Did not use same session");
        assertEquals(1, data2.metadata().epoch(), "Did not have final epoch");
        assertFalse(data2.canUseTopicIds());
    }
Test that calling FetchSessionHandler#Builder#build twice fails.
clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java