jbonofre opened a new pull request, #41013:
URL: https://github.com/apache/beam/pull/41013
On runners that execute unbounded sources as splittable DoFns, such as
Dataflow Runner v2, JmsIO acknowledges only a small fraction of the messages it
reads. The rest are redelivered and processed again.
## Cause
`UnboundedSourceAsSDFRestrictionTracker.getProgress()` calls
`currentRestriction()`, which calls `getCheckpointMark()` on the live reader.
The SDK harness calls `getProgress()` on every progress request during a bundle.
For JmsIO, `getCheckpointMark()` is not a snapshot. It moves the messages
read so far into a new checkpoint, gives that checkpoint the current session
and consumer, and opens new ones. The progress code drops that checkpoint, so
it is never finalized. Only the messages read after the last progress request
get acknowledged.
The dropped checkpoints also keep their session and consumer open until the
reader's connection closes after `closeTimeout`. The broker then redelivers
their messages.
This matches what we saw on a Dataflow streaming job (CLIENT_ACKNOWLEDGE,
`withRequiresDeduping()`):
- about 1% of the messages acknowledged;
- no errors logged;
- a few hundred messages in flight per broker;
- 17 to 23 consumers per broker for 2 readers.
#40605 did not change this, because Dataflow finalizes the in-memory
checkpoint, never a restored copy. We have not yet confirmed that this job runs
on Runner v2; the DEBUG logs below will show it.
## Changes
**Core (`Read.java`):** progress reporting no longer calls
`getCheckpointMark()`. It checks for the empty reader or the maximum watermark
directly. `UnboundedReader#getCheckpointMark()` documents that the elements
read between two calls form one bundle, so it must not be called only to report
progress.
**JmsIO:**
- **Earlier checkpoints:** finalizing a checkpoint also acknowledges the
earlier checkpoints of the same reader that are not finalized yet. This is safe
because a runner commits a checkpoint only once all earlier checkpoints of the
same reader are committed (same javadoc). It fixes JmsIO on Beam versions
without the core change, and when a finalize callback is lost.
- **Resuming:** `createReader(options, checkpoint)` acknowledges the
checkpoint it resumes from, and the earlier ones, when the reader that took it
lives in this JVM. A runner resumes only from committed checkpoints; the direct
runner already finalizes a checkpoint when it resumes from it. This replaces
the per-checkpoint registry of #40605 with a per-reader one: a checkpoint now
carries its reader id and a sequence number instead of a random id. As a
result, an empty checkpoint can also acknowledge the earlier ones.
- **closeTimeout:** when the connection closes after `closeTimeout` with
checkpoints still pending, those checkpoints are dropped instead of being
acknowledged later on a closed connection. An INFO line reports how many
messages the broker will redeliver.
- **Logging:** DEBUG logs for reader creation, connection and close, and for
each checkpoint holding messages, including the first stack frames outside
JmsIO, which show whether a checkpoint ends a bundle or was taken for progress.
Finalization also logs how many messages and earlier checkpoints it
acknowledged. To enable them, use
`--sdkHarnessLogLevelOverrides={"org.apache.beam.sdk.io.jms":"DEBUG"}` on
Runner v2, or `--workerLogLevelOverrides` on the legacy Dataflow worker.
## Compatibility
Checkpoints written by released versions still decode, since
`serialVersionUID` is unchanged. They carry no reader id, so there is nothing
to acknowledge after restoring them, as before. The `id` field from #40605 was
never in a release.
## Tests
- `ReadTest`: progress no longer takes a checkpoint, while `trySplit` still
does. This test fails without the core change.
- `JmsLocalTest`:
- finalizing acknowledges earlier checkpoints, but not later ones;
- an empty checkpoint acknowledges the earlier ones;
- resuming from a restored checkpoint acknowledges it;
- `closeTimeout` drops pending checkpoints.
- `JmsIOTest` (embedded broker, ActiveMQ and Qpid clients):
- a checkpoint taken mid-bundle is acknowledged together with the next one;
- resuming from a checkpoint acknowledges 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]
