Repository: incubator-apex-core Updated Branches: refs/heads/release-3.3 21ad4cc9b -> c210e35f1
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/c210e35f Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/c210e35f Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/c210e35f Branch: refs/heads/release-3.3 Commit: c210e35f11289e577079a2c17418a6d8e99dee2f Parents: 21ad4cc Author: Vlad Rozov <[email protected]> Authored: Sun Feb 28 09:07:29 2016 -0800 Committer: David Yan <[email protected]> Committed: Mon Feb 29 08:53:56 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/c210e35f/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);
