[Failing Test]: JmsIOTest.testAuthenticationWithBadPassword is flaky (ActiveMQ)

[mxtymoshyk]

What happened?

JmsIOTest.testAuthenticationWithBadPassword (ActiveMQ parameter) fails intermittently in beam_PreCommit_Java_IOs_Direct on PRs that don't touch JmsIO.

Example run: https://github.com/apache/beam/actions/runs/36177335884/job/108211021658

JmsIOTest > testAuthenticationWithBadPassword[with client class class org.apache.activemq.ActiveMQConnectionFactory] FAILED
    java.lang.AssertionError:
    Expected: a string containing "User name [test_user] or password is invalid."
         but: was "peer (vm://localhost:43505?transport.transformer=jms#97) stopped."

When a login fails, the ActiveMQ broker schedules its side of the connection to stop shortly afterwards. The vm:// transport is async by default, so the security error reply can still be waiting on the client side. If the runner is short on CPU (the same job also logged java.net.SocketTimeoutException: Connect timed out), the broker side stops first. connection.start() (

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

Line 774 in 011e1b3

connection.start();

) then fails with the peer ... stopped. error instead of the auth error that the test checks for (

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

Lines 184 to 217 in 011e1b3

	private void runPipelineExpectingJmsConnectException(String innerMessage) {
	try {
	pipeline.run();
	fail();
	} catch (Exception e) {
	assertThat(Throwables.getRootCause(e).getMessage(), containsString(innerMessage));
	}
	}
	@Test
	public void testAuthenticationRequired() {
	// A topic produces one source, avoiding races between concurrent authentication failures.
	pipeline.apply(JmsIO.read().withConnectionFactory(connectionFactory).withTopic(TOPIC));
	String errorMessage =
	this.connectionFactoryClass == ActiveMQConnectionFactory.class
	? "User name [null] or password is invalid."
	: "Client failed to authenticate using SASL: ANONYMOUS";
	runPipelineExpectingJmsConnectException(errorMessage);
	}
	@Test
	public void testAuthenticationWithBadPassword() {
	pipeline.apply(
	JmsIO.read()
	.withConnectionFactory(connectionFactory)
	.withTopic(TOPIC)
	.withUsername(USERNAME)
	.withPassword("BAD"));
	String errorMessage =
	this.connectionFactoryClass == ActiveMQConnectionFactory.class
	? "User name [" + USERNAME + "] or password is invalid."
	: "Client failed to authenticate using SASL: PLAIN";
	runPipelineExpectingJmsConnectException(errorMessage);

).

Proposed fix, either of:

• Make the test connections use a synchronous vm:// transport (?async=false), so the error reply is delivered before the broker stops the connection.
• Relax the assertion to accept the stopped. message for ActiveMQ.

Earlier deflake: #39920 switched the auth tests to a topic. This race is still there.

Issue Failure

Failure: Test is flaky

Issue Priority

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

Issue Components

• Component: Java SDK
• Component: IO connector

[mxtymoshyk]

.take-issue

[mxtymoshyk]

.take-issue