[GitHub] [incubator-gobblin] sv2000 commented on a change in pull request #2900: [GOBBLIN-1040] HighLevelConsumer re-design by removing references to …

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

sv2000 commented on a change in pull request #2900: [GOBBLIN-1040] HighLevelConsumer re-design by removing references to …
URL: https://github.com/apache/incubator-gobblin/pull/2900#discussion_r386165680
 
 

 ##########
 File path: gobblin-modules/gobblin-kafka-09/src/main/java/org/apache/gobblin/kafka/client/Kafka09ConsumerClient.java
 ##########
 @@ -160,14 +164,30 @@ public long getLatestOffset(KafkaPartition partition) throws KafkaOffsetRetrieva
 
     this.consumer.assign(Lists.newArrayList(new TopicPartition(partition.getTopicName(), partition.getId())));
     this.consumer.seek(new TopicPartition(partition.getTopicName(), partition.getId()), nextOffset);
-    ConsumerRecords<K, V> consumerRecords = consumer.poll(super.fetchTimeoutMillis);
-    return Iterators.transform(consumerRecords.iterator(), new Function<ConsumerRecord<K, V>, KafkaConsumerRecord>() {
+    return consume();
+  }
 
-      @Override
-      public KafkaConsumerRecord apply(ConsumerRecord<K, V> input) {
-        return new Kafka09ConsumerRecord<>(input);
-      }
-    });
+  @Override
+  public synchronized Iterator<KafkaConsumerRecord> consume() {
 
 Review comment:
   Why the use of "synchronized" here? Do we expect 2 or more threads to call consume() on the same underlying Kafka consumer? 

----------------------------------------------------------------
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


With regards,
Apache Git Services