Cleanup

Author: sushantmane

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/DolState.java renamed to clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/DolStamp.java

@@ -8,14 +8,14 @@
  * Tracks Declaration of Leadership (DoL) state during STANDBY to LEADER transition.
  * DoL mechanism ensures the new leader is fully caught up with VT before switching to remote VT or RT.
  */
-public class DolState {
+public class DolStamp {
   private final long leadershipTerm;
   private final String hostId;
   private volatile boolean dolProduced; // DoL message was acked by broker
   private volatile boolean dolConsumed; // DoL message was consumed back by this replica
   private volatile CompletableFuture<PubSubProduceResult> dolProduceFuture; // Future tracking DoL produce result
 
-  public DolState(long leadershipTerm, String hostId) {
+  public DolStamp(long leadershipTerm, String hostId) {
     this.leadershipTerm = leadershipTerm;
     this.hostId = hostId;
     this.dolProduced = false;
@@ -61,7 +61,7 @@ public boolean isReady() {
 
   @Override
   public String toString() {
-    return "DolState{term=" + leadershipTerm + ", host=" + hostId + ", produced=" + dolProduced + ", consumed="
+    return "DolStamp{term=" + leadershipTerm + ", host=" + hostId + ", produced=" + dolProduced + ", consumed="
         + dolConsumed + ", futureSet=" + (dolProduceFuture != null) + "}";
   }
 }

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

@@ -866,11 +866,11 @@ private void initializeAndSendDoLStamp(PartitionConsumptionState partitionConsum
       return;
     }
 
-    DolState dolState = new DolState(leadershipTerm, veniceWriter.get().getWriterId());
-    partitionConsumptionState.setDolState(dolState);
+    DolStamp dolStamp = new DolStamp(leadershipTerm, veniceWriter.get().getWriterId());
+    partitionConsumptionState.setDolState(dolStamp);
     LOGGER.info(
         "Initialized DoL state: {} for replica: {} with term: {} and hostId: {}",
-        dolState,
+        dolStamp,
         partitionConsumptionState.getReplicaId(),
         leadershipTerm,
         veniceWriter.get().getWriterId());
@@ -880,8 +880,8 @@ private void initializeAndSendDoLStamp(PartitionConsumptionState partitionConsum
     CompletableFuture<PubSubProduceResult> dolProduceFuture = veniceWriter.get()
         .sendDoLStamp(partitionConsumptionState.getReplicaTopicPartition(), dolCallback, leadershipTerm);
 
-    // Store the produce future in DolState
-    dolState.setDolProduceFuture(dolProduceFuture);
+    // Store the produce future in DolStamp
+    dolStamp.setDolProduceFuture(dolProduceFuture);
 
     // Chain logging for produce completion
     dolProduceFuture.thenAccept((result) -> {
@@ -926,21 +926,21 @@ public void onCompletion(PubSubProduceResult produceResult, Exception exception)
       }
 
       // Mark DoL as produced
-      DolState dolState = pcs.getDolState();
-      if (dolState != null && dolState.getLeadershipTerm() == leadershipTerm) {
-        dolState.setDolProduced(true);
+      DolStamp dolStamp = pcs.getDolState();
+      if (dolStamp != null && dolStamp.getLeadershipTerm() == leadershipTerm) {
+        dolStamp.setDolProduced(true);
         LOGGER.info(
-            "DoL message produce confirmed for partition {} at position {} (term: {}) - dolState: {}",
+            "DoL message produce confirmed for partition {} at position {} (term: {}) - dolStamp: {}",
             pcs.getPartition(),
             produceResult.getPubSubPosition(),
             leadershipTerm,
-            dolState);
+            dolStamp);
       } else {
         LOGGER.warn(
-            "DoL state mismatch or null for partition {} - expected term: {}, dolState: {}",
+            "DoL state mismatch or null for partition {} - expected term: {}, dolStamp: {}",
             pcs.getPartition(),
             leadershipTerm,
-            dolState);
+            dolStamp);
       }
     }
   }
@@ -954,21 +954,21 @@ public void onCompletion(PubSubProduceResult produceResult, Exception exception)
    */
   private boolean canSwitchToLeaderTopic(PartitionConsumptionState pcs) {
     // Check if DoL mechanism is enabled via config (system stores vs user stores)
-    DolState dolState = pcs.getDolState();
-    if (shouldUseDolMechanism() && dolState != null) {
+    DolStamp dolStamp = pcs.getDolState();
+    if (shouldUseDolMechanism() && dolStamp != null) {
       // Check if DoL state is ready (both produced and consumed)
-      if (dolState.isReady()) {
+      if (dolStamp.isReady()) {
         LOGGER.info(
-            "DoL mechanism complete for replica: {} - switching to leader topic. DolState: {}",
+            "DoL mechanism complete for replica: {} - switching to leader topic. DolStamp: {}",
             pcs.getReplicaId(),
-            dolState);
+            dolStamp);
         // Clear DoL state as we're done with this transition
         pcs.clearDolState();
         return true;
       }
 
       // DoL not ready yet, stay on local VT
-      LOGGER.info("DoL mechanism not ready for partition {} - DolState: {}", pcs.getReplicaId(), dolState);
+      LOGGER.info("DoL mechanism not ready for partition {} - DolStamp: {}", pcs.getReplicaId(), dolStamp);
       return false;
     } else {
       // Use legacy time-based mechanism

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

@@ -129,7 +129,7 @@ enum LatchStatus {
   /**
    * Tracks DoL state during STANDBY to LEADER transition. Null when not in transition or DoL not enabled.
    */
-  private volatile DolState dolState = null;
+  private volatile DolStamp dolStamp = null;
 
   /**
    * The highest leadership term observed by this replica. Currently used only
@@ -533,16 +533,16 @@ public final LeaderFollowerStateType getLeaderFollowerState() {
     return this.leaderFollowerState;
   }
 
-  public DolState getDolState() {
-    return this.dolState;
+  public DolStamp getDolState() {
+    return this.dolStamp;
   }
 
-  public void setDolState(DolState dolState) {
-    this.dolState = dolState;
+  public void setDolState(DolStamp dolStamp) {
+    this.dolStamp = dolStamp;
   }
 
   public void clearDolState() {
-    this.dolState = null;
+    this.dolStamp = null;
   }
 
   public long getHighestLeadershipTerm() {

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

@@ -272,7 +272,7 @@ public abstract class StoreIngestionTask implements Runnable, Closeable {
   /**
    * Keeps track of producer states inside version topic that drainer threads have processed so far.
    * Producers states in this validator will be flushed to the metadata partition of the storage engine regularly in
-   * {@link #syncOffset(String, PartitionConsumptionState)}
+   * {@link #syncOffset(PartitionConsumptionState)}
    * NOTE: consumerDiv will be used in place of this when {@link #isGlobalRtDivEnabled()} is true.
    */
   private final DataIntegrityValidator drainerDiv;
@@ -1335,7 +1335,7 @@ private int handleSingleMessage(
 
   /**
    * Checks if the consumed message is a DoL (Declaration of Leadership) stamp and marks it as consumed
-   * if it matches the current DolState. This method is called during message consumption to detect when
+   * if it matches the current DolStamp. This method is called during message consumption to detect when
    * the leader replica has successfully consumed back its own DoL stamp, indicating it's fully caught up
    * with the version topic and ready to switch to consuming from remote VT or RT.
    *
@@ -1380,8 +1380,8 @@ private void checkAndHandleDoLMessage(
     }
 
     // Get current DoL state - may be null if not in STANDBY->LEADER transition
-    DolState currentDolState = partitionConsumptionState.getDolState();
-    if (currentDolState == null) {
+    DolStamp currentDolStamp = partitionConsumptionState.getDolState();
+    if (currentDolStamp == null) {
       // Not currently waiting for a DoL, just log for observability
       LOGGER.debug(
           "Replica: {} consumed DoL stamp for term: {} from host: {} (timestamp: {}), but not currently waiting for DoL",
@@ -1393,32 +1393,32 @@ private void checkAndHandleDoLMessage(
     }
 
     // Check if this DoL matches the expected term and host
-    long expectedTermId = currentDolState.getLeadershipTerm();
-    String expectedHostId = currentDolState.getHostId();
+    long expectedTermId = currentDolStamp.getLeadershipTerm();
+    String expectedHostId = currentDolStamp.getHostId();
 
     if (consumedTermId == expectedTermId && consumedHostId.equals(expectedHostId)) {
       // Successfully consumed our own DoL stamp - mark as consumed
-      currentDolState.setDolConsumed(true);
+      currentDolStamp.setDolConsumed(true);
       LOGGER.info(
           "Replica {}: finished DoL loopback. The leader wrote its DoL stamp to the "
               + "local VT and successfully consumed it again, confirming the replica is "
-              + "fully caught up. [term={}, host={}, timestamp={}]. DolState={}",
+              + "fully caught up. [term={}, host={}, timestamp={}]. DolStamp={}",
           replicaId,
           consumedTermId,
           consumedHostId,
           messageTimestamp,
-          currentDolState);
+          currentDolStamp);
     } else {
       // Received a DoL stamp that doesn't match our expected state
       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 DolState: {}",
+              + "This may indicate a stale message or concurrent leadership changes. Current DolStamp: {}",
           replicaId,
           expectedTermId,
           expectedHostId,
           consumedTermId,
           consumedHostId,
-          currentDolState);
+          currentDolStamp);
     }
   }
 

internal/venice-common/src/main/java/com/linkedin/venice/writer/VeniceWriter.java

@@ -2243,6 +2243,12 @@ public CompletableFuture<PubSubProduceResult> sendDoLStamp(
       PubSubProducerCallback callback,
       long leadershipTerm) {
     KafkaMessageEnvelope kafkaMessageEnvelope = getDoLStampKME(leadershipTerm, writerId, heartBeatMessage);
+
+    logger.info(
+        "Sending DoL stamp message to topic-partition {} for leadership term {} kme: {}",
+        topicPartition,
+        leadershipTerm,
+        kafkaMessageEnvelope);
     return producerAdapter.sendMessage(
         topicPartition.getPubSubTopic().getName(),
         topicPartition.getPartitionNumber(),