Fix Jms drop record

[Abacn]

Attempt to fix - #30054

The issue identified

There were two issues in the JmsIO read checkpoint implementation

1. checkpoint was a mutable member of the reader, returned to the runner each time getCheckpointMark is called:

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

Line 600 in 27f1c07

returncheckpointMark;

This is not correct as the checkpoint by definition should not change after made. Here, record after the checkpoint made still add to the same checkpoint:

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

Line 566 in 27f1c07

checkpointMark.add(message);

When checkpoint is finalized (by finalizeCheckpoint call from runner), messages read between the checkpoint made and finalizing the checkpoint incorrectly gets acknowledged

2. Jms specification of message acknowledgement

It is surprising, Jms message acknowledge isn't per message, but per session. The spec states

acknowledgment takes place on the session level: Acknowledging a consumed message automatically acknowledges the receipt of all messages that have been consumed by its session.

Even if 1 is fixed, the symptom would remain the same as messages between checkpoint being made and the checkpoint finalized would get acknowledged.

The fix

Corresponding to these two issues, two fixes are made

1. Introduce Checkpoint.Preparer class pretty much the same as the original the mutable "checkpoint". Instead, when getCheckpointMark is called, it create an actual checkpoint for later use, and the Preparer is reset.

2. Recreate session at the time checkpoint is made. In this way, the current session won't consume more data before that checkpoint is finalized.

Caveat

1. Since the message is acknowledged by finalizeCheckpoint called anytime later, the current session needs to be alive otherwise the acknowledge will fail (see diff). Therefore, the owner of current session is transferred from the Reader to the checkpoint. However, checkpoint finalization is best effort. If it does not happen, duplicate message may seen downstream.

2. reader calls consumer.receiveNoWait to get a message. However, the Jms specification does not guarantee a message will be returned (otherwise null) when there are unacknowledged messages on the server side. From the test, I find receiveNoWait of one consumer can return null when there is other consumer active (not closed). This leads to the integration test failing the original assert that all messages received by the sink when streaming pipeline timeout. There are always ~1-9 outstanding messages. Checking unacknowledged messages, there are ~10 unacknowledged. So there is no data loss here, but we have to adjust the test assertion.

On the other hand, if close the consumer immediately after the checkpoint is made (but keep the session active), data gets consumed, not acknowledged, and even the associate session was still open, will get redelivered into other sessions' consumers, causing a surge of duplication. So we still need to defer closing the consumer after acknowledge messages, along with closing the session

Please add a meaningful description for your change here

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[Abacn]

There is still a flaky test, actually also flaky on master: and filed #30225

[Abacn]

See #30226 the newly added test testCheckpointMarkAndFinalizeSeparately would fail on master

[github-actions]

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @kennknowles for label java.
R: @damondouglas for label io.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[liferoad]

Can we make the message better like this will be retried later?

[Abacn]

Once we loss a healthy session, there is no way to acknowledge these messages, and there won't be retry. So I put this log as error level.

[Abacn]

Yeah, will add ..."may cause duplication"

[liferoad]

infor or debug? Not sure the rule. :)

[Abacn]

For the latter two, as long as we successfully acknowledged the messages, there won't be duplicates for the messages in this checkpoint. So error in closing consumer or session is less harm, so I put it as info level here.

[liferoad]

update the comment?

[damondouglas]

I appreciate you listed the issues this PR addresses but what wasn't clear is which tests assert which intended behavior was previously broken but now resolved. You are closer to JMS and this Beam connector than I and feel free to ignore or dismiss if you feel this is obvious to someone who is a content expert or not feasible within unit or integration test contexts.

[damondouglas]

Feel free to dismiss if this is not feasible; I don't understand enough about JMS to know whether what I'm asking here is. Was wondering if this PR could add:

1. failing test that attempts to resolve pending messages
2. Add @Ignore annotation to test and reference a GitHub issue that aims to resolve failing test.

[Abacn]

My fix indeed broke this integration test, and I convinced myself that the cause is that the direct runner having outstanding checkpoint doesn't finalize, holding active consumer, and leading to other consumers not receiving remaining messages within the time of test. See Caveat 2 in the PR description for detailed explanation.

For

failing test that attempts to resolve pending messages

