You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by dh...@apache.org on 2016/05/11 20:27:24 UTC
[1/2] incubator-beam git commit: Closes #290
Repository: incubator-beam
Updated Branches:
refs/heads/master e6cb86a3c -> 7108bc025
Closes #290
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/7108bc02
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/7108bc02
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/7108bc02
Branch: refs/heads/master
Commit: 7108bc0250c9f40ae3ebebdfac888da7dd05ec67
Parents: e6cb86a 93931f3
Author: Dan Halperin <dh...@google.com>
Authored: Wed May 11 13:18:31 2016 -0700
Committer: Dan Halperin <dh...@google.com>
Committed: Wed May 11 13:18:31 2016 -0700
----------------------------------------------------------------------
.../org/apache/beam/sdk/io/kafka/KafkaIO.java | 35 ++++++++++++++++----
1 file changed, 29 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
[2/2] incubator-beam git commit: [BEAM-52] Fix a
ConcurrentModificationException in KafkaIO
Posted by dh...@apache.org.
[BEAM-52] Fix a ConcurrentModificationException in KafkaIO
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/93931f3a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/93931f3a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/93931f3a
Branch: refs/heads/master
Commit: 93931f3ade087dcf924ba77b0fd1394752ed3954
Parents: e6cb86a
Author: Raghu Angadi <ra...@google.com>
Authored: Wed May 4 18:34:20 2016 -0700
Committer: Dan Halperin <dh...@google.com>
Committed: Wed May 11 13:18:31 2016 -0700
----------------------------------------------------------------------
.../org/apache/beam/sdk/io/kafka/KafkaIO.java | 35 ++++++++++++++++----
1 file changed, 29 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/93931f3a/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 4f353df..3d22824 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
@@ -81,6 +81,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import javax.annotation.Nullable;
@@ -686,7 +687,7 @@ public class KafkaIO {
private final ExecutorService consumerPollThread = Executors.newSingleThreadExecutor();
private final SynchronousQueue<ConsumerRecords<byte[], byte[]>> availableRecordsQueue =
new SynchronousQueue<>();
- private volatile boolean closed = false;
+ private AtomicBoolean closed = new AtomicBoolean(false);
// Backlog support :
// Kafka consumer does not have an API to fetch latest offset for topic. We need to seekToEnd()
@@ -792,10 +793,10 @@ public class KafkaIO {
private void consumerPollLoop() {
// Read in a loop and enqueue the batch of records, if any, to availableRecordsQueue
- while (!closed) {
+ while (!closed.get()) {
try {
ConsumerRecords<byte[], byte[]> records = consumer.poll(KAFKA_POLL_TIMEOUT.getMillis());
- if (!records.isEmpty()) {
+ if (!records.isEmpty() && !closed.get()) {
availableRecordsQueue.put(records); // blocks until dequeued.
}
} catch (InterruptedException e) {
@@ -817,6 +818,7 @@ public class KafkaIO {
records = availableRecordsQueue.poll(NEW_RECORDS_POLL_TIMEOUT.getMillis(),
TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
LOG.warn("{}: Unexpected", this, e);
return;
}
@@ -1041,11 +1043,32 @@ public class KafkaIO {
@Override
public void close() throws IOException {
- closed = true;
- availableRecordsQueue.poll(); // drain unread batch, this unblocks consumer thread.
- consumer.wakeup();
+ closed.set(true);
consumerPollThread.shutdown();
offsetFetcherThread.shutdown();
+
+ boolean isShutdown = false;
+
+ // Wait for threads to shutdown. Trying this a loop to handle a tiny race where poll thread
+ // might block to enqueue right after availableRecordsQueue.poll() below.
+ while (!isShutdown) {
+
+ consumer.wakeup();
+ offsetConsumer.wakeup();
+ availableRecordsQueue.poll(); // drain unread batch, this unblocks consumer thread.
+ try {
+ isShutdown = consumerPollThread.awaitTermination(10, TimeUnit.SECONDS)
+ && offsetFetcherThread.awaitTermination(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e); // not expected
+ }
+
+ if (!isShutdown) {
+ LOG.warn("An internal thread is taking a long time to shutdown. will retry.");
+ }
+ }
+
Closeables.close(offsetConsumer, true);
Closeables.close(consumer, true);
}