@Test
    public void testVerifyFullFetchResponsePartitionsWithTopicIds() {
        Map<String, Uuid> topicIds = new HashMap<>();
        Map<Uuid, String> topicNames = new HashMap<>();
        FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1);
        addTopicId(topicIds, topicNames, "foo", ApiKeys.FETCH.latestVersion());
        addTopicId(topicIds, topicNames, "bar", ApiKeys.FETCH.latestVersion());
        addTopicId(topicIds, topicNames, "extra2", ApiKeys.FETCH.latestVersion());
        FetchResponse resp1 = FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID,
            respMap(new RespEntry("foo", 0, topicIds.get("foo"), 10, 20),
                    new RespEntry("extra2", 1, topicIds.get("extra2"), 10, 20),
                    new RespEntry("bar", 0, topicIds.get("bar"), 10, 20)));
        String issue = handler.verifyFullFetchResponsePartitions(resp1.responseData(topicNames, ApiKeys.FETCH.latestVersion()).keySet(),
                resp1.topicIds(), ApiKeys.FETCH.latestVersion());
        assertTrue(issue.contains("extraPartitions="));
        assertFalse(issue.contains("omittedPartitions="));
        FetchSessionHandler.Builder builder = handler.newBuilder();
        builder.add(new TopicPartition("foo", 0),
                new FetchRequest.PartitionData(topicIds.get("foo"), 0, 100, 200, Optional.empty()));
        builder.add(new TopicPartition("bar", 0),
                new FetchRequest.PartitionData(topicIds.get("bar"), 20, 120, 220, Optional.empty()));
        builder.build();
        FetchResponse resp2 = FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID,
            respMap(new RespEntry("foo", 0, topicIds.get("foo"), 10, 20),
                    new RespEntry("extra2", 1, topicIds.get("extra2"), 10, 20),
                    new RespEntry("bar", 0, topicIds.get("bar"), 10, 20)));
        String issue2 = handler.verifyFullFetchResponsePartitions(resp2.responseData(topicNames, ApiKeys.FETCH.latestVersion()).keySet(),
                resp2.topicIds(), ApiKeys.FETCH.latestVersion());
        assertTrue(issue2.contains("extraPartitions="));
        assertFalse(issue2.contains("omittedPartitions="));
        FetchResponse resp3 = FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID,
            respMap(new RespEntry("foo", 0, topicIds.get("foo"), 10, 20),
                    new RespEntry("bar", 0, topicIds.get("bar"), 10, 20)));
        String issue3 = handler.verifyFullFetchResponsePartitions(resp3.responseData(topicNames, ApiKeys.FETCH.latestVersion()).keySet(),
                resp3.topicIds(), ApiKeys.FETCH.latestVersion());
        assertNull(issue3);
    }
Test that calling FetchSessionHandler#Builder#build twice fails.
clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java