public Map<TopicPartition, Long> retryEndOffsets(Set<TopicPartition> partitions, Duration timeoutDuration, long retryBackoffMs) {

        try {
            return RetryUtil.retryUntilTimeout(
                    () -> endOffsets(partitions),
                    () -> "list offsets for topic partitions",
                    timeoutDuration,
                    retryBackoffMs);
        } catch (UnsupportedVersionException e) {
            // Older brokers don't support this admin method, so rethrow it without wrapping it
            throw e;
        } catch (Exception e) {
            throw ConnectUtils.maybeWrap(e, "Failed to list offsets for topic partitions");
        }
    }
@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
            );
        }
    }