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 retryEndOffsetsShouldRetryWhenTopicNotFound() {
        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 (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(new MockTime(), cluster)) {
            Map<TopicPartition, Long> offsetMap = new HashMap<>();
            offsetMap.put(tp1, offset);
            env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
            env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.UNKNOWN_TOPIC_OR_PARTITION));
            env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
            env.kafkaClient().prepareResponse(listOffsetsResult(tp1, offset));

            TopicAdmin admin = new TopicAdmin(env.adminClient());
            Map<TopicPartition, Long> endoffsets = admin.retryEndOffsets(tps, Duration.ofMillis(100), 1);
            assertEquals(Collections.singletonMap(tp1, offset), endoffsets);
        }
    }