@Test
    public void testTopLevelErrorResetsMetadata() {
        Map<String, Uuid> topicIds = new HashMap<>();
        Map<Uuid, String> topicNames = new HashMap<>();
        FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1);
        FetchSessionHandler.Builder builder = handler.newBuilder();
        addTopicId(topicIds, topicNames, "foo", ApiKeys.FETCH.latestVersion());
        Uuid fooId = topicIds.getOrDefault("foo", Uuid.ZERO_UUID);
        builder.add(new TopicPartition("foo", 0),
                new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty()));
        builder.add(new TopicPartition("foo", 1),
                new FetchRequest.PartitionData(fooId, 10, 110, 210, Optional.empty()));
        FetchSessionHandler.FetchRequestData data = builder.build();
        assertEquals(INVALID_SESSION_ID, data.metadata().sessionId());
        assertEquals(INITIAL_EPOCH, data.metadata().epoch());

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

        // Test an incremental fetch request which adds an ID unknown to the broker.
        FetchSessionHandler.Builder builder2 = handler.newBuilder();
        addTopicId(topicIds, topicNames, "unknown", ApiKeys.FETCH.latestVersion());
        builder2.add(new TopicPartition("unknown", 0),
                new FetchRequest.PartitionData(topicIds.getOrDefault("unknown", Uuid.ZERO_UUID), 0, 100, 200, Optional.empty()));
        FetchSessionHandler.FetchRequestData data2 = builder2.build();
        assertFalse(data2.metadata().isFull());
        assertEquals(123, data2.metadata().sessionId());
        assertEquals(FetchMetadata.nextEpoch(INITIAL_EPOCH), data2.metadata().epoch());

        // Return and handle a response with a top level error
        FetchResponse resp2 = FetchResponse.of(Errors.UNKNOWN_TOPIC_ID, 0, 123,
            respMap(new RespEntry("unknown", 0, Uuid.randomUuid(), Errors.UNKNOWN_TOPIC_ID)));
        assertFalse(handler.handleResponse(resp2, ApiKeys.FETCH.latestVersion()));

        // Ensure we start with a new epoch. This will close the session in the next request.
        FetchSessionHandler.Builder builder3 = handler.newBuilder();
        FetchSessionHandler.FetchRequestData data3 = builder3.build();
        assertEquals(123, data3.metadata().sessionId());
        assertEquals(INITIAL_EPOCH, data3.metadata().epoch());
    }
Test that calling FetchSessionHandler#Builder#build twice fails.
clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java