perf(mongodb-to-mongodb): single-stream CDC optimization, stateful deduplication, and timestamp-aware coalescing - #4298
Conversation
… order and O(1) covered seeks
…nd background CDC cursor prefetcher
…and timestamp-aware coalescing - Enforce stateful deduplication whenever CDC mutations are enabled (standalone or combined with backfill) to drop stale out-of-order mutations across change streams and retries. - Add timestamp sort key comparison in WriteBatchesFn batch coalescing so older events within an unordered batch do not overwrite newer replacements or deletes. - Add unit tests for stale insert after delete and newer insert after delete coalescing.
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 delivers comprehensive performance and correctness enhancements for the mongodb-to-mongodb Dataflow Flex Template. By introducing robust stateful deduplication, timestamp-aware batch coalescing, and optimized binary serialization, the changes significantly improve the reliability and efficiency of streaming CDC pipelines. Additionally, the inclusion of a background prefetcher for change stream cursors helps mitigate latency bottlenecks. 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
|
![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 several performance optimizations and robustness improvements to the MongoDB-to-MongoDB pipeline. Key changes include caching deduplication keys, migrating the DocumentWithMetadataCoder to deterministic binary BSON serialization, implementing a background prefetcher thread to pipeline change stream events across SDF boundaries, and optimizing type-isolated split generation with O(1) covered index scans. Additionally, bulk write coalescing now respects event timestamps to prevent stale updates. The reviewer feedback suggests closing the MongoDB cursor before joining the prefetcher thread to prevent blocking on network calls, uniquely naming prefetcher threads using an atomic counter for better debugging, and catching ArithmeticException when converting Decimal128 values to avoid potential crashes on NaN or infinite values.
| running = false; | ||
| synchronized (pollLock) { | ||
| pollLock.notifyAll(); | ||
| } | ||
| if (prefetcherThread != null) { | ||
| prefetcherThread.interrupt(); | ||
| try { | ||
| prefetcherThread.join(1000L); | ||
| } catch (InterruptedException ignored) { | ||
| Thread.currentThread().interrupt(); | ||
| } | ||
| } | ||
| eventQueue.clear(); |
There was a problem hiding this comment.
Close the cursor before calling prefetcherThread.join(1000L). Because cursor.tryNext() is a blocking network call, standard thread interruption will not unblock it. Closing the cursor first immediately closes the underlying socket, forcing tryNext() to throw an exception and allowing the prefetcher thread to exit instantly without waiting for the 1-second join timeout.
| running = false; | |
| synchronized (pollLock) { | |
| pollLock.notifyAll(); | |
| } | |
| if (prefetcherThread != null) { | |
| prefetcherThread.interrupt(); | |
| try { | |
| prefetcherThread.join(1000L); | |
| } catch (InterruptedException ignored) { | |
| Thread.currentThread().interrupt(); | |
| } | |
| } | |
| eventQueue.clear(); | |
| running = false; | |
| synchronized (pollLock) { | |
| pollLock.notifyAll(); | |
| } | |
| if (cursor != null) { | |
| try { | |
| cursor.close(); | |
| } catch (Exception ignored) { | |
| } | |
| } | |
| if (prefetcherThread != null) { | |
| prefetcherThread.interrupt(); | |
| try { | |
| prefetcherThread.join(1000L); | |
| } catch (InterruptedException ignored) { | |
| Thread.currentThread().interrupt(); | |
| } | |
| } | |
| eventQueue.clear(); |
| public static class PartitionCursorHolder implements AutoCloseable { | ||
| public static final int PREFETCH_QUEUE_CAPACITY = 4096; |
There was a problem hiding this comment.
To improve maintainability and simplify debugging, define a static AtomicInteger counter to uniquely name each background prefetcher thread. Currently, all threads are named "mongodb-cdc-prefetcher", making it difficult to distinguish them in thread dumps or logs when multiple partitions are processed concurrently.
| public static class PartitionCursorHolder implements AutoCloseable { | |
| public static final int PREFETCH_QUEUE_CAPACITY = 4096; | |
| public static class PartitionCursorHolder implements AutoCloseable { | |
| public static final int PREFETCH_QUEUE_CAPACITY = 4096; | |
| private static final java.util.concurrent.atomic.AtomicInteger THREAD_COUNTER = | |
| new java.util.concurrent.atomic.AtomicInteger(0); |
| this.prefetcherThread = new Thread(this::runPrefetchLoop, "mongodb-cdc-prefetcher"); | ||
| this.prefetcherThread.setDaemon(true); | ||
| this.prefetcherThread.start(); |
There was a problem hiding this comment.
Use the unique thread counter to name the prefetcher thread, allowing easier identification of individual partition threads in thread dumps.
| this.prefetcherThread = new Thread(this::runPrefetchLoop, "mongodb-cdc-prefetcher"); | |
| this.prefetcherThread.setDaemon(true); | |
| this.prefetcherThread.start(); | |
| this.prefetcherThread = new Thread( | |
| this::runPrefetchLoop, | |
| "mongodb-cdc-prefetcher-" + THREAD_COUNTER.getAndIncrement()); |
| if (val.isDecimal128()) { | ||
| return val.asDecimal128().getValue().doubleValue(); | ||
| try { | ||
| return val.asDecimal128().getValue().bigDecimalValue(); | ||
| } catch (NumberFormatException e) { | ||
| return BigDecimal.ZERO; | ||
| } | ||
| } |
There was a problem hiding this comment.
Catch ArithmeticException in addition to NumberFormatException when calling bigDecimalValue() on Decimal128. If the BSON value is Decimal128.NaN or infinite, bigDecimalValue() throws ArithmeticException, which would otherwise propagate and crash the split generation process.
| if (val.isDecimal128()) { | |
| return val.asDecimal128().getValue().doubleValue(); | |
| try { | |
| return val.asDecimal128().getValue().bigDecimalValue(); | |
| } catch (NumberFormatException e) { | |
| return BigDecimal.ZERO; | |
| } | |
| } | |
| if (val.isDecimal128()) { | |
| try { | |
| return val.asDecimal128().getValue().bigDecimalValue(); | |
| } catch (NumberFormatException | ArithmeticException e) { | |
| return BigDecimal.ZERO; | |
| } | |
| } |
Codecov Report❌ Patch coverage is ❌ Your patch check has failed because the patch coverage (74.40%) 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 #4298 +/- ##
============================================
+ Coverage 56.35% 56.49% +0.13%
+ Complexity 7858 7461 -397
============================================
Files 1154 1155 +1
Lines 73194 73443 +249
Branches 8580 8632 +52
============================================
+ Hits 41252 41494 +242
+ Misses 29158 29113 -45
- Partials 2784 2836 +52
🚀 New features to boost your workflow:
|
1 participant
Description
This PR provides comprehensive performance and correctness optimizations for the
mongodb-to-mongodbDataflow Flex Template under high-throughput streaming CDC workloads:1. Background CDC Cursor Prefetching & Buffering Queue
MongoDbChangeStreamReader, introducedPartitionCursorHolderwith a dedicated background prefetcher thread per change stream partition.cursor.tryNext()asynchronously into a boundedArrayBlockingQueue<ChangeStreamDocument<Document>>(PREFETCH_QUEUE_CAPACITY = 4096).getMorenetwork RPCs across Beam Splittable DoFn (SDF) slice boundaries, eliminating round-trip cursor stall and worker idle time.2. Stateful Deduplication in Streaming CDC
includeCdc) rather than restricting it to combined backfill+CDC mode.3. Timestamp-Aware Batch Coalescing
WriteBatchesFn, addedTimestampSortKeycomparison when coalescing mutations for the same document key within an unordered batch.4. Serialization & Aggregation Pipeline Optimizations
DocumentWithMetadataCoderto drastically reduce JVM object allocation and GC pressure on high-QPS streaming workers.updateDescriptionpayloads from the change stream aggregation pipeline whenfullDocumentpost-images are configured, reducing MongoDB network serialization overhead.Testing
MongoDbTransformsTest:testWriteBatchesCoalescing_staleInsertAfterDelete_preservesDeletetestWriteBatchesCoalescing_newerInsertAfterDelete_replacesWithInsertMongoDbChangeStreamReaderTest:mvn test -pl v2/mongodb-to-mongodb -Dtest=MongoDbTransformsTest: 24/24 passing (0 failures).mvn test -pl v2/mongodb-to-mongodb -Dtest=MongoDbToMongoDbTest: 15/15 passing (0 failures).mvn test -pl v2/mongodb-to-mongodb -Dtest=MongoDbChangeStreamReaderTest: All passing.