Add CDC-bootstrap-aware dedup logic to SourceBasedDeduper by mittalprince · Pull Request #1012 · linkedin/brooklin

mittalprince · 2026-05-16T19:24:26Z

Summary

SourceBasedDeduper: Added two time-window guards to prevent unsafe cross-type dedup between CDC-only and CDC+BST streams:
- newStreamGracePeriodMs: blocks a new CDC-only stream from deduping against a CDC+BST stream still in its catch-up grace period. The CDC+BST stream's offsets are behind the live tail — deduping would cause the CDC-only stream to miss events.
- snapshotWorstRefreshDays: blocks a new CDC+BST stream from deduping against a CDC-only stream created within the worst-case GPFS snapshot staleness window (4 days). The CDC-only stream's offset
  history would not reach back far enough for the CDC+BST stream's required earlier start position.
SourceBasedDeduperFactory: Pass deduperProperties through to the SourceBasedDeduper constructor so domain-sliced connector config reaches the deduper correctly.

Both windows default to 0 (disabled), preserving existing behavior for all non-CDC connectors. When system.cdcBootstrapRequired is absent on the new stream the legacy source-equality path is taken
unchanged.

Testing Done

Add two time-window guards to prevent unsafe cross-type dedup: - newStreamGracePeriodMs: blocks CDC-only from deduping against a CDC+BST stream still catching up on lag (grace period active) - snapshotWorstRefreshDays: blocks CDC+BST from deduping against a CDC-only stream too recent to cover the GPFS snapshot window Both default to 0 (disabled), preserving existing behavior. Also pass deduperProperties through SourceBasedDeduperFactory.

akshayrai · 2026-05-18T05:59:23Z

+        .filter(d -> d.getSource().equals(newStream.getSource()))
+        .collect(Collectors.toList());
+
+    List<Datastream> cdcOnlyCandidates = sameSourceCandidates.stream()


To be precise - Since legacy datastreams do not have this metadata and we are defaulting to false now, this list could potentially have bootstrap enabled streams as well. But I think that is ok since those streams would have already caught up. Right?

Akshay I don't think that should be the case bootstrap stream can't be candidate for cdc dedup. Here allStreams coming as input to deduper are only the streams with same connectorName only. So a BST stream (e.g. EspressoBootstrap) will never appear as a candidate when deduping a CDC stream (e.g. Espresso). Coordinator already filters allDatastreams by connectorName before passing them to the deduper.

sorry, I meant to say bootstrap enabled cdc streams and not bootstrap stream.

yes Akshay, those old created stream already caught up the lag and good to dedup.

akshayrai · 2026-05-18T06:45:51Z

+   * @return {@code true} if the window has expired, {@code false} otherwise
+   */
+  private static boolean hasWindowExpired(Datastream stream, long windowMs) {
+    if (!stream.hasMetadata() || stream.getMetadata() == null


nit: I think it will be safer to return true when windowMs == 0 regardless of creation_ms metadata present or not. Suppose some legacy datastream doesn't have this config.

Akshay, I didn’t fully understand your thought process here. This flow executes only when the system.cdcBootstrapRequired flag is set. If that condition is true, then windowMs should definitely exist.
Otherwise, returning true would mean the stream can always be deduped, which shouldn’t be the case in every scenario.

- Use VerifiableProperties.getLong instead of Long.parseLong for config - Use Math.multiplyExact for snapshotWorstRefreshMs calculation - Return true from hasWindowExpired when windowMs == 0 regardless of metadata - Remove redundant getMetadata() == null checks (hasMetadata() covers this) - Fix hasWindowExpired Javadoc to match actual >= semantics

Move the system.cdcBootstrapRequired string literal to a shared constant in DatastreamMetadataConstants so internal consumers can reference it without duplicating the string. SourceBasedDeduper now uses it as the default value for its config key.

Define CDC_START_POSITION_EARLIEST and CDC_START_POSITION_LATEST as shared constants so internal consumers can reference them without duplicating the string literals.

akshayrai · 2026-05-18T08:31:18Z

+  public static final String CDC_START_POSITION_EARLIEST = "earliest";
+
+  /**
+   * Start position value indicating the consumer should begin from the latest offset.
+   */
+  public static final String CDC_START_POSITION_LATEST = "latest";


these 2 don't really fit in this class since they are not brooklin system properties. Is there any other place we can put them?

Added in KafkaDatastreamMetadataConstants

akshayrai · 2026-05-18T08:42:19Z

+        .filter(d -> d.getSource().equals(newStream.getSource()))
+        .collect(Collectors.toList());
+
+    List<Datastream> cdcOnlyCandidates = sameSourceCandidates.stream()


sorry, I meant to say bootstrap enabled cdc streams and not bootstrap stream.

earliest/latest are Kafka-specific values, not Brooklin system properties, so they belong in KafkaDatastreamMetadataConstants rather than DatastreamMetadataConstants.

akshayrai

Thanks for the quick turn-around.

kanishkjaiswal2015 · 2026-05-18T08:57:37Z

+    }
+    if (!stream.hasMetadata()
+        || !stream.getMetadata().containsKey(DatastreamMetadataConstants.CREATION_MS)) {
+      return false;


so for streams that dont have this metadata, we would not dedupe against it? Should this be other way around as they would mostly be older streams that probably would have enough overlap with the source topic?

I have checked there are no such streams where creationtime.ms field doesn't exist. I have checked in all prod cluster for tidb & espresso.

kanishkjaiswal2015 · 2026-05-18T08:58:28Z

+   * @return {@code true} if the window has expired, {@code false} otherwise
+   */
+  private static boolean hasWindowExpired(Datastream stream, long windowMs) {
+    if (windowMs == 0) {


is this condition ever true?

mittalprince marked this pull request as ready for review May 16, 2026 19:24

mittalprince force-pushed the pmittal/datapipes-33203-cdc-bootstrap-sla-dedup branch from 91923a5 to 0f06d67 Compare May 16, 2026 19:42

akshayrai reviewed May 18, 2026

View reviewed changes

mittalprince force-pushed the pmittal/datapipes-33203-cdc-bootstrap-sla-dedup branch from e00abaa to 78fc111 Compare May 18, 2026 08:04

pmittal added 2 commits May 18, 2026 13:39

Add CDC start position constants to DatastreamMetadataConstants

55c3da6

Define CDC_START_POSITION_EARLIEST and CDC_START_POSITION_LATEST as shared constants so internal consumers can reference them without duplicating the string literals.

akshayrai reviewed May 18, 2026

View reviewed changes

Move CDC start position constants to KafkaDatastreamMetadataConstants

ab3f8ab

earliest/latest are Kafka-specific values, not Brooklin system properties, so they belong in KafkaDatastreamMetadataConstants rather than DatastreamMetadataConstants.

akshayrai approved these changes May 18, 2026

View reviewed changes

kanishkjaiswal2015 reviewed May 18, 2026

View reviewed changes

kanishkjaiswal2015 approved these changes May 18, 2026

View reviewed changes

mittalprince merged commit 86b6c98 into master May 19, 2026
1 check passed

mittalprince mentioned this pull request May 19, 2026

Releasing OSS version 5.5.17 #1013

Merged

Conversation

mittalprince commented May 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Testing Done

Uh oh!

akshayrai May 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

akshayrai left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mittalprince commented May 16, 2026 •

edited

Loading

akshayrai May 18, 2026 •

edited

Loading