Actually, we already have such "failing test that attempts to resolve pending messages": #30225 . it's also flaky on master for same cause

Update: commented in #30225

[damondouglas]

This assertion confused me. Are the total number of records actualRecords + remainRecords? To move this PR along, feel free to resolve this comment after answering.

[Abacn]

For actualRecords (records get read in pipeline), there are two category

• A: acknowledged records
• B: unacknowledged records that read

There is a third category, that is records not get read

• C: unacknowledged records that not read

remainRecords is unacknowledged message, and we have

actualRecords=A+B
remainRecords=B+C
OPTIONS.getNumberOfRecords() =A+B+C

that was why

OPTIONS.getNumberOfRecords() <= actualRecords + remainRecords

I should rename "remainRecords" as "unackedRecords" for clarification

[Abacn]

Adjusted asserts to be more clear and added comments

[kennknowles]

Overall design SGTM. It sounds like it would have a performance impact, but is still worth it. Do you know if the recreating a session has a performance impact that could matter?

(of course anything is better than data loss)

[Abacn]

Overall design SGTM. It sounds like it would have a performance impact, but is still worth it. Do you know if the recreating a session has a performance impact that could matter?

(of course anything is better than data loss)

Yeah, this PR

• Recreate session and consumer on every getCheckpointMark , which has performance cost

However, it also

• acknowlege single message in a session (per JMS specification this is suffice), which has performance improvement

So as long as time_cost( create-close-session + create-close-consumer) comparable to N*time_cost( acknowlege a message ) , N is the number of message per checkpoint, it shouldn't have negative effect to the performance. Will do more testing of course.

[Abacn]

Just re-enable a skipped integration test to see if #26175 is still an issue for now

[Abacn]

Thanks, @damondouglas PTAL

[Abacn]

turns out we still cannot un-ignore ampq case. It can pass consistently on my machine, but it still fails on CI

[liferoad]

Great job!

[damondouglas]

@Abacn LGTM. This looks like it was really challenging to work through.

[ppawel]

@Abacn I just upgraded to Beam 2.55.0 in my project and my streaming pipeline that uses JmsIO gets stuck during consuming messages, it was working fine with Beam 2.53.0 and earlier. I tracked this down to changes in this PR related to managing JMS resources (consumers, sessions etc.)

I have a test case in my project that simply publishes around 60 messages to a queue and then a Beam pipeline is executed to consume and process those messages. This test case was passing before but now it works like this:

1. 60 messages are published to a queue in Solace (message broker we use).
2. Pipeline is started, first consumer is created in JmsIO and it fetches all 60 messages into the internal buffer of the consumer.
3. Beam pipeline consumes 10 messages (advance is called 10 times in JmsIO).
4. 10 messages are acked, checkpoint is made and recreateSession is called in JmsIO (this is the new code introduced in this PR).
5. New consumer is created, old one is closed.
6. New consumer does not get the messages anymore - the broker seems to think that the first consumer is just late with acknowledging the remaining 50 messages.
7. Pipeline hangs forever, advance is called all the time but brings no new messages. 50 messages remain sitting in the queue not consumed.

Does this scenario make sense or do I miss something in how JmsIO should work?

Btw, I tried using JmsIO with two different JMS implementations (Solace JMS client and ApacheQpid JMS client) just in case it was something to do with how the spec is implemented but they both have the same behavior in this test case above.

[Abacn]

Hi @ppawel thanks for reporting this.

In the case of 50 outstanding messages, are these messages already acknowledged in the broker side (which means data loss)? Or there are still unacknowledged but do not send to other session by the broker ?

[Abacn]

Also, which runner are you using? I may noticed this issue and noted in https://github.com/apache/beam/pull/30218/files#diff-a63812b51f93708cc60430f314b496ae1110425c6a8ae4c85e59573cfb8f0938R204

[ppawel]

@Abacn I am using GCP Dataflow but the test case is with DirectRunner.

I am now digging deeper into this and this is easily reproducible when you force the data source splits to 1 in UnboundedJmsSource#split. This means that you will have only one connection/session/consumer and there will basically be a deadlock between the first (original) consumer and the one created when doing a checkpoint.

I temporarily removed the code that creates the new consumer and also removed closing the consumer and session from JmsCheckpointMark and now it works fine as expected. So I think this is most likely related to closing/opening the session/consumer.

