KAFKA-13770: Restore compatibility with KafkaBasedLog using older Kafka brokers

connect/runtime/src/main/java/org/apache/kafka/connect/util/TopicAdmin.java

@@ -714,6 +714,7 @@ public Map<TopicPartition, Long> endOffsets(Set<TopicPartition> partitions) {
      *                          must be 0 or more
      * @return                  the map of offset for each topic partition, or an empty map if the supplied partitions
      *                          are null or empty
+     * @throws UnsupportedVersionException if the broker is too old to support the admin client API to read end offsets
      * @throws ConnectException if {@code timeoutDuration} is exhausted
      * @see TopicAdmin#endOffsets(Set)
      */
@@ -725,6 +726,9 @@ public Map<TopicPartition, Long> retryEndOffsets(Set<TopicPartition> 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 new ConnectException("Failed to list offsets for topic partitions.", e);
         }

connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java

@@ -467,8 +467,31 @@ public void verifyingGettingTopicCleanupPolicies() {
         }
     }
 
+    /**
+     * 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.
+     */
+    @Test
+    public void retryEndOffsetsShouldRethrowUnknownVersionException() {
+        String topicName = "myTopic";
+        TopicPartition tp1 = new TopicPartition(topicName, 0);
+        Set<TopicPartition> tps = Collections.singleton(tp1);
+        Long offset = null; // response should use error
+        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));
+            // Expect the admin client list offsets will throw unsupported version, simulating older brokers
+            env.kafkaClient().prepareResponse(listOffsetsResultWithUnsupportedVersion(tp1, offset));
+            TopicAdmin admin = new TopicAdmin(null, env.adminClient());
+            // The retryEndOffsets should catch and rethrow an unsupported version exception
+            assertThrows(UnsupportedVersionException.class, () -> admin.retryEndOffsets(tps, Duration.ofMillis(100), 1));
+        }
+    }
+
     @Test
-    public void retryEndOffsetsShouldThrowConnectException() {
+    public void retryEndOffsetsShouldWrapNonRetriableExceptionsWithConnectException() {
         String topicName = "myTopic";
         TopicPartition tp1 = new TopicPartition(topicName, 0);
         Set<TopicPartition> tps = Collections.singleton(tp1);