[controller] Add rewindEpochTimeInSecondsOverride to pass absolute epoch

timestamps through the pipeline

Pass absolute epoch rewind timestamps directly through the Venice pipeline
instead of converting epoch→relative→epoch which introduces drift. This
unblocks EOP hybrid stores from using epoch-based rewind and eliminates
time-of-use vs time-of-check drift for SOP stores.

Changes:
- Add rewindEpochTimeInSecondsOverride field to StoreMetaValue (v42) and
  AdminOperation (v96) Avro schemas
- Thread the new parameter through VPJ → ControllerClient → CreateVersion →
  Admin → VeniceHelixAdmin → VeniceParentHelixAdmin → AdminExecutionTask
- Store epoch override on HybridStoreConfig (interface + impl + ReadOnly)
- Use epoch override directly in RealTimeTopicSwitcher.getRewindStartTime()
  when set (>=0), bypassing relative→epoch conversion
- Default -1 sentinel preserves backward compatibility with existing callers

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: pthirun

build.gradle

@@ -327,7 +327,7 @@ subprojects {
       // when actually using the new protocol. Example to pin KME to v12 when introducing v13:
       // project(':internal:venice-common').file('src/main/resources/avro/KafkaMessageEnvelope/v12', PathValidation.DIRECTORY)
       def versionOverrides = [
-        project(':internal:venice-common').file('src/main/resources/avro/StoreMetaValue/v40', PathValidation.DIRECTORY)
+        project(':internal:venice-common').file('src/main/resources/avro/StoreMetaValue/v42', PathValidation.DIRECTORY)
       ]
 
       def schemaDirs = [sourceDir]

clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/PushJobSetting.java

@@ -53,6 +53,7 @@ public class PushJobSetting implements Serializable {
   public String kafkaInputTopic;
   public int repushSourceVersion;
   public long rewindTimeInSecondsOverride;
+  public long rewindEpochTimeInSecondsOverride;
   public boolean pushToSeparateRealtimeTopicEnabled;
   public boolean kafkaInputCombinerEnabled;
   public boolean kafkaInputBuildNewDictEnabled;

clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/VenicePushJob.java

@@ -126,7 +126,6 @@
 import com.linkedin.venice.jobs.ComputeJob;
 import com.linkedin.venice.jobs.DataWriterComputeJob;
 import com.linkedin.venice.message.KafkaKey;
-import com.linkedin.venice.meta.BufferReplayPolicy;
 import com.linkedin.venice.meta.HybridStoreConfig;
 import com.linkedin.venice.meta.Store;
 import com.linkedin.venice.meta.StoreInfo;
@@ -507,6 +506,7 @@ private PushJobSetting getPushJobSetting(VeniceProperties props) {
     pushJobSettingToReturn.inputURI = pushJobSettingToReturn.isSourceKafka ? "" : getInputURI(props);
     pushJobSettingToReturn.storeName = props.getString(VENICE_STORE_NAME_PROP);
     pushJobSettingToReturn.rewindTimeInSecondsOverride = props.getLong(REWIND_TIME_IN_SECONDS_OVERRIDE, NOT_SET);
+    pushJobSettingToReturn.rewindEpochTimeInSecondsOverride = NOT_SET;
 
     // If we didn't specify a rewind time
     if (pushJobSettingToReturn.rewindTimeInSecondsOverride == NOT_SET) {
@@ -523,14 +523,17 @@ private PushJobSetting getPushJobSetting(VeniceProperties props) {
                   REWIND_EPOCH_TIME_IN_SECONDS_OVERRIDE,
                   REWIND_EPOCH_TIME_IN_SECONDS_OVERRIDE));
         }
-        // Set the rewindTimeInSecondsOverride to be the time that is now - the provided timestamp so that we rewind
-        // from start of push to the provided timestamp with some extra buffer time since things aren't perfectly
-        // instantaneous
+
+        // Store epoch directly for controllers that understand it
+        pushJobSettingToReturn.rewindEpochTimeInSecondsOverride = rewindTimestamp;
+
+        // Compute relative fallback for old controllers that don't understand the epoch field
         long bufferTime = props.getLong(REWIND_EPOCH_TIME_BUFFER_IN_SECONDS_OVERRIDE, 60);
         pushJobSettingToReturn.rewindTimeInSecondsOverride = (nowInSeconds - rewindTimestamp) + bufferTime;
-        // In order for this config to make sense to the user, the remote rewind policy needs to be validated to be
-        // REWIND_FROM_SOP
-        pushJobSettingToReturn.validateRemoteReplayPolicy = BufferReplayPolicy.REWIND_FROM_SOP;
+
+        // With epoch passthrough, the SOP restriction is no longer needed because the epoch
+        // flows directly and works with any replay policy. The relative fallback is still sent
+        // for backward compatibility with old controllers.
       }
     }
 
