This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 6e21dba8c6d Deflake JmsIOTest checkpoint and authentication tests
(#40291)
6e21dba8c6d is described below
commit 6e21dba8c6d5f42a50fb6ef9c2af5c0b98519f36
Author: Maksym Tymoshyk <[email protected]>
AuthorDate: Fri Oct 9 16:33:32 2026 +0300
Deflake JmsIOTest checkpoint and authentication tests (#40291)
- The slow-ack proxy wrapped every receiveNoWait() result, including null,
so testCheckpointMarkSafety failed with an NPE when qpid returned no
message
on a slow runner. Only wrap non-null results.
- With prefetch disabled, reader.start() can return false the same way
advance() can. Add startWithRetry(), mirroring advanceWithRetry().
- ActiveMQ stops the connection shortly after an authentication failure.
With
the async vm:// transport that stop can overtake the error reply, so the
auth tests saw "peer ... stopped." instead of the auth error. Use a
synchronous vm:// transport for those tests.
Fixes #30225
Fixes #40289
---
.../java/org/apache/beam/sdk/io/jms/JmsIOTest.java | 37 ++++++++++++++++------
1 file changed, 27 insertions(+), 10 deletions(-)
diff --git
a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java
b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java
index c92794c21e5..bdcf27bbfa4 100644
--- a/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java
+++ b/sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOTest.java
@@ -190,10 +190,23 @@ public class JmsIOTest {
}
}
+ /**
+ * ActiveMQ stops the connection shortly after an authentication failure.
With the default async
+ * vm:// transport, that stop can overtake the error reply on a slow
machine, and the client sees
+ * "peer ... stopped." instead of the authentication error. A synchronous
transport delivers the
+ * reply first.
+ */
+ private ConnectionFactory connectionFactoryForAuthFailure() {
+ return connectionFactoryClass == ActiveMQConnectionFactory.class
+ ? new ActiveMQConnectionFactory(String.format("%s:%d?async=false",
brokerUrl, brokerPort))
+ : connectionFactory;
+ }
+
@Test
public void testAuthenticationRequired() {
// A topic produces one source, avoiding races between concurrent
authentication failures.
-
pipeline.apply(JmsIO.read().withConnectionFactory(connectionFactory).withTopic(TOPIC));
+ pipeline.apply(
+
JmsIO.read().withConnectionFactory(connectionFactoryForAuthFailure()).withTopic(TOPIC));
String errorMessage =
this.connectionFactoryClass == ActiveMQConnectionFactory.class
? "User name [null] or password is invalid."
@@ -205,7 +218,7 @@ public class JmsIOTest {
public void testAuthenticationWithBadPassword() {
pipeline.apply(
JmsIO.read()
- .withConnectionFactory(connectionFactory)
+ .withConnectionFactory(connectionFactoryForAuthFailure())
.withTopic(TOPIC)
.withUsername(USERNAME)
.withPassword("BAD"));
@@ -361,7 +374,7 @@ public class JmsIOTest {
JmsIO.UnboundedJmsReader reader = source.createReader(null, null);
// start the reader and move to the first record
- assertTrue(reader.start());
+ assertTrue(startWithRetry(reader));
Set<ByteBuffer> uniqueIds = new HashSet<>();
uniqueIds.add(ByteBuffer.wrap(reader.getCurrentRecordId()));
@@ -494,6 +507,10 @@ public class JmsIOTest {
assertEquals(100, count);
}
+ private boolean startWithRetry(UnboundedSource.UnboundedReader reader)
throws IOException {
+ return reader.start() || advanceWithRetry(reader);
+ }
+
private boolean advanceWithRetry(UnboundedSource.UnboundedReader reader)
throws IOException {
for (int attempt = 0; attempt < 10; attempt++) {
if (reader.advance()) {
@@ -517,7 +534,7 @@ public class JmsIOTest {
UnboundedJmsReader reader = setupReaderForTest();
// start the reader and move to the first record
- assertTrue(reader.start());
+ assertTrue(startWithRetry(reader));
// consume 3 messages (NB: start already consumed the first message)
for (int i = 0; i < 3; i++) {
@@ -552,7 +569,7 @@ public class JmsIOTest {
UnboundedJmsReader reader = setupReaderForTest();
// start the reader and move to the first record
- assertTrue(reader.start());
+ assertTrue(startWithRetry(reader));
// consume 2 message (NB: start already consumed the first message)
assertTrue(advanceWithRetry(reader));
@@ -585,7 +602,7 @@ public class JmsIOTest {
public void testCheckpointMarkAndFinalizeSeparatelyIndividualAcknowledge()
throws Exception {
UnboundedJmsReader reader =
setupReaderForTest(JmsIO.AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE);
- assertTrue(reader.start());
+ assertTrue(startWithRetry(reader));
assertTrue(advanceWithRetry(reader));
assertTrue(advanceWithRetry(reader));
@@ -610,7 +627,7 @@ public class JmsIOTest {
public void testCheckpointMarkAndFinalizeSeparatelyClientAcknowledgeUnsafe()
throws Exception {
UnboundedJmsReader reader =
setupReaderForTest(JmsIO.AcknowledgeMode.CLIENT_ACKNOWLEDGE_UNSAFE);
- assertTrue(reader.start());
+ assertTrue(startWithRetry(reader));
assertTrue(advanceWithRetry(reader));
assertTrue(advanceWithRetry(reader));
@@ -747,7 +764,7 @@ public class JmsIOTest {
JmsIO.UnboundedJmsReader reader =
source.createReader(PipelineOptionsFactory.create(), null);
// start the reader and move to the first record
- assertTrue(reader.start());
+ assertTrue(startWithRetry(reader));
// consume half the messages (NB: start already consumed the first message)
for (int i = 0; i < (messagesToProcess / 2) - 1; i++) {
@@ -847,7 +864,7 @@ public class JmsIOTest {
JmsIO.UnboundedJmsReader reader =
source.createReader(PipelineOptionsFactory.create(), null);
// start the reader and move to the first record
- assertTrue(reader.start());
+ assertTrue(startWithRetry(reader));
// consume 3 more messages (NB: start already consumed the first message)
for (int i = 0; i < 3; i++) {
@@ -1116,7 +1133,7 @@ public class JmsIOTest {
new Class[] {proxyInterface},
(proxy, method, args) -> {
Object result = method.invoke(target, args);
- if (method.getName().equals(methodName)) {
+ if (result != null && method.getName().equals(methodName)) {
result = resultTransformer.apply((MethodArgT) result);
}
return result;
