tkaymak commented on code in PR #39253:
URL: https://github.com/apache/beam/pull/39253#discussion_r3579861135
##########
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]