@@ -2418,6 +2421,7 @@ void createNewStoreVersion(
             Optional.ofNullable(setting.sourceGridFabric),
             setting.livenessHeartbeatEnabled,
             setting.rewindTimeInSecondsOverride,
+            setting.rewindEpochTimeInSecondsOverride,
             setting.deferVersionSwap,
             setting.targetedRegions,
             pushJobSetting.repushSourceVersion,

clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/TestVenicePushJobCheckpoints.java

@@ -730,6 +730,7 @@ private void configureControllerClientMock(
             any(),
             anyBoolean(),
             anyLong(),
+            anyLong(),
             anyBoolean(),
             any(),
             anyInt(),

clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/VenicePushJobTest.java

@@ -1737,6 +1737,7 @@ private VersionCreationResponse mockVersionCreationResponse(ControllerClient cli
               any(),
               anyBoolean(),
               anyLong(),
+              anyLong(),
               anyBoolean(),
               any(),
               anyInt(),

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/ControllerApiConstants.java

@@ -170,6 +170,8 @@ public class ControllerApiConstants {
 
   public static final String REWIND_TIME_IN_SECONDS_OVERRIDE = "rewind_time_in_seconds_override";
 
+  public static final String REWIND_EPOCH_TIME_IN_SECONDS_OVERRIDE = "rewind_epoch_time_in_seconds_override";
+
   public static final String DEFER_VERSION_SWAP = "defer_version_swap";
 
   public static final String REPUSH_SOURCE_VERSION = "repush_source_version";

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/ControllerClient.java

@@ -54,6 +54,7 @@
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.REPLICATION_METADATA_VERSION_ID;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.REPUSH_SOURCE_VERSION;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.REPUSH_TTL_SECONDS;
+import static com.linkedin.venice.controllerapi.ControllerApiConstants.REWIND_EPOCH_TIME_IN_SECONDS_OVERRIDE;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.REWIND_TIME_IN_SECONDS_OVERRIDE;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.SCHEMA_COMPAT_TYPE;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.SCHEMA_ID;
@@ -321,6 +322,7 @@ public VersionCreationResponse requestTopicForWrites(
         sourceGridFabric,
         batchJobHeartbeatEnabled,
         rewindTimeInSecondsOverride,
+        -1,
         false,
         null,
         -1,
@@ -355,6 +357,7 @@ public VersionCreationResponse requestTopicForWrites(
         sourceGridFabric,
         batchJobHeartbeatEnabled,
         rewindTimeInSecondsOverride,
+        -1,
         deferVersionSwap,
         null,
         -1,
@@ -401,6 +404,7 @@ public VersionCreationResponse requestTopicForWrites(
       Optional<String> sourceGridFabric,
       boolean batchJobHeartbeatEnabled,
       long rewindTimeInSecondsOverride,
+      long rewindEpochTimeInSecondsOverride,
       boolean deferVersionSwap,
       String targetedRegions,
       int repushSourceVersion,
@@ -419,6 +423,7 @@ public VersionCreationResponse requestTopicForWrites(
         .add(SOURCE_GRID_FABRIC, sourceGridFabric)
         .add(BATCH_JOB_HEARTBEAT_ENABLED, batchJobHeartbeatEnabled)
         .add(REWIND_TIME_IN_SECONDS_OVERRIDE, rewindTimeInSecondsOverride)
+        .add(REWIND_EPOCH_TIME_IN_SECONDS_OVERRIDE, rewindEpochTimeInSecondsOverride)
         .add(DEFER_VERSION_SWAP, deferVersionSwap)
         .add(REPUSH_SOURCE_VERSION, repushSourceVersion)
         .add(SEPARATE_REAL_TIME_TOPIC_ENABLED, pushToSeparateRealtimeTopic)

internal/venice-common/src/main/java/com/linkedin/venice/controllerapi/RequestTopicForPushRequest.java

@@ -19,6 +19,7 @@ public class RequestTopicForPushRequest {
   private boolean isWriteComputeEnabled = false;
   private boolean separateRealTimeTopicEnabled = false;
   private long rewindTimeInSecondsOverride = -1L;
+  private long rewindEpochTimeInSecondsOverride = -1L;
   private boolean deferVersionSwap = false;
   private String targetedRegions = null;
   private int repushSourceVersion = -1;
@@ -86,6 +87,10 @@ public long getRewindTimeInSecondsOverride() {
     return rewindTimeInSecondsOverride;
   }
 
+  public long getRewindEpochTimeInSecondsOverride() {
+    return rewindEpochTimeInSecondsOverride;
+  }
+
   public boolean isDeferVersionSwap() {
     return deferVersionSwap;
   }
@@ -138,6 +143,10 @@ public void setRewindTimeInSecondsOverride(long rewindTimeInSecondsOverride) {
     this.rewindTimeInSecondsOverride = rewindTimeInSecondsOverride;
   }
 
+  public void setRewindEpochTimeInSecondsOverride(long rewindEpochTimeInSecondsOverride) {
+    this.rewindEpochTimeInSecondsOverride = rewindEpochTimeInSecondsOverride;
+  }
+
   public void setDeferVersionSwap(boolean deferVersionSwap) {
     this.deferVersionSwap = deferVersionSwap;
   }

internal/venice-common/src/main/java/com/linkedin/venice/meta/HybridStoreConfig.java

@@ -30,5 +30,9 @@ public interface HybridStoreConfig extends DataModelBackedStructure<StoreHybridC
 
   void setRealTimeTopicName(String realTimeTopicName);
 
+  long getRewindEpochTimeInSecondsOverride();
+
+  void setRewindEpochTimeInSecondsOverride(long rewindEpochTimeInSecondsOverride);
+
   HybridStoreConfig clone();
 }

internal/venice-common/src/main/java/com/linkedin/venice/meta/HybridStoreConfigImpl.java

@@ -112,6 +112,16 @@ public void setRealTimeTopicName(String realTimeTopicName) {
     this.hybridConfig.realTimeTopicName = realTimeTopicName;
   }
 
+  @Override
+  public long getRewindEpochTimeInSecondsOverride() {
+    return this.hybridConfig.rewindEpochTimeInSecondsOverride;
+  }
+
+  @Override
+  public void setRewindEpochTimeInSecondsOverride(long rewindEpochTimeInSecondsOverride) {
+    this.hybridConfig.rewindEpochTimeInSecondsOverride = rewindEpochTimeInSecondsOverride;
+  }
+
   @Override
   public StoreHybridConfig dataModel() {
     return this.hybridConfig;
@@ -136,12 +146,14 @@ public int hashCode() {
 
   @JsonIgnore
   public HybridStoreConfig clone() {
-    return new HybridStoreConfigImpl(
+    HybridStoreConfigImpl cloned = new HybridStoreConfigImpl(
         getRewindTimeInSeconds(),
         getOffsetLagThresholdToGoOnline(),
         getProducerTimestampLagThresholdToGoOnlineInSeconds(),
         getDataReplicationPolicy(),
         getBufferReplayPolicy(),
         getRealTimeTopicName());
+    cloned.setRewindEpochTimeInSecondsOverride(getRewindEpochTimeInSecondsOverride());
+    return cloned;
   }
 }