Unordered mapUsingServiceAsync loses items when there are no watermarks [HZ-1928]

[k-jamroz]

Describe the bug

Unordered mapUsingServiceAsync loses last items when there are no watermarks.

To Reproduce

https://hazelcastcommunity.slack.com/archives/C013LM6JK7S/p1671652200501389

[k-jamroz]

It is likely that the problem was introduced in 5.2.z with multiple watermarks support.

[github-actions]

Internal Jira issue: HZ-1928

[k-jamroz]

Another problem with unordered async processor is that when the same element is processed multiple times. In such case internal counters are corrupted and processor crashes on assertion (and hangs the entire job).

Reproducer (add in AsyncTransformUsingServicePTest):

    @Test
    public void test_completedFutures_sameElementInterleavedWithWatermark() {
        TestSupport
                .verifyProcessor(getSupplier((ctx, item) ->  completedFuture(singleton(item + "-1"))))
                .hazelcastInstance(instance())
                .input(asList("a", "a", wm(10), "a"))
                .disableProgressAssertion()
                .expectOutput(asList("a-1", "a-1", wm(10), "a-1"));
    }

causes:

java.lang.AssertionError: count=-1

	at com.hazelcast.jet.impl.processor.AsyncTransformUsingServiceUnorderedP.tryFlushQueue(AsyncTransformUsingServiceUnorderedP.java:367)
	at com.hazelcast.jet.impl.processor.AsyncTransformUsingServiceUnorderedP.complete(AsyncTransformUsingServiceUnorderedP.java:261)
	at com.hazelcast.jet.core.test.TestSupport.lambda$runTest$8(TestSupport.java:768)
	at com.hazelcast.jet.core.test.TestSupport.doCall(TestSupport.java:960)
	at com.hazelcast.jet.core.test.TestSupport.runTest(TestSupport.java:768)
	at com.hazelcast.jet.core.test.TestSupport.run(TestSupport.java:632)
	at com.hazelcast.jet.core.test.TestSupport.assertOutput(TestSupport.java:423)
	at com.hazelcast.jet.core.test.TestSupport.expectOutputs(TestSupport.java:344)
	at com.hazelcast.jet.core.test.TestSupport.expectOutput(TestSupport.java:325)
	at com.hazelcast.jet.impl.processor.AsyncTransformUsingServicePTest.test_completedFutures_sameElementInterleavedWithWatermark(AsyncTransformUsingServicePTest.java:196)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104)
	at java.base/java.lang.reflect.Method.invoke(Method.java:578)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at com.hazelcast.test.FailOnTimeoutStatement$CallableStatement.call(FailOnTimeoutStatement.java:115)
	at com.hazelcast.test.FailOnTimeoutStatement$CallableStatement.call(FailOnTimeoutStatement.java:107)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
	at java.base/java.lang.Thread.run(Thread.java:1589)