In the case of 50 outstanding messages, are these messages already acknowledged in the broker side (which means data loss)? Or there are still unacknowledged but do not send to other session by the broker ?

They are marked as "not acknowledged" by the broker but as "received" by one consumer. The other consumer does not get those messages.

I guess the broker would send those messages to the second consumer when the first one is closed but closing only happens when finalizing the checkpoint and that doesn't happen because only the new consumer (which does not have any messages) is being called in advance so in the end nothing is being done, it just spins around calling receive.

[Abacn]

We have to recreate the consumer/session to properly handle checkpoints, otherwise there were data losses. This is due to the limitation of Jms spec such that message acknowledge in Jms is per session. All messages that delivered by the time of acknowledging a message within a session will be marked as acknowledged. And in Beam the checkpoint is finalized asynchronously.

Agree with the analysis and I think the cause is same here: https://github.com/apache/beam/pull/30218/files#diff-a63812b51f93708cc60430f314b496ae1110425c6a8ae4c85e59573cfb8f0938R204-R207

The yet-finalized checkpoint hold an active session which may contain messages in its internal buffer. Those message won't get released until a checkpoint being finalized.

I tested my PR with IBM MQ and it had no issue. So this is still implementation related, though.

Is there a way to release the messages in internal buffer but do not acknowledge the receive message within the same session?

[ppawel]

We have to recreate the consumer/session to properly handle checkpoints, otherwise there were data losses. This is due to the limitation that message acknowledge in Jms is per session. All messages that delivered by the time of acknowledging a message within a session will be marked as acknowledged.

I've never looked so deep into JMS spec, to be honest it is a bit strange if it in fact works like this with regards to acks (that it's enough to ack one single message). I think this could also be subject to testing between implementations, I can check it at some point with my case (Solace broker) but first need to deal somehow with this deadlock situation.

Agree with the analysis and I think the cause is same here: https://github.com/apache/beam/pull/30218/files#diff-a63812b51f93708cc60430f314b496ae1110425c6a8ae4c85e59573cfb8f0938R204-R207

OK but as I understand, this is only for the direct runner, in a runner like Dataflow, finalizing the checkpoint might or might not happen at some point and "fix" the deadlock but the root cause will still be there.

Is there a way to release the messages in internal buffer but do not acknowledge the receive message within the same session?

I don't think it's possible at JMS API level to do this, and even at the implementation level I don't see any easy access to those internal queues/buffers in both clients. There are some parameters to control how big is the buffer etc. but I think JmsIO should ideally work regardless of such parameters. After all, those buffers are there for a reason (performance).

I see the comment in org.apache.beam.sdk.io.jms.JmsIOTest#testCheckpointMark regarding testing without "prefetch" - I think it would be good to also test some scenarios including prefetch and this internal buffer. I can't promise anything due to time constraints but I might try to implement a test case that reproduces my issue.

[Abacn]

Yeah, it seems authors of this IO connector were aware of prefetch in implementation affects how the message gets delivered.

Here I hava a proposal,

introduce a "checkPointTimeout" option for JmsIO.read, default to 0 (never timeout), and if works as

when the last advance return true and current advance return false passed the checkPointTimeout, then we revoke the lastly made jmscheckpoint, that is close the session without acknowledge messages in it

This will cause some duplicates, but it may unblock the mesdages sit in the internal buffer of the previous session

[Abacn]

#30853 appears working. Basically it sets a timeout after that the checkpoint is forced to be finalized. This is unsafe in general, but it indeed released the records stuck in the session internal buffer

[ppawel]

I will try it out today but with this approach I am a bit worried about performance impact. I think waiting for the timeout would mean that during every checkpoint (every 10 messages?) there would be a delay with consuming further messages.

This performance impact would be "masked" a bit if you leave the splits to the original logic instead of forcing to 1 but then there another problem with JmsIO comes out - #21075. In my case, this gets really bad - after some hours of streaming, we are reaching the limit of 100 broker connections and there really is no justification to ask for raising the limit because when I force the split to 1 (=one JMS connection), then the workload is handled just fine.

So bottom line for my case is that until the connection blowup issue is fixed somehow, I need to force it to 1. Which means I am facing the deadlock issue, and the timeout workaround will probably not be usable for me due to high performance impact.