@Test
    public void retryEndOffsetsShouldWrapNonRetriableExceptionsWithConnectException() {
        String topicName = "myTopic";
        TopicPartition tp1 = new TopicPartition(topicName, 0);
        Set<TopicPartition> tps = Collections.singleton(tp1);
        Long offset = 1000L;
        Cluster cluster = createCluster(1, "myTopic", 1);

        try (final AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(new MockTime(), cluster)) {
            Map<TopicPartition, Long> offsetMap = new HashMap<>();
            offsetMap.put(tp1, offset);
            env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());

            // This error should be treated as non-retriable and cause TopicAdmin::retryEndOffsets to fail
            env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.TOPIC_AUTHORIZATION_FAILED, Errors.NONE));
            // But, in case there's a bug in our logic, prepare a valid response afterward so that TopicAdmin::retryEndOffsets
            // will return successfully if we retry (which should in turn cause this test to fail)
            env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
            env.kafkaClient().prepareResponse(listOffsetsResult(tp1, offset));

            TopicAdmin admin = new TopicAdmin(env.adminClient());
            ConnectException exception = assertThrows(ConnectException.class, () ->
                admin.retryEndOffsets(tps, Duration.ofMillis(100), 1)
            );

            Throwable cause = exception.getCause();
            assertNotNull(cause, "cause of failure should be preserved");
            assertTrue(
                    cause instanceof TopicAuthorizationException,
                    "cause of failure should be accurately reported; expected topic authorization error, but was " + cause
            );
        }
    }
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