Fix flaky test

Author: sushantmane

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java

@@ -1395,40 +1395,54 @@ private void checkAndHandleDoLMessage(
       return;
     }
 
-    // Check if this DoL matches the expected term and host
+    // Validate DoL matches expected term and host
     long expectedTermId = currentDolStamp.getLeadershipTerm();
     String expectedHostId = currentDolStamp.getHostId();
-    if (!expectedHostId.equals(consumedHostId) || consumedTermId < expectedTermId) {
-      // DoL message is either from a different host or an older term - ignore it
+
+    // Ignore DoL from different host
+    if (!expectedHostId.equals(consumedHostId)) {
+      LOGGER.debug(
+          "Replica: {} ignoring DoL from different host. Expected: {}, received: {}",
+          replicaId,
+          expectedHostId,
+          consumedHostId);
       return;
     }
 
-    if (consumedTermId == expectedTermId) {
-      // DoL (Declaration of Leadership) loopback complete - ready to switch to leader source topic
-      currentDolStamp.setDolConsumed(true);
-      long loopbackLatencyMs = currentDolStamp.getLatencyMs();
-      LOGGER.info(
-          "Replica {}: DoL loopback complete - successfully produced to and consumed back from local VT, "
-              + "confirming replica is fully caught up and ready to switch to leader source topic. "
-              + "Loopback latency: {} ms. [term={}, host={}, timestamp={}]. DolStamp={}",
+    // Ignore stale DoL from older term
+    if (consumedTermId < expectedTermId) {
+      LOGGER.debug(
+          "Replica: {} ignoring stale DoL from older term. Expected: {}, received: {}",
           replicaId,
-          loopbackLatencyMs,
-          consumedTermId,
-          consumedHostId,
-          messageTimestamp,
-          currentDolStamp);
-    } else {
-      // Received a DoL stamp that doesn't match our expected state
+          expectedTermId,
+          consumedTermId);
+      return;
+    }
+
+    // Handle DoL from future term - indicates race or concurrent leadership change
+    if (consumedTermId > expectedTermId) {
       LOGGER.warn(
-          "Replica: {} consumed DoL stamp with mismatched metadata. Expected: [term={}, host={}], Received: [term={}, host={}]. "
-              + "This may indicate a stale message or concurrent leadership changes. Current DolStamp: {}",
+          "Replica: {} consumed DoL from future term. Expected: {}, received: {}. DolStamp: {}",
           replicaId,
           expectedTermId,
-          expectedHostId,
           consumedTermId,
-          consumedHostId,
           currentDolStamp);
+      return;
     }
+
+    // DoL loopback complete - term and host match
+    currentDolStamp.setDolConsumed(true);
+    long loopbackLatencyMs = currentDolStamp.getLatencyMs();
+    LOGGER.info(
+        "Replica {}: DoL loopback complete - successfully produced to and consumed back from local VT, "
+            + "confirming replica is fully caught up and ready to switch to leader source topic. "
+            + "Loopback latency: {} ms. [term={}, host={}, timestamp={}]. DolStamp={}",
+        replicaId,
+        loopbackLatencyMs,
+        consumedTermId,
+        consumedHostId,
+        messageTimestamp,
+        currentDolStamp);
   }
 
   protected boolean shouldUseDolMechanism() {