feat(datastream-mongodb-to-firestore): implement high-throughput shadowless migration - #4128
Conversation
michaeltle-goog
commented
Aug 10, 2026
michaeltle-goog
commented
- Implement shadowless stateful deduplication and monotonic ordering via StatefulDeduplicationFn and LatestChangeEventCombineFn.
- Add asynchronous non-transactional bulk write engine with configurable rate ramp-up and multi-collection batching in MongoDbBulkTransforms.
- Implement two-tier DLQ routing (retryable vs severe permanent failure triage) with ThrottledLogger.
- Support collision-free BSON type preservation and type-tagged documentIdToString.
- Add comprehensive unit test suites covering stateful deduplication, combiner pre-compaction, bulk writing, and error classification.
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request introduces a high-throughput shadowless migration mode for the Datastream to Firestore (MongoDB compatibility) pipeline. By leveraging in-memory stateful processing and an asynchronous bulk write engine, the pipeline can now achieve significantly higher throughput while maintaining data consistency and ordering. The changes also include robust error handling with a two-tier dead-letter queue system and improved BSON type support. Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
…owless migration - Implement shadowless stateful deduplication and monotonic ordering via StatefulDeduplicationFn and LatestChangeEventCombineFn. - Add asynchronous non-transactional bulk write engine with configurable rate ramp-up and multi-collection batching in MongoDbBulkTransforms. - Implement two-tier DLQ routing (retryable vs severe permanent failure triage) with ThrottledLogger. - Support collision-free BSON type preservation and type-tagged documentIdToString. - Add comprehensive unit test suites covering stateful deduplication, combiner pre-compaction, bulk writing, and error classification.
d89466e to
a0c7bec
Compare
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a high-throughput shadowless mode to the Datastream MongoDB to Firestore template, adding support for in-memory stateful deduplication, pre-compaction, and asynchronous bulk writes with rate limiting. Key feedback highlights critical issues in the new implementation: coalesced events in MongoDbBulkTransforms are prematurely marked as successful before the bulk write is attempted, risking silent data loss; ThrottledLogger.ensureInitialized() lacks thread safety; unhandled exceptions during collection retrieval and document parsing could crash the pipeline; and missing validation checks for connectionUri and the 500-document batchSize limit for Firestore compatibility should be addressed.
| Map<Object, MongoDbChangeEventContext> latestPerDoc = new java.util.LinkedHashMap<>(); | ||
| for (MongoDbChangeEventContext event : batch) { | ||
| Object docId = event.getDocumentId(); | ||
| MongoDbChangeEventContext existing = latestPerDoc.get(docId); | ||
| if (existing == null) { | ||
| latestPerDoc.put(docId, event); | ||
| } else { | ||
| long eventTs = Utils.getTimestampNanos(event.getTimestampDoc()); | ||
| long existingTs = Utils.getTimestampNanos(existing.getTimestampDoc()); | ||
| if (eventTs > existingTs || (eventTs == existingTs && event.getIsDlqReconsumed())) { | ||
| // Count coalesced older event as resolved/superseded | ||
| successQueue.add(existing); | ||
| successfulWrites.inc(); | ||
| latestPerDoc.put(docId, event); | ||
| } else { | ||
| // Drop current event as superseded by earlier in-batch event | ||
| successQueue.add(event); | ||
| successfulWrites.inc(); | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
Coalesced/superseded events are added to successQueue and marked as successful before the bulk write is even attempted. If the bulk write fails (either transiently or permanently), these older events will have already been reported as successful, leading to silent data loss or out-of-sync states. They should only be marked as successful after the active event succeeds or is safely written to the DLQ.
Map<Object, MongoDbChangeEventContext> latestPerDoc = new java.util.LinkedHashMap<>();
Map<Object, List<MongoDbChangeEventContext>> supersededPerDoc = new java.util.HashMap<>();
for (MongoDbChangeEventContext event : batch) {
Object docId = event.getDocumentId();
MongoDbChangeEventContext existing = latestPerDoc.get(docId);
if (existing == null) {
latestPerDoc.put(docId, event);
} else {
long eventTs = Utils.getTimestampNanos(event.getTimestampDoc());
long existingTs = Utils.getTimestampNanos(existing.getTimestampDoc());
if (eventTs > existingTs || (eventTs == existingTs && event.getIsDlqReconsumed())) {
supersededPerDoc.computeIfAbsent(docId, k -> new ArrayList<>()).add(existing);
latestPerDoc.put(docId, event);
} else {
supersededPerDoc.computeIfAbsent(docId, k -> new ArrayList<>()).add(event);
}
}
}
Codecov Report❌ Patch coverage is ❌ Your patch check has failed because the patch coverage (48.88%) is below the target coverage (80.00%). You can increase the patch coverage or adjust the target coverage. Additional details and impacted files@@ Coverage Diff @@
## main #4128 +/- ##
============================================
- Coverage 56.18% 55.73% -0.46%
- Complexity 7342 7553 +211
============================================
Files 1126 1138 +12
Lines 68766 70774 +2008
Branches 7785 8128 +343
============================================
+ Hits 38637 39446 +809
- Misses 27635 28715 +1080
- Partials 2494 2613 +119
🚀 New features to boost your workflow:
|
… and address PR 4128 review comments - Disentangle CDC oplog increment and Backfill wall-clock nanoseconds comparison using composite monotonic TimestampSortKey to fix 30-orphan document discrepancy. - Defer emission of intra-batch superseded events in MongoDbBulkTransforms until bulkWrite succeeds; properly thread supersededPerDoc through retries and DLQ routing. - Add connectionUri non-null, non-empty, and protocol prefix (mongodb://, mongodb+srv://) validation in validateOptions(). - Make ThrottledLogger thread-safe with volatile fields, double-checked locking, and readObject deserialization hook. - Enclose executeBatch operation setup and execution inside try-catch with resilient DLQ fallback. - Add comprehensive unit tests covering TimestampSortKey, ThrottledLogger concurrency, URI validation, and partial bulk write failure triage.
…ss tests for 123 string vs 123L
…hecks in MongoDbChangeEventContext
…d test coverage review findings
…w bottleneck and stream directly to StatefulDeduplication
… coders to eliminate GC overhead in StatefulDeduplication
… before keyed shuffle to eliminate watermark stalls
…mestamps before keyed shuffle to eliminate watermark stalls" This reverts commit 72bdcd7.
…ram and make stateful deduplication default shadowless implementation
There was a problem hiding this comment.
Code Review
This pull request introduces a high-throughput shadowless mode for the Datastream MongoDB to Firestore template, featuring stateful deduplication (StatefulDeduplicationFn) to ensure monotonic event ordering, a custom asynchronous bulk writer (MongoDbBulkTransforms) with rate-limiting and error triage, and deterministic binary coders to eliminate serialization overhead. The review feedback provides critical improvements, including implementing state expiry via processing-time timers in StatefulDeduplicationFn to prevent memory leaks, using LONG_CODER instead of INT_CODER in TimestampSortKeyCoder to avoid sub-second truncation bugs, and adding validation for databaseName in validateOptions to ensure early pipeline failures with proper logging.
| import org.apache.beam.sdk.state.StateSpec; | ||
| import org.apache.beam.sdk.state.StateSpecs; | ||
| import org.apache.beam.sdk.state.ValueState; | ||
| import org.apache.beam.sdk.transforms.DoFn; |
There was a problem hiding this comment.
Add imports for Beam's Timer and Joda-Time's Duration to support state expiry and prevent memory leaks in streaming pipelines.
import org.apache.beam.sdk.state.StateSpec;
import org.apache.beam.sdk.state.StateSpecs;
import org.apache.beam.sdk.state.TimeDomain;
import org.apache.beam.sdk.state.Timer;
import org.apache.beam.sdk.state.TimerSpec;
import org.apache.beam.sdk.state.TimerSpecs;
import org.apache.beam.sdk.state.ValueState;
import org.apache.beam.sdk.transforms.DoFn;
import org.joda.time.Duration;| @StateId("latestTimestamp") | ||
| private final StateSpec<ValueState<TimestampSortKey>> latestTimestampSpec = | ||
| StateSpecs.value(TimestampSortKeyCoder.of()); |
There was a problem hiding this comment.
Define a processing-time timer to clear the state after a period of inactivity (e.g., 3 days) to prevent infinite state growth and memory leaks.
| @StateId("latestTimestamp") | |
| private final StateSpec<ValueState<TimestampSortKey>> latestTimestampSpec = | |
| StateSpecs.value(TimestampSortKeyCoder.of()); | |
| @StateId("latestTimestamp") | |
| private final StateSpec<ValueState<TimestampSortKey>> latestTimestampSpec = | |
| StateSpecs.value(TimestampSortKeyCoder.of()); | |
| @TimerId("expiryTimer") | |
| private final TimerSpec expiryTimerSpec = TimerSpecs.timer(TimeDomain.PROCESSING_TIME); |
…tion, and accurate DAG logs
…e TupleTags, and harden null safety
…a' field and isolate shadow metadata stripping
…mits and use _original_document in DLQ
… rate limits and use _original_document in DLQ" This reverts commit 5df3db8.
1 participant
