@Test
    public void testTopicIdUsageGrantedOnIdUpgrade() {
        // We want to test adding a topic ID to an existing partition and a new partition in the incremental request.
        // 0 is the existing partition and 1 is the new one.
        List<Integer> partitions = Arrays.asList(0, 1);
        partitions.forEach(partition -> {
            String testType = partition == 0 ? "updating a partition" : "adding a new partition";
            FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1);
            FetchSessionHandler.Builder builder = handler.newBuilder();
            builder.add(new TopicPartition("foo", 0),
                    new FetchRequest.PartitionData(Uuid.ZERO_UUID, 0, 100, 200, Optional.empty()));
            FetchSessionHandler.FetchRequestData data = builder.build();
            assertMapsEqual(reqMap(new ReqEntry("foo", Uuid.ZERO_UUID, 0, 0, 100, 200)),
                    data.toSend(), data.sessionPartitions());
            assertTrue(data.metadata().isFull());
            assertFalse(data.canUseTopicIds());

            FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123,
                    respMap(new RespEntry("foo", 0, Uuid.ZERO_UUID, 10, 20)));
            handler.handleResponse(resp, (short) 12);

            // Try to add a topic ID to an already existing topic partition (0) or a new partition (1) in the session.
            Uuid topicId = Uuid.randomUuid();
            FetchSessionHandler.Builder builder2 = handler.newBuilder();
            builder2.add(new TopicPartition("foo", partition),
                    new FetchRequest.PartitionData(topicId, 10, 110, 210, Optional.empty()));
            FetchSessionHandler.FetchRequestData data2 = builder2.build();
            // Should have the same session ID, and next epoch and can only use topic IDs if the partition was updated.
            boolean updated = partition == 0;
            // The receiving broker will handle closing the session.
            assertEquals(123, data2.metadata().sessionId(), "Did not use same session when " + testType);
            assertEquals(1, data2.metadata().epoch(), "Did not have correct epoch when " + testType);
            assertEquals(updated, data2.canUseTopicIds());
        });
    }
Test that calling FetchSessionHandler#Builder#build twice fails.
clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java