KAFKA-12879: Revert changes from KAFKA-12339 and instead add retry capability to KafkaBasedLog

clients/src/main/java/org/apache/kafka/clients/admin/internals/MetadataOperationContext.java

@@ -82,7 +82,6 @@ public Collection<String> topics() {
 
     public static void handleMetadataErrors(MetadataResponse response) {
         for (TopicMetadata tm : response.topicMetadata()) {
-            if (shouldRefreshMetadata(tm.error())) throw tm.error().exception();
             for (PartitionMetadata pm : tm.partitionMetadata()) {
                 if (shouldRefreshMetadata(pm.error)) {
                     throw pm.error.exception();

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

@@ -489,16 +489,12 @@ private static FindCoordinatorResponse prepareOldFindCoordinatorResponse(Errors
     }
 
     private static MetadataResponse prepareMetadataResponse(Cluster cluster, Errors error) {
-        return prepareMetadataResponse(cluster, error, error);
-    }
-
-    private static MetadataResponse prepareMetadataResponse(Cluster cluster, Errors topicError, Errors partitionError) {
         List<MetadataResponseTopic> metadata = new ArrayList<>();
         for (String topic : cluster.topics()) {
             List<MetadataResponsePartition> pms = new ArrayList<>();
             for (PartitionInfo pInfo : cluster.availablePartitionsForTopic(topic)) {
                 MetadataResponsePartition pm  = new MetadataResponsePartition()
-                    .setErrorCode(partitionError.code())
+                    .setErrorCode(error.code())
                     .setPartitionIndex(pInfo.partition())
                     .setLeaderId(pInfo.leader().id())
                     .setLeaderEpoch(234)
@@ -508,7 +504,7 @@ private static MetadataResponse prepareMetadataResponse(Cluster cluster, Errors
                 pms.add(pm);
             }
             MetadataResponseTopic tm = new MetadataResponseTopic()
-                .setErrorCode(topicError.code())
+                .setErrorCode(error.code())
                 .setName(topic)
                 .setIsInternal(false)
                 .setPartitions(pms);
@@ -4339,40 +4335,6 @@ public void testListOffsets() throws Exception {
         }
     }
 
-    @Test
-    public void testListOffsetsRetriableErrorOnMetadata() throws Exception {
-        Node node = new Node(0, "localhost", 8120);
-        List<Node> nodes = Collections.singletonList(node);
-        final Cluster cluster = new Cluster(
-            "mockClusterId",
-            nodes,
-            Collections.singleton(new PartitionInfo("foo", 0, node, new Node[]{node}, new Node[]{node})),
-            Collections.emptySet(),
-            Collections.emptySet(),
-            node);
-        final TopicPartition tp0 = new TopicPartition("foo", 0);
-
-        try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster)) {
-            env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
-            env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.UNKNOWN_TOPIC_OR_PARTITION, Errors.NONE));
-            // metadata refresh because of UNKNOWN_TOPIC_OR_PARTITION
-            env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));
-            // listoffsets response from broker 0
-            ListOffsetsResponseData responseData = new ListOffsetsResponseData()
-                .setThrottleTimeMs(0)
-                .setTopics(Collections.singletonList(ListOffsetsResponse.singletonListOffsetsTopicResponse(tp0, Errors.NONE, -1L, 123L, 321)));
-            env.kafkaClient().prepareResponseFrom(new ListOffsetsResponse(responseData), node);
-
-            ListOffsetsResult result = env.adminClient().listOffsets(Collections.singletonMap(tp0, OffsetSpec.latest()));
-
-            Map<TopicPartition, ListOffsetsResultInfo> offsets = result.all().get(3, TimeUnit.SECONDS);
-            assertEquals(1, offsets.size());
-            assertEquals(123L, offsets.get(tp0).offset());
-            assertEquals(321, offsets.get(tp0).leaderEpoch().get().intValue());
-            assertEquals(-1L, offsets.get(tp0).timestamp());
-        }
-    }
-
     @Test
     public void testListOffsetsRetriableErrors() throws Exception {
 

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

@@ -76,6 +76,10 @@ public class KafkaBasedLog<K, V> {
     private static final Logger log = LoggerFactory.getLogger(KafkaBasedLog.class);
     private static final long CREATE_TOPIC_TIMEOUT_NS = TimeUnit.SECONDS.toNanos(30);
     private static final long MAX_SLEEP_MS = TimeUnit.SECONDS.toMillis(1);
+    // 15min of admin retry duration to ensure successful metadata propagation.  10 seconds of backoff
+    // in between retries
+    private static final Duration ADMIN_CLIENT_RETRY_DURATION = Duration.ofMinutes(15);
+    private static final long ADMIN_CLIENT_RETRY_BACKOFF_MS = TimeUnit.SECONDS.toMillis(10);
 
     private Time time;
     private final String topic;
@@ -194,7 +198,7 @@ public void start() {
         // when a 'group.id' is specified (if offsets happen to have been committed unexpectedly).
         consumer.seekToBeginning(partitions);
 
-        readToLogEnd();
+        readToLogEnd(true);
 
         thread = new WorkThread();
         thread.start();
@@ -319,9 +323,16 @@ private void poll(long timeoutMs) {
         }
     }
 
-    private void readToLogEnd() {
+    /**
+     * This method finds the end offsets of the Kafka log's topic partitions, optionally retrying
+     * if the {@code listOffsets()} method of the admin client throws a {@link RetriableException}.
+     *
+     * @param shouldRetry Boolean flag to enable retry for the admin client {@code listOffsets()} call.
+     * @see TopicAdmin#retryEndOffsets
+     */
+    private void readToLogEnd(boolean shouldRetry) {
         Set<TopicPartition> assignment = consumer.assignment();
-        Map<TopicPartition, Long> endOffsets = readEndOffsets(assignment);
+        Map<TopicPartition, Long> endOffsets = readEndOffsets(assignment, shouldRetry);
         log.trace("Reading to end of log offsets {}", endOffsets);
 
         while (!endOffsets.isEmpty()) {
@@ -345,7 +356,7 @@ private void readToLogEnd() {
     }
 
     // Visible for testing
-    Map<TopicPartition, Long> readEndOffsets(Set<TopicPartition> assignment) {
+    Map<TopicPartition, Long> readEndOffsets(Set<TopicPartition> assignment, boolean shouldRetry) {
         log.trace("Reading to end of offset log");
 
         // Note that we'd prefer to not use the consumer to find the end offsets for the assigned topic partitions.
@@ -360,6 +371,12 @@ Map<TopicPartition, Long> readEndOffsets(Set<TopicPartition> assignment) {
             // Use the admin client to immediately find the end offsets for the assigned topic partitions.
             // Unlike using the consumer
             try {
+                if (shouldRetry) {
+                    return admin.retryEndOffsets(assignment,
+                            ADMIN_CLIENT_RETRY_DURATION,
+                            ADMIN_CLIENT_RETRY_BACKOFF_MS);
+                }
+
                 return admin.endOffsets(assignment);
             } catch (UnsupportedVersionException e) {
                 // This may happen with really old brokers that don't support the auto topic creation
@@ -395,7 +412,7 @@ public void run() {
 
                     if (numCallbacks > 0) {
                         try {
-                            readToLogEnd();
+                            readToLogEnd(false);
                             log.trace("Finished read to end log for topic {}", topic);
                         } catch (TimeoutException e) {
                             log.warn("Timeout while reading log to end for topic '{}'. Retrying automatically. " +

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

@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.connect.util;
+
+import org.apache.kafka.common.errors.RetriableException;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Optional;
+import java.util.concurrent.Callable;
+import java.util.function.Supplier;
+
+public class RetryUtil {
+    private static final Logger log = LoggerFactory.getLogger(RetryUtil.class);
+
+    /**
+     * The method executes the callable at least once, optionally retrying the callable if
+     * {@link org.apache.kafka.connect.errors.RetriableException} is being thrown.  If timeout is exhausted,
+     * then the last exception is wrapped with a {@link org.apache.kafka.connect.errors.ConnectException} and thrown.
+     *
+     * <p>{@code description} supplies the message that indicates the purpose of the callable since the message will
+     * be used for logging.  For example, "list offsets". If the supplier is null or the supplied string is
+     * null, {@code callable} will be used as the default string.
+     *
+     * <p>The task will be executed at least once. No retries will be performed 
+     * if {@code timeoutDuration} is 0 or negative, or if {@code timeoutDuration} is less than {@code retryBackoffMs}.
+     *
+     * <p>A {@code retryBackoffMs} that is negative or zero will result in no delays between retries.
+     *
+     * @param callable          the function to execute
+     * @param description       supplier that provides custom message for logging purpose
+     * @param timeoutDuration   timeout duration; must not be null
+     * @param retryBackoffMs    the number of milliseconds to delay upon receiving a
+     *                          {@link org.apache.kafka.connect.errors.RetriableException} before retrying again
+     * @throws ConnectException If the task exhausted all the retries
+     */
+    public static <T> T retryUntilTimeout(Callable<T> callable, Supplier<String> description, Duration timeoutDuration, long retryBackoffMs) throws Exception {
+        // if null supplier or string is provided, the message will be default to "callabe"
+        final String descriptionStr = Optional.ofNullable(description)
+                .map(Supplier::get)
+                .orElse("callable");
+
+        // handling null duration
+        final long timeoutMs = Optional.ofNullable(timeoutDuration)
+                .map(Duration::toMillis)
+                .orElse(0L);
+
+        if (retryBackoffMs < 0) {
+            log.debug("Assuming no retry backoff since retryBackoffMs={} is negative", retryBackoffMs);
+            retryBackoffMs = 0;
+        }
+        if (timeoutMs <= 0 || retryBackoffMs >= timeoutMs) {
+            log.debug("Executing {} only once, since timeoutMs={} is not larger than retryBackoffMs={}",
+                    descriptionStr, timeoutMs, retryBackoffMs);
+            return callable.call();
+        }
+
+        final long end = System.currentTimeMillis() + timeoutMs;
+        int attempt = 0;
+        Throwable lastError = null;
+        do {
+            attempt++;
+            try {
+                return callable.call();
+            } catch (RetriableException | org.apache.kafka.connect.errors.RetriableException e) {
+                log.warn("Attempt {} to {} resulted in RetriableException; retrying automatically. " +
+                        "Reason: {}", attempt, descriptionStr, e.getMessage(), e);
+                lastError = e;
+            } catch (WakeupException e) {
+                lastError = e;
+            }
+
+            long millisRemaining = Math.max(0, end - System.currentTimeMillis());
+            if (millisRemaining < retryBackoffMs) {
+                // exit when the time remaining is less than retryBackoffMs
+                break;
+            }
+            Utils.sleep(retryBackoffMs);
+        } while (System.currentTimeMillis() < end);
+
+        throw new ConnectException("Fail to " + descriptionStr + " after " + attempt + " attempts.  Reason: " + lastError.getMessage(), lastError);
+    }
+}

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

@@ -655,7 +655,7 @@ public Map<String, Config> describeTopicConfigs(String... topicNames) {
      * @throws TimeoutException if the offset metadata could not be fetched before the amount of time allocated
      *         by {@code request.timeout.ms} expires, and this call can be retried
      * @throws LeaderNotAvailableException if the leader was not available and this call can be retried
-     * @throws RetriableException if a retriable error occurs, or the thread is interrupted while attempting 
+     * @throws RetriableException if a retriable error occurs, or the thread is interrupted while attempting
      *         to perform this operation
      * @throws ConnectException if a non retriable error occurs
      */
@@ -703,6 +703,33 @@ public Map<TopicPartition, Long> endOffsets(Set<TopicPartition> partitions) {
         return result;
     }
 
+    /**
+     * Fetch the most recent offset for each of the supplied {@link TopicPartition} objects, and performs retry when
+     * {@link org.apache.kafka.connect.errors.RetriableException} is thrown.
+     *
+     * @param partitions        the topic partitions
+     * @param timeoutDuration   timeout duration; may not be null
+     * @param retryBackoffMs    the number of milliseconds to delay upon receiving a
+     *                          {@link org.apache.kafka.connect.errors.RetriableException} before retrying again;
+     *                          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 ConnectException if {@code timeoutDuration} is exhausted
+     * @see TopicAdmin#endOffsets(Set)
+     */
+    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 (Exception e) {
+            throw new ConnectException("Failed to list offsets for topic partitions.", e);
+        }
+    }
+
     @Override
     public void close() {
         admin.close();

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

@@ -490,13 +490,15 @@ public void testReadEndOffsetsUsingAdmin() throws Exception {
         Map<TopicPartition, Long> endOffsets = new HashMap<>();
         endOffsets.put(TP0, 0L);
         endOffsets.put(TP1, 0L);
+        admin.retryEndOffsets(EasyMock.eq(tps), EasyMock.anyObject(), EasyMock.anyLong());
+        PowerMock.expectLastCall().andReturn(endOffsets).times(1);
         admin.endOffsets(EasyMock.eq(tps));
-        PowerMock.expectLastCall().andReturn(endOffsets).times(2);
+        PowerMock.expectLastCall().andReturn(endOffsets).times(1);
 
         PowerMock.replayAll();
 
         store.start();
-        assertEquals(endOffsets, store.readEndOffsets(tps));
+        assertEquals(endOffsets, store.readEndOffsets(tps, false));
     }
 
     @Test
@@ -507,7 +509,7 @@ public void testReadEndOffsetsUsingAdminThatFailsWithUnsupported() throws Except
 
         Set<TopicPartition> tps = new HashSet<>(Arrays.asList(TP0, TP1));
         // Getting end offsets using the admin client should fail with unsupported version
-        admin.endOffsets(EasyMock.eq(tps));
+        admin.retryEndOffsets(EasyMock.eq(tps), EasyMock.anyObject(), EasyMock.anyLong());
         PowerMock.expectLastCall().andThrow(new UnsupportedVersionException("too old"));
 
         // Falls back to the consumer
@@ -519,7 +521,7 @@ public void testReadEndOffsetsUsingAdminThatFailsWithUnsupported() throws Except
         PowerMock.replayAll();
 
         store.start();
-        assertEquals(endOffsets, store.readEndOffsets(tps));
+        assertEquals(endOffsets, store.readEndOffsets(tps, false));
     }
 
     @Test
@@ -533,7 +535,7 @@ public void testReadEndOffsetsUsingAdminThatFailsWithRetriable() throws Exceptio
         endOffsets.put(TP0, 0L);
         endOffsets.put(TP1, 0L);
         // Getting end offsets upon startup should work fine
-        admin.endOffsets(EasyMock.eq(tps));
+        admin.retryEndOffsets(EasyMock.eq(tps), EasyMock.anyObject(), EasyMock.anyLong());
         PowerMock.expectLastCall().andReturn(endOffsets).times(1);
         // Getting end offsets using the admin client should fail with leader not available
         admin.endOffsets(EasyMock.eq(tps));
@@ -542,7 +544,7 @@ public void testReadEndOffsetsUsingAdminThatFailsWithRetriable() throws Exceptio
         PowerMock.replayAll();
 
         store.start();
-        assertThrows(LeaderNotAvailableException.class, () -> store.readEndOffsets(tps));
+        assertThrows(LeaderNotAvailableException.class, () -> store.readEndOffsets(tps, false));
     }
 
     @SuppressWarnings("unchecked")