[jira] [Updated] (SAMZA-1829) MessageStreamAssert check throws NPE on expected.toArray()

["Cameron Lee (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/SAMZA-1829?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Cameron Lee updated SAMZA-1829:
-------------------------------
    Comment: was deleted

(was: Related to https://issues.apache.org/jira/browse/SAMZA-2036)

> MessageStreamAssert check throws NPE on expected.toArray()
> ----------------------------------------------------------
>
>                 Key: SAMZA-1829
>                 URL: https://issues.apache.org/jira/browse/SAMZA-1829
>             Project: Samza
>          Issue Type: Bug
>            Reporter: Yi Pan (Data Infrastructure)
>            Priority: Major
>             Fix For: 1.0
>
>
> In MessageStreamAssert.CheckAgainstExpected, expected is a marked as a transient variable and it is not initialized in SinkFunction.init() either. Hence, causing NPE when check() is called:
> {code}
> 7794 [Samza StreamProcessor Container Thread-0] ERROR org.apache.samza.task.AsyncRunLoop - Caught throwable and stopping run loop
> org.apache.samza.SamzaException: Callback failed for task Partition 0, ssp SystemStreamPartition [kafka, BroadcastTest-1-broadcast-pv, 0], offset 0.
> 	at org.apache.samza.task.TaskCallbackImpl.failure(TaskCallbackImpl.java:89)
> 	at org.apache.samza.task.AsyncStreamTaskAdapter.process(AsyncStreamTaskAdapter.java:75)
> 	at org.apache.samza.task.AsyncStreamTaskAdapter.processAsync(AsyncStreamTaskAdapter.java:63)
> 	at org.apache.samza.container.TaskInstance$$anonfun$process$1.apply$mcV$sp(TaskInstance.scala:188)
> 	at org.apache.samza.container.TaskInstanceExceptionHandler.maybeHandle(TaskInstanceExceptionHandler.scala:54)
> 	at org.apache.samza.container.TaskInstance.process(TaskInstance.scala:186)
> 	at org.apache.samza.task.AsyncRunLoop$AsyncTaskWorker.process(AsyncRunLoop.java:465)
> 	at org.apache.samza.task.AsyncRunLoop$AsyncTaskWorker.run(AsyncRunLoop.java:407)
> 	at org.apache.samza.task.AsyncRunLoop$AsyncTaskWorker.access$300(AsyncRunLoop.java:341)
> 	at org.apache.samza.task.AsyncRunLoop.runTasks(AsyncRunLoop.java:232)
> 	at org.apache.samza.task.AsyncRunLoop.run(AsyncRunLoop.java:166)
> 	at org.apache.samza.container.SamzaContainer.run(SamzaContainer.scala:825)
> 	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> 	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> 	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> 	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> 	at java.lang.Thread.run(Thread.java:745)
> Caused by: java.lang.NullPointerException
> 	at org.apache.samza.test.framework.MessageStreamAssert$CheckAgainstExpected.apply(MessageStreamAssert.java:164)
> 	at org.apache.samza.operators.impl.SinkOperatorImpl.handleMessage(SinkOperatorImpl.java:54)
> 	at org.apache.samza.operators.impl.OperatorImpl.onMessage(OperatorImpl.java:183)
> 	at org.apache.samza.operators.impl.OperatorImpl.lambda$null$1(OperatorImpl.java:199)
> 	at java.lang.Iterable.forEach(Iterable.java:75)
> 	at org.apache.samza.operators.impl.OperatorImpl.lambda$onMessage$2(OperatorImpl.java:198)
> 	at java.util.Collections$SingletonList.forEach(Collections.java:4822)
> 	at org.apache.samza.operators.impl.OperatorImpl.onMessage(OperatorImpl.java:197)
> 	at org.apache.samza.task.StreamOperatorTask.process(StreamOperatorTask.java:113)
> 	at org.apache.samza.task.AsyncStreamTaskAdapter.process(AsyncStreamTaskAdapter.java:72)
> 	... 15 more
> {code}



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)