Repository: flink Updated Branches: refs/heads/master e927ec0be -> c47cb7af1
[FLINK-3392] [kafka] Fix unsynchronized access in ClosableBlockingQueue Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/c47cb7af Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/c47cb7af Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/c47cb7af Branch: refs/heads/master Commit: c47cb7af1474b08ee1b7b7a813d03da022015ed1 Parents: e927ec0 Author: Stephan Ewen <[email protected]> Authored: Fri Feb 12 19:30:22 2016 +0100 Committer: Stephan Ewen <[email protected]> Committed: Fri Feb 12 19:30:22 2016 +0100 ---------------------------------------------------------------------- .../connectors/kafka/internals/ClosableBlockingQueue.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/c47cb7af/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java index 856c2ad..23ff276 100644 --- a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java +++ b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java @@ -104,7 +104,12 @@ public class ClosableBlockingQueue<E> { * @return The number of elements currently in the queue. */ public int size() { - return elements.size(); + lock.lock(); + try { + return elements.size(); + } finally { + lock.unlock(); + } } /**
