tkaymak commented on code in PR #39253:
URL: https://github.com/apache/beam/pull/39253#discussion_r3579861114
##########
sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsCheckpointMark.java:
##########
@@ -41,29 +45,32 @@ class JmsCheckpointMark implements
UnboundedSource.CheckpointMark, Serializable
private static final Logger LOG =
LoggerFactory.getLogger(JmsCheckpointMark.class);
private Instant oldestMessageTimestamp;
- private transient @Nullable Message lastMessage;
+ private transient @Nullable List<Message> messages;
private transient @Nullable MessageConsumer consumer;
private transient @Nullable Session session;
+ private transient @Nullable AtomicInteger activeCheckpoints;
private JmsCheckpointMark(
Instant oldestMessageTimestamp,
- @Nullable Message lastMessage,
+ @Nullable List<Message> messages,
@Nullable MessageConsumer consumer,
- @Nullable Session session) {
+ @Nullable Session session,
+ @Nullable AtomicInteger activeCheckpoints) {
this.oldestMessageTimestamp = oldestMessageTimestamp;
- this.lastMessage = lastMessage;
+ this.messages = messages;
this.consumer = consumer;
this.session = session;
+ this.activeCheckpoints = activeCheckpoints;
}
/** Acknowledge all outstanding message. */
@Override
public void finalizeCheckpoint() {
try {
- // Jms spec will implicitly acknowledge _all_ messaged already received
by the same
- // session if one message in this session is being acknowledged.
- if (lastMessage != null) {
- lastMessage.acknowledge();
+ if (messages != null) {
+ for (Message message : messages) {
+ message.acknowledge();
Review Comment:
Question on the two new modes when finalization runs on a separate thread.
In INDIVIDUAL_ACKNOWLEDGE and CLIENT_ACKNOWLEDGE_UNSAFE the mark holds Messages
bound to the reader's single long lived session. My understanding is that
finalizeCheckpoint() may be called on a different thread from the reader on
some runners. You would know the Dataflow specifics far better than I do. If
so, does message.acknowledge() here end up running concurrently with the
reader's receive loop on the same Session, which JMS (section 4.4.6) says must
have a single thread of control? CLIENT_ACKNOWLEDGE looks immune since each
mark owns a private session. Could you confirm how ack is serialized against
receive for the other two modes, or which providers this has been validated
against? This ties into shunping's Dataflow question above.
##########
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();
}
```
##########
sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java:
##########
@@ -783,21 +896,43 @@ public void close() {
private void doClose() {
try {
closeAutoscaler();
- closeConsumer();
- ScheduledExecutorService executorService =
- options.as(ExecutorOptions.class).getScheduledExecutorService();
- executorService.schedule(
- () -> {
- LOG.debug("Closing connection after delay {}",
source.spec.getCloseTimeout());
- // Discard the checkpoints and set the reader as inactive
- checkpointMarkPreparer.discard();
- closeSession();
- closeConnection();
- },
- source.spec.getCloseTimeout().getMillis(),
- TimeUnit.MILLISECONDS);
+ // Discard the checkpoints and set the reader as inactive
+ checkpointMarkPreparer.discard();
+ if (source.spec.getAcknowledgeMode() ==
AcknowledgeMode.CLIENT_ACKNOWLEDGE) {
+ // checkpointMark holds session in CLIENT_ACKNOWLEDGE mode. Therefore
+ // we can close consumer and session immediately.
+ closeConsumer();
+ closeSession();
+ }
+ if (activeCheckpoints.get() == 0) {
+ closeConsumer();
+ closeSession();
+ closeConnection();
+ } else {
+ ScheduledExecutorService executorService =
+ options.as(ExecutorOptions.class).getScheduledExecutorService();
+ executorService.submit(
+ () -> {
+ long startTime = System.currentTimeMillis();
+ long timeoutMillis = source.spec.getCloseTimeout().getMillis();
+ while (activeCheckpoints.get() > 0
+ && System.currentTimeMillis() - startTime < timeoutMillis)
{
+ try {
+ Thread.sleep(1_000); // poll in 1 sec interval
+ } catch (InterruptedException ignored) {
+ break;
+ }
+ }
+ LOG.debug(
+ "Closing connection after checkpoints finalized or
timeout: {}",
+ source.spec.getCloseTimeout());
+ closeConsumer();
+ closeSession();
+ closeConnection();
+ });
Review Comment:
The close path submit()s a while (...) Thread.sleep(1000) loop onto the
shared ExecutorOptions scheduled pool. On any runner where checkpoint
finalization is best effort or delayed, the count may never reach 0, so the
loop runs up to the full closeTimeout (default 60s). When many readers close at
once, for example on scale down, that parks a lot of pool threads and can
starve other scheduled work. I am not sure how often Dataflow drops
finalization in practice, you would know better, but even setting that aside
the busy wait on a shared pool seems worth avoiding. Suggest rescheduling via
schedule() with a short period, or having the last finalizeCheckpoint decrement
trigger the close when a closed flag is set (last one out), with a single
schedule(closeTimeout) fallback.
##########
sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java:
##########
@@ -484,6 +510,23 @@ public Read<T> withRequiresDeduping() {
return builder().setRequiresDeduping(true).build();
}
+ /** Specify the {@link AcknowledgeMode} used for consuming and
acknowledging JMS messages. */
+ public Read<T> withAcknowledgeMode(AcknowledgeMode acknowledgeMode) {
+ checkArgument(acknowledgeMode != null, "acknowledgeMode can not be
null");
+ return builder().setAcknowledgeMode(acknowledgeMode).build();
+ }
+
+ /**
+ * Specify the custom integer code for individual message acknowledgment
when using {@link
+ * AcknowledgeMode#INDIVIDUAL_ACKNOWLEDGE}.
+ *
+ * <p>Different JMS providers use different proprietary integer constants
for individual
+ * acknowledgment (e.g., ActiveMQ uses 4, Qpid JMS / ActiveMQ Artemis /
IBM MQ use 101).
Review Comment:
The withIndividualAcknowledgeModeCode javadoc says "Qpid JMS / ActiveMQ
Artemis / IBM MQ use 101". IBM MQ classes for JMS only support the standard
acknowledgment modes (AUTO=1, CLIENT=2, DUPS_OK=3) and have no individual
acknowledge constant, so 101 is not a valid mode there and createSession(false,
101) on IBM MQ would throw a JMSException rather than give individual ack.
Suggest dropping IBM MQ from that list. For per message behavior on IBM MQ the
guidance is CLIENT_ACKNOWLEDGE, which acks cumulatively. The "ActiveMQ uses 4"
part checks out against ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE. Minor, since
getAckModeCode correctly throws for unrecognized providers, it might be worth a
one line note that INDIVIDUAL mode on other providers requires an explicit
withIndividualAcknowledgeModeCode.
##########
sdks/java/io/jms/src/main/java/org/apache/beam/sdk/io/jms/JmsIO.java:
##########
@@ -752,16 +857,24 @@ public CheckpointMark getCheckpointMark() {
MessageConsumer consumerToClose;
Session sessionTofinalize;
+ AcknowledgeMode mode = source.spec.getAcknowledgeMode();
synchronized (this) {
- consumerToClose = consumer;
- sessionTofinalize = session;
- }
- try {
- recreateSession();
- } catch (IOException e) {
- throw new RuntimeException(e);
+ if (mode == AcknowledgeMode.CLIENT_ACKNOWLEDGE) {
+ consumerToClose = consumer;
+ sessionTofinalize = session;
+ try {
+ recreateSession();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ consumerToClose = null;
+ sessionTofinalize = null;
+ }
}
- return checkpointMarkPreparer.newCheckpoint(consumerToClose,
sessionTofinalize);
+ activeCheckpoints.incrementAndGet();
Review Comment:
activeCheckpoints.incrementAndGet() runs outside the synchronized(this)
block and unconditionally after the isEmpty() check. If doClose() or discard()
races in between, newCheckpoint takes the discarded branch and returns
emptyCheckpoint() whose activeCheckpoints is null (JmsCheckpointMark.java:265),
so the increment is never balanced and the executor loop waits the full
closeTimeout. In CLIENT_ACKNOWLEDGE, the old consumerToClose and
sessionTofinalize captured just above (after recreateSession already swapped in
a new session) are then silently dropped, leaking the old session and its
prefetched messages. Suggest moving the increment inside newCheckpoint under
the write lock on the non discarded branch, and closing the passed in consumer
and session in the discarded branch.
--
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]
