Non-deterministic test failure in TensorflowIrisIT

[mspruc]

TensorFlowIrisIT test can non-deterministically fail with error:

org.apache.wayang.core.api.exception.WayangException: Executing T[JavaCollect[convert out@JavaMap[1->1, id=400f7211]]] failed.
        at org.apache.wayang.java.execution.JavaExecutor.execute(JavaExecutor.java:92)
        at org.apache.wayang.core.platform.PushExecutorTemplate.execute(PushExecutorTemplate.java:73)
        at org.apache.wayang.core.platform.PushExecutorTemplate$StageExecution.execute(PushExecutorTemplate.java:195)
        at org.apache.wayang.core.platform.PushExecutorTemplate$StageExecution.doExecute(PushExecutorTemplate.java:166)
        at org.apache.wayang.core.util.OneTimeExecutable.tryExecute(OneTimeExecutable.java:41)
        at org.apache.wayang.core.util.OneTimeExecutable.execute(OneTimeExecutable.java:54)
        at org.apache.wayang.core.platform.PushExecutorTemplate$StageExecution.executeStage(PushExecutorTemplate.java:156)
        at org.apache.wayang.core.platform.PushExecutorTemplate.execute(PushExecutorTemplate.java:61)
        at org.apache.wayang.core.platform.CrossPlatformExecutor.execute(CrossPlatformExecutor.java:378)
        at org.apache.wayang.core.platform.CrossPlatformExecutor.executeSingleStage(CrossPlatformExecutor.java:248)
        at org.apache.wayang.core.platform.CrossPlatformExecutor.runToBreakpoint(CrossPlatformExecutor.java:320)
        at org.apache.wayang.core.platform.CrossPlatformExecutor.executeUntilBreakpoint(CrossPlatformExecutor.java:156)
        at org.apache.wayang.core.api.Job.execute(Job.java:542)
        at org.apache.wayang.core.api.Job.doExecute(Job.java:326)
        at org.apache.wayang.core.util.OneTimeExecutable.tryExecute(OneTimeExecutable.java:41)
        at org.apache.wayang.core.util.OneTimeExecutable.execute(OneTimeExecutable.java:54)
        at org.apache.wayang.core.api.Job.execute(Job.java:245)
        at org.apache.wayang.core.api.WayangContext.execute(WayangContext.java:124)
        at org.apache.wayang.core.api.WayangContext.execute(WayangContext.java:112)
        at org.apache.wayang.core.api.WayangContext.execute(WayangContext.java:100)
        at org.apache.wayang.tests.TensorflowIrisIT.test(TensorflowIrisIT.java:155)
        at java.base/java.lang.reflect.Method.invoke(Method.java:569)
        at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
        at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
Caused by: java.lang.IllegalArgumentException: Comparison method violates its general contract!
        at java.base/java.util.TimSort.mergeLo(TimSort.java:781)
        at java.base/java.util.TimSort.mergeAt(TimSort.java:518)
        at java.base/java.util.TimSort.mergeCollapse(TimSort.java:448)
        at java.base/java.util.TimSort.sort(TimSort.java:245)
        at java.base/java.util.Arrays.sort(Arrays.java:1307)
        at java.base/java.util.ArrayList.sort(ArrayList.java:1721)
        at java.base/java.util.stream.SortedOps$RefSortingSink.end(SortedOps.java:392)
        at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:510)
        at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
        at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921)
        at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
        at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682)
        at org.apache.wayang.java.operators.JavaCollectOperator.evaluate(JavaCollectOperator.java:61)
        at org.apache.wayang.java.execution.JavaExecutor.execute(JavaExecutor.java:82)
        ... 23 more

with execution plan:

>>> ExecutionStage[T[JavaTextFileSource[0->1, id=24eec6dd]]]:
>     JavaTextFileSource[0->1, id=24eec6dd] => StreamChannel => JavaMap[1->1, id=20a8b3b9]
>     JavaMap[1->1, id=20a8b3b9] => StreamChannel => JavaCollect[convert out@JavaMap[1->1, id=20a8b3b9]]
> Out JavaCollect[convert out@JavaMap[1->1, id=20a8b3b9]] => CollectionChannel
> Out JavaCollect[convert out@JavaMap[1->1, id=20a8b3b9]] => CollectionChannel

