Early-Results Soak-Test Failure

[gurbuzali]

java.util.concurrent.CompletionException: com.hazelcast.jet.JetException: Exception in ProcessorTasklet{EarlyResultsTest/verification#1}: java.lang.AssertionError: expected: 0, actual: 100
at java.util.concurrent.CompletableFuture.reportJoin(CompletableFuture.java:375)
at java.util.concurrent.CompletableFuture.join(CompletableFuture.java:1947)
at com.hazelcast.jet.Job.join(Job.java:155)
at com.hazelcast.jet.tests.earlyresults.EarlyResultsTest.test(EarlyResultsTest.java:69)
at com.hazelcast.jet.tests.common.AbstractSoakTest.testInternal(AbstractSoakTest.java:144)
at com.hazelcast.jet.tests.common.AbstractSoakTest.run(AbstractSoakTest.java:83)
at com.hazelcast.jet.tests.earlyresults.EarlyResultsTest.main(EarlyResultsTest.java:50)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at com.hazelcast.jet.impl.JetBootstrap.executeJar(JetBootstrap.java:126)
at com.hazelcast.jet.server.JetCommandLine.submit(JetCommandLine.java:218)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at com.hazelcast.jet.picocli.CommandLine.execute(CommandLine.java:1189)
at com.hazelcast.jet.picocli.CommandLine.access$800(CommandLine.java:141)
at com.hazelcast.jet.picocli.CommandLine$RunAll.handle(CommandLine.java:1416)
at com.hazelcast.jet.picocli.CommandLine$RunAll.handle(CommandLine.java:1376)
at com.hazelcast.jet.picocli.CommandLine$AbstractParseResultHandler.handleParseResult(CommandLine.java:1243)
at com.hazelcast.jet.picocli.CommandLine.parseWithHandlers(CommandLine.java:1526)
at com.hazelcast.jet.server.JetCommandLine.runCommandLine(JetCommandLine.java:162)
at com.hazelcast.jet.server.JetCommandLine.main(JetCommandLine.java:139)
Caused by: com.hazelcast.jet.JetException: Exception in ProcessorTasklet{EarlyResultsTest/verification#1}: java.lang.AssertionError: expected: 0, actual: 100
at com.hazelcast.jet.impl.execution.TaskletExecutionService$BlockingWorker.run(TaskletExecutionService.java:304)
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:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.AssertionError: expected: 0, actual: 100
at com.hazelcast.jet.tests.common.AbstractSoakTest.assertEquals(AbstractSoakTest.java:228)
at com.hazelcast.jet.tests.common.AbstractSoakTest.assertEquals(AbstractSoakTest.java:223)
at com.hazelcast.jet.tests.earlyresults.EarlyResultsTest.access$200(EarlyResultsTest.java:39)
at com.hazelcast.jet.tests.earlyresults.EarlyResultsTest$VerificationContext.verify(EarlyResultsTest.java:123)
at com.hazelcast.function.BiConsumerEx.accept(BiConsumerEx.java:47)
at com.hazelcast.jet.impl.connector.WriteBufferedP.lambda$new$0(WriteBufferedP.java:55)
at com.hazelcast.jet.core.Inbox.drain(Inbox.java:134)
at com.hazelcast.jet.impl.connector.WriteBufferedP.process(WriteBufferedP.java:70)
at com.hazelcast.jet.impl.processor.ProcessorWrapper.process(ProcessorWrapper.java:93)
at com.hazelcast.jet.impl.pipeline.FunctionAdapter$AdaptingProcessor.process(FunctionAdapter.java:226)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.processInbox(ProcessorTasklet.java:416)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.stateMachineStep(ProcessorTasklet.java:291)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.stateMachineStep(ProcessorTasklet.java:286)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.stateMachineStep(ProcessorTasklet.java:268)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.call(ProcessorTasklet.java:242)
at com.hazelcast.jet.impl.execution.TaskletExecutionService$BlockingWorker.run(TaskletExecutionService.java:293)
... 5 more
20:21:42.714 [ERROR] [c.h.j.t.e.EarlyResultsTest] hz.client_1 [jet-isolated] [4.3] [4.0.3] com.hazelcast.jet.JetException: Exception in ProcessorTasklet{EarlyResultsTest/verification#1}: java.lang.AssertionError: expected: 0, actual: 100
java.util.concurrent.CompletionException: com.hazelcast.jet.JetException: Exception in ProcessorTasklet{EarlyResultsTest/verification#1}: java.lang.AssertionError: expected: 0, actual: 100
at java.util.concurrent.CompletableFuture.reportJoin(CompletableFuture.java:375)
at java.util.concurrent.CompletableFuture.join(CompletableFuture.java:1947)
at com.hazelcast.jet.Job.join(Job.java:155)
at com.hazelcast.jet.tests.earlyresults.EarlyResultsTest.test(EarlyResultsTest.java:69)
at com.hazelcast.jet.tests.common.AbstractSoakTest.testInternal(AbstractSoakTest.java:144)
at com.hazelcast.jet.tests.common.AbstractSoakTest.run(AbstractSoakTest.java:83)
at com.hazelcast.jet.tests.earlyresults.EarlyResultsTest.main(EarlyResultsTest.java:50)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at com.hazelcast.jet.impl.JetBootstrap.executeJar(JetBootstrap.java:126)
at com.hazelcast.jet.server.JetCommandLine.submit(JetCommandLine.java:218)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at com.hazelcast.jet.picocli.CommandLine.execute(CommandLine.java:1189)
at com.hazelcast.jet.picocli.CommandLine.access$800(CommandLine.java:141)
at com.hazelcast.jet.picocli.CommandLine$RunAll.handle(CommandLine.java:1416)
at com.hazelcast.jet.picocli.CommandLine$RunAll.handle(CommandLine.java:1376)
at com.hazelcast.jet.picocli.CommandLine$AbstractParseResultHandler.handleParseResult(CommandLine.java:1243)
at com.hazelcast.jet.picocli.CommandLine.parseWithHandlers(CommandLine.java:1526)
at com.hazelcast.jet.server.JetCommandLine.runCommandLine(JetCommandLine.java:162)
at com.hazelcast.jet.server.JetCommandLine.main(JetCommandLine.java:139)
Caused by: com.hazelcast.jet.JetException: Exception in ProcessorTasklet{EarlyResultsTest/verification#1}: java.lang.AssertionError: expected: 0, actual: 100
at com.hazelcast.jet.impl.execution.TaskletExecutionService$BlockingWorker.run(TaskletExecutionService.java:304)
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:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.AssertionError: expected: 0, actual: 100
at com.hazelcast.jet.tests.common.AbstractSoakTest.assertEquals(AbstractSoakTest.java:228)
at com.hazelcast.jet.tests.common.AbstractSoakTest.assertEquals(AbstractSoakTest.java:223)
at com.hazelcast.jet.tests.earlyresults.EarlyResultsTest.access$200(EarlyResultsTest.java:39)
at com.hazelcast.jet.tests.earlyresults.EarlyResultsTest$VerificationContext.verify(EarlyResultsTest.java:123)
at com.hazelcast.function.BiConsumerEx.accept(BiConsumerEx.java:47)
at com.hazelcast.jet.impl.connector.WriteBufferedP.lambda$new$0(WriteBufferedP.java:55)
at com.hazelcast.jet.core.Inbox.drain(Inbox.java:134)
at com.hazelcast.jet.impl.connector.WriteBufferedP.process(WriteBufferedP.java:70)
at com.hazelcast.jet.impl.processor.ProcessorWrapper.process(ProcessorWrapper.java:93)
at com.hazelcast.jet.impl.pipeline.FunctionAdapter$AdaptingProcessor.process(FunctionAdapter.java:226)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.processInbox(ProcessorTasklet.java:416)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.stateMachineStep(ProcessorTasklet.java:291)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.stateMachineStep(ProcessorTasklet.java:286)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.stateMachineStep(ProcessorTasklet.java:268)
at com.hazelcast.jet.impl.execution.ProcessorTasklet.call(ProcessorTasklet.java:242)
at com.hazelcast.jet.impl.execution.TaskletExecutionService$BlockingWorker.run(TaskletExecutionService.java:293)
... 5 more
20:21:42.714 [ INFO] [c.h.j.t.e.EarlyResultsTest] hz.client_1 [jet-isolated] [4.3] [4.0.3] Finished with failure at test

[ufukyilmaz]

The cause of the error is that we encounter either early (most likely) or final results of the second window without seeing the final result of the first window in the verification sink. When I look at the pipeline, I couldn't find a place where these window results will be reordered. Although I run the test many times with different time parameters, I did not reproduce the error on my own computer. The pipeline of the job is below:

private Pipeline pipeline() {
        Pipeline p = Pipeline.create();
        int windowSizeLocal = windowSize; // windowSize = 100
        Sink<KeyedWindowResult<String, Long>> verificationSink = SinkBuilder
                .sinkBuilder("verification", c -> new VerificationContext(c.logger(), windowSizeLocal))
                .receiveFn(VerificationContext::verify)
                .build();
        // 20 * 3170 trade per second
        StreamStage<Map.Entry<String, Long>> sourceStage = p.readFrom(TradeGenerator.tradeSource(tradePerSecond))
                                                            .withNativeTimestamps(0)
                                                            .setName("Stream from EarlyResult-TradeGenerator");

        sourceStage.groupingKey(Map.Entry::getKey)
                    // earlyResultsPeriod= 5/3 sec. (3 early results per window)
                   .window(WindowDefinition.tumbling(windowSize).setEarlyResultsPeriod(earlyResultsPeriod))
                   .aggregate(AggregateOperations.counting())
                   .writeTo(verificationSink);
        return p;
} 

[mtopolnik]

For the record, the above "20 trades per second" is actually 20*3170 trades per wall-clock second. The event time proceeds by 1 millisecond every 3170 trades.