[Failing Test]: JmsIOTest. testCheckpointMark flaky

[Abacn]

What happened?

Example run: https://github.com/apache/beam/runs/21250090063 (an PR unrelated to Jms):

java.lang.AssertionError
	at org.junit.Assert.fail(Assert.java:87)
	at org.junit.Assert.assertTrue(Assert.java:42)
	at org.junit.Assert.assertTrue(Assert.java:53)
	at org.apache.beam.sdk.io.jms.JmsIOTest.testCheckpointMark(JmsIOTest.java:463)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

Fails at here:

beam/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java

Line 463 in d5aa44c

assertTrue(reader.advance());

In fact, consumer.receiveNoWait call at

beam/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java

Line 559 in d5aa44c

Message message = this.consumer.receiveNoWait();

never guarantees a message will be returned when there are still unacked message on the server: https://stackoverflow.com/questions/36626634/does-jms-receivenowait-guarantee-message-delivery-when-messages-are-available

So there is a chance that the call returns null and fails assert

Issue Failure

Failure: Test is flaky

Issue Priority

Priority: 2 (backlog / disabled test but we think the product is healthy)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[Abacn]

The underlying cause is that there is no guarantee receiveNoWait here:

beam/sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java

Line 559 in 27f1c07

Message message = this.consumer.receiveNoWait();

return a nonnull value when there is pending record in the server side, per JMS specification. This can also affects integration test for the same reason, as we see not all records are read within timeout.