kkonstantine commented on a change in pull request #10158: URL: https://github.com/apache/kafka/pull/10158#discussion_r578998452
########## File path: connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java ########## @@ -366,6 +346,42 @@ 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 { + return admin.endOffsets(assignment); + } catch (UnsupportedVersionException e) { + // This may happen with really old brokers that don't support the auto topic creation + // field in metadata requests + log.debug("Reading to end of log offsets with consumer since admin client is unsupported: {}", e.getMessage()); + useAdminForListOffsets = false; Review comment: Using `admin = null` here allows to GC the unused admin instance earlier, right? Not a big gain, but also I don't see much benefit by using a variable such as `useAdminForListOffsets` ########## File path: connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java ########## @@ -161,6 +164,7 @@ public void start() { // Create the topic admin client and initialize the topic ... admin = topicAdminSupplier.get(); // may be null + useAdminForListOffsets = admin != null; Review comment: It's not obvious to me why we have to wait to call `topicAdminSupplier.get();` in `start`. I think we can call it in the constructor and not save the field `topicAdminSupplier` but keep only `admin` as a member field. Unfortunately it can't be declared `final` because it may be reset (set to `null`) in the exception handling below but still, using fewer member fields seems reasonable and does not reduce the usability in this class IMO. ########## File path: connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java ########## @@ -366,6 +346,42 @@ 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 Review comment: Since we are moving this comment, probably it's a good idea to fix this incomplete sentence here too ---------------------------------------------------------------- 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: us...@infra.apache.org