tkaymak commented on code in PR #39253:
URL: https://github.com/apache/beam/pull/39253#discussion_r3586955042
##########
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:
Thanks, that mostly clears it up. One part I still want to nail down before
approving.
The CheckpointMark contract in UnboundedSource.java says finalizeCheckpoint
"may be called from any thread, concurrently with calls to the UnboundedReader
it was created from", and the reader is single threaded but finalize is
exempted. For INDIVIDUAL_ACKNOWLEDGE and CLIENT_ACKNOWLEDGE_UNSAFE
getCheckpointMark does not recreate the session, so the mark's messages stay
bound to the reader's live session, and finalizeCheckpoint calls
message.acknowledge() on it without taking the reader monitor.
CLIENT_ACKNOWLEDGE avoids this because it hands each checkpoint its own session.
On Dataflow, where commit finalization runs asynchronously on a pool thread
while the cached reader keeps advancing, what stops acknowledge() from racing
the reader's receiveNoWait() on the same session? Same question for the legacy
Flink UnboundedSourceWrapper, which calls finalizeCheckpoint outside the
checkpoint lock. The integration tests run on the direct runner, where finalize
is on the reader thread, so they would not exercise this.
If I am missing the mechanism that serializes them I am happy to be
corrected. Otherwise acking under the reader monitor for these two modes, or
giving them per checkpoint sessions like CLIENT_ACKNOWLEDGE, would close it.
--
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]
