Repository: incubator-apex-core Updated Branches: refs/heads/master 00939d54b -> 7bbb4cfd2
APEXCORE-362 - NPE in StreamingContainerManager. Fixed race condition between the thread that insert into endWindowStatsOperatorMap and the thread that removes entries when endWindowStatsOperatorMap exceeds 1000 entries. Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/7bbb4cfd Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/7bbb4cfd Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/7bbb4cfd Branch: refs/heads/master Commit: 7bbb4cfd2af6d5c5c99c58222c44a848bc6534b2 Parents: 00939d5 Author: Vlad Rozov <[email protected]> Authored: Sun Feb 28 09:07:29 2016 -0800 Committer: David Yan <[email protected]> Committed: Mon Feb 29 08:54:31 2016 -0800 ---------------------------------------------------------------------- .../com/datatorrent/stram/StreamingContainerManager.java | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/7bbb4cfd/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java ---------------------------------------------------------------------- diff --git a/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java b/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java index ae79728..aa79243 100644 --- a/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java +++ b/engine/src/main/java/com/datatorrent/stram/StreamingContainerManager.java @@ -1700,8 +1700,12 @@ public class StreamingContainerManager implements PlanContext if (stats.windowId > currentEndWindowStatsWindowId) { Map<Integer, EndWindowStats> endWindowStatsMap = endWindowStatsOperatorMap.get(stats.windowId); if (endWindowStatsMap == null) { - endWindowStatsOperatorMap.putIfAbsent(stats.windowId, new ConcurrentSkipListMap<Integer, EndWindowStats>()); - endWindowStatsMap = endWindowStatsOperatorMap.get(stats.windowId); + endWindowStatsMap = new ConcurrentSkipListMap<Integer, EndWindowStats>(); + Map<Integer, EndWindowStats> endWindowStatsMapPrevious = + endWindowStatsOperatorMap.putIfAbsent(stats.windowId, endWindowStatsMap); + if (endWindowStatsMapPrevious != null) { + endWindowStatsMap = endWindowStatsMapPrevious; + } } endWindowStatsMap.put(shb.getNodeId(), endWindowStats);
