@Test
    public void endOffsetsShouldReturnOffsetsForOnePartition() {
        String topicName = "myTopic";
        TopicPartition tp1 = new TopicPartition(topicName, 0);
        Set<TopicPartition> tps = Collections.singleton(tp1);
        long offset = 1000L;
        Cluster cluster = createCluster(1, topicName, 1);
        try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(new MockTime(), cluster)) {
            env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
            env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
            env.kafkaClient().prepareResponse(listOffsetsResult(tp1, offset));
            TopicAdmin admin = new TopicAdmin(env.adminClient());
            Map<TopicPartition, Long> offsets = admin.endOffsets(tps);
            assertEquals(1, offsets.size());
            assertEquals(Long.valueOf(offset), offsets.get(tp1));
        }
    }
TopicAdmin can be used to read the end offsets, but the admin client API used to do this was
added to the broker in 0.11.0.0. This means that if Connect talks to older brokers,
the admin client cannot be used to read end offsets, and will throw an UnsupportedVersionException.
connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java