beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From amits...@apache.org
Subject [1/2] incubator-beam git commit: [BEAM-744] UnboundedKafkaReader should return as soon as it can.
Date Wed, 19 Oct 2016 18:43:42 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/master ea04e618e -> b0cb2e87b


[BEAM-744] UnboundedKafkaReader should return as soon as it can.

Use timeout directly in nextBatch()


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/84c6649c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/84c6649c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/84c6649c

Branch: refs/heads/master
Commit: 84c6649cd63c33ca79ad43e8973dbf765e27a5d0
Parents: ea04e61
Author: Sela <ansela@paypal.com>
Authored: Tue Oct 18 22:03:25 2016 +0300
Committer: Sela <ansela@paypal.com>
Committed: Wed Oct 19 21:34:07 2016 +0300

----------------------------------------------------------------------
 .../java/org/apache/beam/sdk/io/kafka/KafkaIO.java    | 14 +++++---------
 1 file changed, 5 insertions(+), 9 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/84c6649c/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
index 2030789..834104e 100644
--- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
+++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
@@ -756,9 +756,6 @@ public class KafkaIO {
     private Iterator<PartitionState> curBatch = Collections.emptyIterator();
 
     private static final Duration KAFKA_POLL_TIMEOUT = Duration.millis(1000);
-    // how long to wait for new records from kafka consumer inside start()
-    private static final Duration START_NEW_RECORDS_POLL_TIMEOUT = Duration.standardSeconds(5);
-    // how long to wait for new records from kafka consumer inside advance()
     private static final Duration NEW_RECORDS_POLL_TIMEOUT = Duration.millis(10);
 
     // Use a separate thread to read Kafka messages. Kafka Consumer does all its work including
@@ -888,12 +885,13 @@ public class KafkaIO {
       LOG.info("{}: Returning from consumer pool loop", this);
     }
 
-    private void nextBatch(Duration timeout) {
+    private void nextBatch() {
       curBatch = Collections.emptyIterator();
 
       ConsumerRecords<byte[], byte[]> records;
       try {
-        records = availableRecordsQueue.poll(timeout.getMillis(),
+        // poll available records, wait (if necessary) up to the specified timeout.
+        records = availableRecordsQueue.poll(NEW_RECORDS_POLL_TIMEOUT.getMillis(),
                                              TimeUnit.MILLISECONDS);
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
@@ -966,9 +964,7 @@ public class KafkaIO {
             }
           }, 0, OFFSET_UPDATE_INTERVAL_SECONDS, TimeUnit.SECONDS);
 
-      // Wait for longer than normal when fetching a batch to improve chances a record is
available
-      // when start() returns.
-      nextBatch(START_NEW_RECORDS_POLL_TIMEOUT);
+      nextBatch();
       return advance();
     }
 
@@ -1032,7 +1028,7 @@ public class KafkaIO {
           return true;
 
         } else { // -- (b)
-          nextBatch(NEW_RECORDS_POLL_TIMEOUT);
+          nextBatch();
 
           if (!curBatch.hasNext()) {
             return false;


Mime
View raw message