You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by "Bharath Kumarasubramanian (JIRA)" <ji...@apache.org> on 2019/04/26 22:40:00 UTC
[jira] [Resolved] (SAMZA-2092) Make window operator thread safe
[ https://issues.apache.org/jira/browse/SAMZA-2092?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Bharath Kumarasubramanian resolved SAMZA-2092.
----------------------------------------------
Resolution: Fixed
Fixed in SAMZA-2055
> Make window operator thread safe
> --------------------------------
>
> Key: SAMZA-2092
> URL: https://issues.apache.org/jira/browse/SAMZA-2092
> Project: Samza
> Issue Type: Bug
> Reporter: Bharath Kumarasubramanian
> Assignee: Bharath Kumarasubramanian
> Priority: Major
>
> Currently, window operator maintains some state which when modified concurrently could result in undesired behavior. This is a problem for high level stream applications with task.max.concurrency > 1.
> Here is a sample stacktrace that illustrates NPE caused due to concurrent modification on callback state queue maintained by trigger handler.
> {code:java}
> Caused by: java.lang.NullPointerException at org.apache.samza.operators.impl.TriggerScheduler$TriggerCallbackState.compareTo(TriggerScheduler.java:111) at org.apache.samza.operators.impl.TriggerScheduler$TriggerCallbackState.compareTo(TriggerScheduler.java:83) at java.util.PriorityQueue.siftUpComparable(PriorityQueue.java:656) at java.util.PriorityQueue.siftUp(PriorityQueue.java:647) at java.util.PriorityQueue.offer(PriorityQueue.java:344) at java.util.PriorityQueue.add(PriorityQueue.java:321) at org.apache.samza.operators.impl.TriggerScheduler.scheduleCallback(TriggerScheduler.java:55) at org.apache.samza.operators.triggers.TimeTriggerImpl.onMessage(TimeTriggerImpl.java:53) at org.apache.samza.operators.impl.WindowOperatorImpl$TriggerImplHandler.onMessage(WindowOperatorImpl.java:400) at org.apache.samza.operators.impl.WindowOperatorImpl.handleMessage(WindowOperatorImpl.java:170) at org.apache.samza.operators.impl.OperatorImpl.onMessage(OperatorImpl.java:184){code}
> In the above scenario, one thread is adding a callback state to the priority queue, while another thread is canceling one of the existing callback state. Due to concurrent modification, add operation fails with NPE since the reference used for comparison is null.
> As part of this ticket, we also want to make sure all high level operators wherever applicable is thread safe. It is a pre-requisite for async API support in high level applications.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)