@Test
    public void testOkToAddNewIdAfterTopicRemovedFromSession() {
        Uuid topicId = Uuid.randomUuid();
        FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1);
        FetchSessionHandler.Builder builder = handler.newBuilder();
        builder.add(new TopicPartition("foo", 0),
                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());
        assertTrue(data.canUseTopicIds());

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

        // Remove the partition from the session. Return a session ID as though the session is still open.
        FetchSessionHandler.Builder builder2 = handler.newBuilder();
        FetchSessionHandler.FetchRequestData data2 = builder2.build();
        assertMapsEqual(new LinkedHashMap<>(),
                data2.toSend(), data2.sessionPartitions());
        FetchResponse resp2 = FetchResponse.of(Errors.NONE, 0, 123,
                new LinkedHashMap<>());
        handler.handleResponse(resp2, ApiKeys.FETCH.latestVersion());

        // After the topic is removed, add a recreated topic with a new ID.
        FetchSessionHandler.Builder builder3 = handler.newBuilder();
        builder3.add(new TopicPartition("foo", 0),
                new FetchRequest.PartitionData(Uuid.randomUuid(), 0, 100, 200, Optional.empty()));
        FetchSessionHandler.FetchRequestData data3 = builder3.build();
        // Should have the same session ID and epoch 2.
        assertEquals(123, data3.metadata().sessionId(), "Did not use same session");
        assertEquals(2, data3.metadata().epoch(), "Did not have the correct session epoch");
        assertTrue(data.canUseTopicIds());
    }
Test that calling FetchSessionHandler#Builder#build twice fails.
clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java