Add CDC+BST catch-up dedup hook in SourceBasedDeduper

1. SourceBasedDeduper: add protected isCdcBstStreamCaughtUp(Datastream)
hook (returns false by default) so subclasses can short-circuit the
6-hour grace window when consumer-group lag has already drained.
Update findDedupCandidateForNewCdcOnlyStream to allow dedup if either
the window has expired OR the hook returns true.
2. TestSourceBasedDeduper: add two tests covering the caught-up and
not-caught-up hook paths.

Author: mittalprince

datastream-server/src/main/java/com/linkedin/datastream/server/SourceBasedDeduper.java

@@ -62,6 +62,15 @@
  *       a new independent destination must be created. Once this window has expired, the CDC-only
  *       stream is old enough that deduping the new CDC+BST stream against it is safe.</li>
  * </ul>
+ *
+ * <h3>Early catch-up detection (subclass hook)</h3>
+ * <p>Subclasses may override {@link #isCdcBstStreamCaughtUp(Datastream)} to allow a CDC-only
+ * stream to dedup against a CDC+BST stream <em>before</em> the
+ * {@link #CONFIG_NEW_STREAM_GRACE_PERIOD_MS} window expires, when the CDC+BST stream's consumer
+ * group lag has already dropped to an acceptable level. The base implementation returns
+ * {@code false} (time-window only). Subclasses with access to a Kafka client should override
+ * this method to query consumer group lag using {@code group.id} from the stream's metadata and
+ * the broker addresses from the stream's source connection string.
  */
 public class SourceBasedDeduper extends AbstractDatastreamDeduper {
   private static final Logger LOG = LoggerFactory.getLogger(SourceBasedDeduper.class);
@@ -247,10 +256,11 @@ private Optional<Datastream> findDedupCandidateForNewCdcBootstrapStream(Datastre
    * effective offset positions are compatible.
    *
    * <ol>
-   *   <li><b>Existing CDC+BST candidate whose grace window has expired</b>
-   *       ({@code creationTime + newStreamGracePeriodMs < now}) — dedup against it.
+   *   <li><b>Existing CDC+BST candidate whose grace window has expired OR that has caught up</b>
+   *       ({@code creationTime + newStreamGracePeriodMs < now}, or
+   *       {@link #isCdcBstStreamCaughtUp(Datastream)} returns {@code true}) — dedup against it.
    *       The CDC+BST stream started from an earlier offset to catch up on historical lag, but
-   *       its grace window expiring means it has finished catching up and its consumer position
+   *       once its grace window expires or its lag drops to the threshold, its consumer position
    *       has reached the live tail. At that point the two streams' effective offsets converge
    *       and sharing a destination is safe.</li>
    *   <li><b>Existing CDC-only candidate</b> — dedup immediately, no timing check needed.
@@ -269,12 +279,12 @@ private Optional<Datastream> findDedupCandidateForNewCdcBootstrapStream(Datastre
   private Optional<Datastream> findDedupCandidateForNewCdcOnlyStream(Datastream newStream,
       List<Datastream> cdcOnlyCandidates, List<Datastream> cdcBstCandidates) {
 
-    // Step 1: prefer a CDC+BST stream whose grace window has expired (bootstrap phase is done).
+    // Step 1: prefer a CDC+BST stream whose grace window has expired, or that has already caught up.
     Optional<Datastream> eligibleCdcBstStream = cdcBstCandidates.stream()
-        .filter(d -> hasWindowExpired(d, _newStreamGracePeriodMs))
+        .filter(d -> hasWindowExpired(d, _newStreamGracePeriodMs) || isCdcBstStreamCaughtUp(d))
         .findFirst();
     if (eligibleCdcBstStream.isPresent()) {
-      LOG.info("Deduping new CDC stream {} against CDC+BST stream {} whose grace window has expired",
+      LOG.info("Deduping new CDC stream {} against CDC+BST stream {} (grace window expired or lag within threshold)",
           newStream.getName(), eligibleCdcBstStream.get().getName());
       return eligibleCdcBstStream;
     }
@@ -307,6 +317,26 @@ private boolean isCdcBootstrapRequiredStream(Datastream stream) {
     return Boolean.parseBoolean(stream.getMetadata().getOrDefault(_cdcBootstrapRequiredMetadataKey, "false"));
   }
 
+  /**
+   * Returns {@code true} if the existing CDC+BST stream has caught up to the live tail, allowing
+   * a new CDC-only stream to dedup against it before the {@link #CONFIG_NEW_STREAM_GRACE_PERIOD_MS}
+   * window expires.
+   *
+   * <p>The base implementation always returns {@code false} (time-window only). Subclasses with
+   * access to a Kafka client should override this to query consumer group lag in real time using
+   * {@code DatastreamMetadataConstants.GROUP_ID} from the stream's metadata and the broker
+   * addresses from the stream's source connection string.
+   *
+   * <p>Implementations must be defensive: return {@code false} on any error so that dedup is
+   * not incorrectly granted.
+   *
+   * @param cdcBstStream existing CDC+BST stream to check
+   * @return {@code true} if the stream has caught up to the live tail
+   */
+  protected boolean isCdcBstStreamCaughtUp(Datastream cdcBstStream) {
+    return false;
+  }
+
   /**
    * Returns {@code true} when the given stream's time window has expired, i.e.
    * {@code System.currentTimeMillis() >= creationTime + windowMs}.

datastream-server/src/test/java/com/linkedin/datastream/server/TestSourceBasedDeduper.java

@@ -260,6 +260,51 @@ public void testDefaultDedupeConfigWindowsDisabled() throws DatastreamValidation
     Assert.assertTrue(result.isPresent());
   }
 
+  // ---- isCdcBstStreamCaughtUp hook: grace window still active but hook says caught up → dedup ----
+  @Test
+  public void testCdcOnlyNewVsExistingCdcBstCaughtUpEarly() throws DatastreamValidationException {
+    Datastream newStream = generateCdcDatastream(0, false, false, -1);
+    // creationMs = now → grace window (1 hour) has NOT expired
+    Datastream candidate = generateCdcDatastream(0, true, true, System.currentTimeMillis());
+
+    // Subclass overrides hook to report stream as caught up
+    Properties props = new Properties();
+    props.setProperty(SourceBasedDeduper.CONFIG_NEW_STREAM_GRACE_PERIOD_MS, "3600000");
+    props.setProperty(SourceBasedDeduper.CONFIG_SNAPSHOT_WORST_REFRESH_DAYS, "4");
+    SourceBasedDeduper deduper = new SourceBasedDeduper(props) {
+      @Override
+      protected boolean isCdcBstStreamCaughtUp(Datastream cdcBstStream) {
+        return true;
+      }
+    };
+
+    Optional<Datastream> result = deduper.findExistingDatastream(newStream, Collections.singletonList(candidate));
+    Assert.assertTrue(result.isPresent());
+    Assert.assertEquals(result.get(), candidate);
+  }
+
+  // ---- isCdcBstStreamCaughtUp hook: grace window active and hook returns false → new destination ----
+  @Test
+  public void testCdcOnlyNewVsExistingCdcBstNotCaughtUp() throws DatastreamValidationException {
+    Datastream newStream = generateCdcDatastream(0, false, false, -1);
+    Datastream candidate = generateCdcDatastream(0, true, true, System.currentTimeMillis());
+
+    // Hook reports not caught up (same as base behaviour)
+    SourceBasedDeduper deduper;
+    Properties props = new Properties();
+    props.setProperty(SourceBasedDeduper.CONFIG_NEW_STREAM_GRACE_PERIOD_MS, "3600000");
+    props.setProperty(SourceBasedDeduper.CONFIG_SNAPSHOT_WORST_REFRESH_DAYS, "4");
+    deduper = new SourceBasedDeduper(props) {
+      @Override
+      protected boolean isCdcBstStreamCaughtUp(Datastream cdcBstStream) {
+        return false;
+      }
+    };
+
+    Optional<Datastream> result = deduper.findExistingDatastream(newStream, Collections.singletonList(candidate));
+    Assert.assertFalse(result.isPresent());
+  }
+
   // ---- Different source in CDC-aware path → no match ----
   @Test
   public void testCdcAwarePathDifferentSourceNoMatch() throws DatastreamValidationException {