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).
--
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]
