tkaymak commented on code in PR #39253:
URL: https://github.com/apache/beam/pull/39253#discussion_r3579861124
##########
sdks/java/io/jms/src/test/java/org/apache/beam/sdk/io/jms/JmsIOIT.java:
##########
@@ -145,84 +149,145 @@ public interface JmsIOITOptions extends
IOTestPipelineOptions, StreamingOptions
@Parameterized.Parameters(name = "with client class {3}")
public static Collection<Object[]> connectionFactories() {
return ImmutableList.of(
- new Object[] {
- "vm://localhost", 5672, "jms.sendAcksAsync=false",
ActiveMQConnectionFactory.class
- });
- // TODO(https://github.com/apache/beam/issues/26175) Test failure on
direct runner due to
- // JmsIO read on amqp slow on CI (passed locally)
- // new Object[] {
- // "amqp://localhost", 5672, "jms.forceAsyncAcks=false",
JmsConnectionFactory.class
- // });
+ new Object[] {"vm://localhost", "jms.sendAcksAsync=false",
ActiveMQConnectionFactory.class},
+ new Object[] {"amqp://localhost", "jms.forceAsyncAcks=false",
JmsConnectionFactory.class});
}
- private final CommonJms commonJms;
+ private static final Map<String, CommonJms> BROKERS = new
ConcurrentHashMap<>();
+
+ private final String brokerUrl;
+ private final Integer brokerPort;
+ private final String forceAsyncAcksParam;
+ private final Class<? extends ConnectionFactory> connectionFactoryClassParam;
+ private CommonJms commonJms;
private ConnectionFactory connectionFactory;
private Class<? extends ConnectionFactory> connectionFactoryClass;
public JmsIOIT(
String brokerUrl,
- Integer brokerPort,
String forceAsyncAcksParam,
Class<? extends ConnectionFactory> connectionFactoryClass) {
- this.commonJms =
- new CommonJms(
- OPTIONS.isLocalJmsBrokerEnabled() ? brokerUrl :
OPTIONS.getJmsBrokerHost(),
- OPTIONS.isLocalJmsBrokerEnabled() ? brokerPort :
OPTIONS.getJmsBrokerPort(),
- forceAsyncAcksParam,
- connectionFactoryClass);
+ this.brokerUrl = brokerUrl;
+ if (OPTIONS.isLocalJmsBrokerEnabled()) {
+ try {
+ this.brokerPort = NetworkTestHelper.getAvailableLocalPort();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to find available port", e);
+ }
+ } else {
+ this.brokerPort = OPTIONS.getJmsBrokerPort();
+ }
+ this.forceAsyncAcksParam = forceAsyncAcksParam;
+ this.connectionFactoryClassParam = connectionFactoryClass;
}
@Before
public void setup() throws Exception {
if (OPTIONS.isLocalJmsBrokerEnabled()) {
- this.commonJms.startBroker();
- connectionFactory = this.commonJms.createConnectionFactory();
- connectionFactoryClass = this.commonJms.getConnectionFactoryClass();
- // use a small number of record for local integration test
+ String key = brokerUrl + ":" + connectionFactoryClassParam.getName();
+ commonJms =
+ BROKERS.computeIfAbsent(
+ key,
+ k -> {
+ CommonJms broker =
+ new CommonJms(
+ brokerUrl, brokerPort, forceAsyncAcksParam,
connectionFactoryClassParam);
+ try {
+ broker.startBroker();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ return broker;
+ });
OPTIONS.setNumberOfRecords(10000);
+ } else {
+ commonJms =
+ new CommonJms(
+ OPTIONS.getJmsBrokerHost(),
+ OPTIONS.getJmsBrokerPort(),
+ forceAsyncAcksParam,
+ connectionFactoryClassParam);
}
}
+ @AfterClass
+ public static void afterClass() throws Exception {
+ for (CommonJms broker : BROKERS.values()) {
+ try {
+ broker.stopBroker();
+ } catch (Exception e) {
+ // ignore errors on shutdown
+ }
+ }
+ BROKERS.clear();
+ }
+
+ private void setupConnection(JmsIO.AcknowledgeMode acknowledgeMode) throws
Exception {
+ if (acknowledgeMode == JmsIO.AcknowledgeMode.CLIENT_ACKNOWLEDGE) {
+ connectionFactory =
this.commonJms.createConnectionFactoryWithSyncAcksAndWithoutPrefetch();
+ } else {
+ connectionFactory = this.commonJms.createConnectionFactory();
+ }
+ connectionFactoryClass = this.commonJms.getConnectionFactoryClass();
+ }
+
@After
public void tearDown() throws Exception {
- if (OPTIONS.isLocalJmsBrokerEnabled()) {
- this.commonJms.stopBroker();
- connectionFactory = null;
- connectionFactoryClass = null;
- }
+ connectionFactory = null;
+ connectionFactoryClass = null;
+ }
+
+ @Test
+ public void testPublishingThenReadingAll() throws Exception {
+ runPublishingThenReadingAll(JmsIO.AcknowledgeMode.CLIENT_ACKNOWLEDGE);
+ }
+
+ @Test
+ public void testPublishingThenReadingAllIndividualAcknowledge() throws
Exception {
+ runPublishingThenReadingAll(JmsIO.AcknowledgeMode.INDIVIDUAL_ACKNOWLEDGE);
}
@Test
- public void testPublishingThenReadingAll() throws IOException, JMSException {
- PipelineResult writeResult = publishingMessages();
+ public void testPublishingThenReadingAllClientAcknowledgeUnsafe() throws
Exception {
+
runPublishingThenReadingAll(JmsIO.AcknowledgeMode.CLIENT_ACKNOWLEDGE_UNSAFE);
+ }
+
+ private void runPublishingThenReadingAll(JmsIO.AcknowledgeMode
acknowledgeMode) throws Exception {
+ setupConnection(acknowledgeMode);
+ String queue = QUEUE + "_" + acknowledgeMode.name();
+ PipelineResult writeResult = publishingMessages(queue);
PipelineResult.State writeState = writeResult.waitUntilFinish();
assertNotEquals(PipelineResult.State.FAILED, writeState);
- PipelineResult readResult = readMessages();
- PipelineResult.State readState =
-
readResult.waitUntilFinish(Duration.standardSeconds(OPTIONS.getReadTimeout()));
- // A workaround to stop the pipeline for waiting for too long
+ PipelineResult readResult = readMessages(acknowledgeMode, queue);
+ MetricsReader metricsReader = new MetricsReader(readResult, NAMESPACE);
+ long startTime = System.currentTimeMillis();
+ long timeoutMillis = OPTIONS.getReadTimeout() * 1000L;
+ PipelineResult.State readState = readResult.getState();
Review Comment:
Outside the streaming IO scope I was tagged for, but flagging it since I
noticed it. `readState` comes from `readResult.getState()`, which on the direct
runner returns `RUNNING` and never `null`. So when `readTimeout` expires with
the pipeline still running (the amqp slow on CI case from #26175 that got this
test disabled), `cancelIfTimeouted`, which only cancels when `readState ==
null`, is a no op. The asserts then fail while the pipeline keeps consuming
from the class scoped shared `BROKERS`, leaking it into the next parameterized
test. Suggest:
```java
if (readState == null || !readState.isTerminal()) {
readResult.cancel();
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
