[GitHub] [kafka] chia7712 commented on a change in pull request #10158: KAFKA-12343: Handle exceptions better in TopicAdmin, including UnsupportedVersionException

[GitBox <gi...@apache.org>]

chia7712 commented on a change in pull request #10158:
URL: https://github.com/apache/kafka/pull/10158#discussion_r578907382



##########
File path: connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java
##########
@@ -366,6 +348,44 @@ private void readToLogEnd() {
         }
     }
 
+    // Visible for testing
+    Map<TopicPartition, Long> readEndOffsets(Set<TopicPartition> assignment) {
+        log.trace("Reading to end of offset log");
+
+        Map<TopicPartition, Long> endOffsets;
+        // Note that we'd prefer to not use the consumer to find the end offsets for the assigned topic partitions.
+        // That is because it's possible that the consumer is already blocked waiting for new records to appear, when
+        // the consumer is already at the end. In such cases, using 'consumer.endOffsets(...)' will block until at least
+        // one more record becomes available, meaning we can't even check whether we're at the end offset.
+        // Since all we're trying to do here is get the end offset, we should use the supplied admin client
+        // (if available)
+        // (which prevents 'consumer.endOffsets(...)'
+        // from
+
+        // Deprecated constructors do not provide an admin supplier, so the admin is potentially null.
+        if (useAdminForListOffsets) {
+            // Use the admin client to immediately find the end offsets for the assigned topic partitions.
+            // Unlike using the consumer
+            try {
+                endOffsets = admin.endOffsets(assignment);

Review comment:
       How about `return admin.endOffsets(assignment);`

##########
File path: connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java
##########
@@ -390,7 +410,11 @@ public void run() {
                             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. " +
-                                "This may occur when brokers are unavailable or unreachable. Reason: {}", topic, e.getMessage());
+                                     "This may occur when brokers are unavailable or unreachable. Reason: {}", topic, e.getMessage());
+                            continue;
+                        } catch (RetriableException | org.apache.kafka.connect.errors.RetriableException e) {

Review comment:
       `readToLogEnd` is called by `start`. Should we add similar exception handle for that? 

##########
File path: connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java
##########
@@ -321,29 +325,7 @@ private void readToLogEnd() {
         log.trace("Reading to end of offset log");

Review comment:
       redundant log message. `readEndOffsets(Set<TopicPartition>)` has similar log.

##########
File path: connect/runtime/src/main/java/org/apache/kafka/connect/util/TopicAdmin.java
##########
@@ -651,6 +651,10 @@ public Config describeTopicConfig(String topic) {
      * @param partitions the topic partitions
      * @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 admin client cannot read end offsets
+     * @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, the operation takes too long, or the

Review comment:
       `the operation takes too long` 
   ^^^^^^^^^^^^^^^^^^^^^^^ this message gets invalid since `TimeoutException` is not wrapped to `RetriableException` anymore.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org