Abacn commented on code in PR #39253:
URL: https://github.com/apache/beam/pull/39253#discussion_r3596343550
##########
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:
I agree a race is theoretically possible. Jms spec's requirement for session
operations running in single thread and Beam's asynchronous checkpointing
resulted in this fundamental incompatibility. recreate session in
CLIENT_ACKNOWLEDGE tried to workaround it
* CLIENT_ACKNOWLEDGE_UNSAFE
As noted previously, this mode is intended to restore the same behavior and
timing prior to Beam 2.50.0, regardless race exists or not, based on the users
report that certain Jms providers work with Beam<2.50.0 but not later.
* INDIVIDUAL_ACKNOWLEDGE is an open question. I did more investigation
public doc emphasizing "session not threadsafe" or "one thread per session"
is referring to receive:
> The JMS contract is that only 1 session is used by one thread at once -
which if you’re using consumers means that only 1 consumer can receive messages
at once if using the same session.
(see
https://activemq.apache.org/components/classic/documentation/multiple-consumers-on-a-queue)
but none saying acknowledge and receive must be in same thread. I think AI
generated answer based on mixing up concepts in the available doc
If prompt is specific, here is an answer I got
> Q: In Jms, Active MQ INDIVIDUAL_ACKNOELEDGE mode, is there a race if
receive new message and acknowledge old message happens concurrently
>
> A: No, there is no race condition between receiving a new message and
acknowledging an old message in INDIVIDUAL_ACKNOWLEDGE mode. ActiveMQ tracks
and deletes messages on the broker based on the specific message object you
call acknowledge() on.
--
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]
