String verifyFullFetchResponsePartitions(Set<TopicPartition> topicPartitions, Set<Uuid> ids, short version) {
        StringBuilder bld = new StringBuilder();
        Set<TopicPartition> extra =
            findMissing(topicPartitions, sessionPartitions.keySet());
        Set<TopicPartition> omitted =
            findMissing(sessionPartitions.keySet(), topicPartitions);
        Set<Uuid> extraIds = new HashSet<>();
        if (version >= 13) {
            extraIds = findMissing(ids, sessionTopicNames.keySet());
        }
        if (!omitted.isEmpty()) {
            bld.append("omittedPartitions=(").append(omitted.stream().map(TopicPartition::toString).collect(Collectors.joining(", "))).append("), ");
        }
        if (!extra.isEmpty()) {
            bld.append("extraPartitions=(").append(extra.stream().map(TopicPartition::toString).collect(Collectors.joining(", "))).append("), ");
        }
        if (!extraIds.isEmpty()) {
            bld.append("extraIds=(").append(extraIds.stream().map(Uuid::toString).collect(Collectors.joining(", "))).append("), ");
        }
        if ((!omitted.isEmpty()) || (!extra.isEmpty()) || (!extraIds.isEmpty())) {
            bld.append("response=(").append(topicPartitions.stream().map(TopicPartition::toString).collect(Collectors.joining(", "))).append(")");
            return bld.toString();
        }
        return null;
    }
@Test
    public void testVerifyFullFetchResponsePartitions() {
        Map<String, Uuid> topicIds = new HashMap<>();
        Map<Uuid, String> topicNames = new HashMap<>();
        // We want to test both on older versions that do not use topic IDs and on newer versions that do.
        List<Short> versions = Arrays.asList((short) 12, ApiKeys.FETCH.latestVersion());
        versions.forEach(version -> {
            FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1);
            addTopicId(topicIds, topicNames, "foo", version);
            addTopicId(topicIds, topicNames, "bar", version);
            Uuid fooId = topicIds.getOrDefault("foo", Uuid.ZERO_UUID);
            Uuid barId = topicIds.getOrDefault("bar", Uuid.ZERO_UUID);
            TopicPartition foo0 = new TopicPartition("foo", 0);
            TopicPartition foo1 = new TopicPartition("foo", 1);
            TopicPartition bar0 = new TopicPartition("bar", 0);
            FetchResponse resp1 = FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID,
                respMap(new RespEntry("foo", 0, fooId, 10, 20),
                        new RespEntry("foo", 1, fooId, 10, 20),
                        new RespEntry("bar", 0, barId, 10, 20)));
            String issue = handler.verifyFullFetchResponsePartitions(resp1.responseData(topicNames, version).keySet(),
                    resp1.topicIds(), version);
            assertTrue(issue.contains("extraPartitions="));
            assertFalse(issue.contains("omittedPartitions="));
            FetchSessionHandler.Builder builder = handler.newBuilder();
            builder.add(foo0, new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty()));
            builder.add(foo1, new FetchRequest.PartitionData(fooId, 10, 110, 210, Optional.empty()));
            builder.add(bar0, new FetchRequest.PartitionData(barId, 20, 120, 220, Optional.empty()));
            builder.build();
            FetchResponse resp2 = FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID,
                respMap(new RespEntry("foo", 0, fooId, 10, 20),
                        new RespEntry("foo", 1, fooId, 10, 20),
                        new RespEntry("bar", 0, barId, 10, 20)));
            String issue2 = handler.verifyFullFetchResponsePartitions(resp2.responseData(topicNames, version).keySet(),
                    resp2.topicIds(), version);
            assertNull(issue2);
            FetchResponse resp3 = FetchResponse.of(Errors.NONE, 0, INVALID_SESSION_ID,
                respMap(new RespEntry("foo", 0, fooId, 10, 20),
                        new RespEntry("foo", 1, fooId, 10, 20)));
            String issue3 = handler.verifyFullFetchResponsePartitions(resp3.responseData(topicNames, version).keySet(),
                    resp3.topicIds(), version);
            assertFalse(issue3.contains("extraPartitions="));
            assertTrue(issue3.contains("omittedPartitions="));
        });
    }