>>> ExecutionStage[T[JavaTextFileSource[0->1, id=52043147]]]:
>     JavaTextFileSource[0->1, id=52043147] => StreamChannel => JavaSort[1->1, id=5f51c50b]
>     JavaSort[1->1, id=5f51c50b] => StreamChannel => JavaMap[1->1, id=400f7211]
>     JavaMap[1->1, id=400f7211] => StreamChannel => JavaCollect[convert out@JavaMap[1->1, id=400f7211]]
> Out JavaCollect[convert out@JavaMap[1->1, id=400f7211]] => CollectionChannel
> Out JavaCollect[convert out@JavaMap[1->1, id=400f7211]] => CollectionChannel

>>> ExecutionStage[T[JavaMap[1->1, id=26a1515a]]]:
> In  CollectionChannel => JavaMap[1->1, id=26a1515a]
>     JavaMap[1->1, id=26a1515a] => StreamChannel => JavaCollect[convert out@JavaMap[1->1, id=26a1515a]]
> Out JavaCollect[convert out@JavaMap[1->1, id=26a1515a]] => CollectionChannel

>>> ExecutionStage[T[JavaMap[1->1, id=57c03512]]]:
> In  CollectionChannel => JavaMap[1->1, id=57c03512]
>     JavaMap[1->1, id=57c03512] => StreamChannel => JavaLocalCallbackSink[1->0, id=22c6f0e9]

>>> ExecutionStage[T[JavaMap[1->1, id=7a7522b9]]]:
> In  CollectionChannel => JavaMap[1->1, id=7a7522b9]
>     JavaMap[1->1, id=7a7522b9] => StreamChannel => JavaCollect[convert out@JavaMap[1->1, id=7a7522b9]]
> Out JavaCollect[convert out@JavaMap[1->1, id=7a7522b9]] => CollectionChannel

>>> ExecutionStage[T[JavaMap[1->1, id=72754682]]]:
> In  CollectionChannel => JavaMap[1->1, id=72754682]
>     JavaMap[1->1, id=72754682] => StreamChannel => JavaCollect[convert out@JavaMap[1->1, id=72754682]]
> Out JavaCollect[convert out@JavaMap[1->1, id=72754682]] => CollectionChannel

>>> ExecutionStage[T[TensorflowCollectionSource[convert out@JavaMap[1->1, id=26a1515a]]]]:
> In  CollectionChannel => TensorflowCollectionSource[convert out@JavaMap[1->1, id=26a1515a]]
> Out TensorflowCollectionSource[convert out@JavaMap[1->1, id=26a1515a]] => TensorChannel

>>> ExecutionStage[T[TensorflowCollectionSource[convert out@JavaMap[1->1, id=7a7522b9]]]]:
> In  CollectionChannel => TensorflowCollectionSource[convert out@JavaMap[1->1, id=7a7522b9]]
> Out TensorflowCollectionSource[convert out@JavaMap[1->1, id=7a7522b9]] => TensorChannel

>>> ExecutionStage[T[TensorflowCollectionSource[convert out@JavaMap[1->1, id=72754682]]]]:
> In  CollectionChannel => TensorflowCollectionSource[convert out@JavaMap[1->1, id=72754682]]
> Out TensorflowCollectionSource[convert out@JavaMap[1->1, id=72754682]] => TensorChannel

>>> ExecutionStage[T[TensorflowPredict[2->1, id=67e6cf80]]]:
> In  CollectionChannel => TensorflowPredict[2->1, id=67e6cf80]
> In  TensorChannel => TensorflowPredict[2->1, id=67e6cf80]
>     TensorflowPredict[2->1, id=67e6cf80] => TensorChannel => TensorflowCollect[convert out@TensorflowPredict[2->1, id=67e6cf80]]
> Out TensorflowCollect[convert out@TensorflowPredict[2->1, id=67e6cf80]] => CollectionChannel

>>> ExecutionStage[T[TensorflowDLTraining[2->1, id=724a17e0]]]:
> In  TensorChannel => TensorflowDLTraining[2->1, id=724a17e0]
> In  TensorChannel => TensorflowDLTraining[2->1, id=724a17e0]
> Out TensorflowDLTraining[2->1, id=724a17e0] => CollectionChannel

>>> ExecutionStage[T[JavaMap[1->1, id=fff262d]]]:
> In  CollectionChannel => JavaMap[1->1, id=fff262d]
>     JavaMap[1->1, id=fff262d] => StreamChannel => JavaLocalCallbackSink[1->0, id=7f347d0c]

[zkaoudi]

This may be a problem in the JavaSortOperator and its comparator. Maybe there are null values being compared.

incubator-wayang/wayang-platforms/wayang-java/src/main/java/org/apache/wayang/java/operators/JavaSortOperator.java

Line 78 in 664e94f

((StreamChannel.Instance) outputs[0]).accept(((JavaChannelInstance) inputs[0]).<Type>provideStream()