scwhittle commented on code in PR #40870:
URL: https://github.com/apache/beam/pull/40870#discussion_r4422033141
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/DetectNewPartitionsAction.java:
##########
@@ -155,14 +159,18 @@ private ProcessContinuation schedulePartitions(
for (Map.Entry<Timestamp, List<PartitionMetadata>> batch :
batches.entrySet()) {
numTimestampsHandledSofar++;
final Timestamp batchCreatedAt = batch.getKey();
+ if (!tracker.tryClaim(batchCreatedAt)) {
Review Comment:
I think that this breaks how restrictions are supposed to be used.
If a claim fails, none of the work associated after the previous successful
claim should be performed. See the design doc:
https://docs.google.com/document/d/1AQmx-T9XjSi1PNoEp5_L-lT0j7BkgTbmQnc6uFEMI4c/edit?tab=t.0#heading=h.3624h6581nxd
I think that the existing logic should:
1. should move the call to updateBatchToScheduled to be after tryClaim
succeeds
2. Could possibly be changed to perform claims for each timestamp as it
iterates over batches instead of the batches. And then once all timestamps are
claimed, the batch gets too lare or upon claim failure the already claimed work
could be updated to scheduled and output. That said, dataflow ensures that no
splits are performed if there hasn't yet been a single claim, so it seems the
existing logic would claim the entire first batch regardless and may make more
progress if polling partitions are slow.
--
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]
