[controller][common][protocol] Add store-level ULE config for RT topics

Add uncleanLeaderElectionEnabledForRTTopics as a store-level config using
the tri-state pattern (NOT_SPECIFIED/ENABLED/DISABLED). When NOT_SPECIFIED,
falls back to the cluster-level config. This enables per-store tracking of
ULE settings and preserves the setting during store migration.

Changes:
- New Avro schema versions (StoreMetaValue v41, AdminOperation v96)
- Store interface/impl (Store, ZKStore, ReadOnlyStore, SystemStore, StoreInfo)
- UpdateStoreQueryParams with migration constructor support
- Controller logic (VeniceParentHelixAdmin, AdminExecutionTask,
VeniceHelixAdmin)
- RealTimeTopicSwitcher store-level override with cluster-level fallback
- resolveUncleanLeaderElection helper for store-then-cluster resolution
- Tests for override, fallback, and resolution logic

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

Author: pthirun

build.gradle

@@ -327,8 +327,8 @@ 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/v39', PathValidation.DIRECTORY),
-          project(':services:venice-controller').file('src/main/resources/avro/AdminOperation/v94', PathValidation.DIRECTORY)
+          project(':internal:venice-common').file('src/main/resources/avro/StoreMetaValue/v41', PathValidation.DIRECTORY),
+          project(':services:venice-controller').file('src/main/resources/avro/AdminOperation/v96', PathValidation.DIRECTORY)
       ]
 
       def schemaDirs = [sourceDir]

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

@@ -268,6 +268,8 @@ public class ControllerApiConstants {
 
   public static final String BLOB_TRANSFER_ENABLED = "blob_transfer_enabled";
   public static final String BLOB_TRANSFER_IN_SERVER_ENABLED = "blob_transfer_in_server_enabled";
+  public static final String UNCLEAN_LEADER_ELECTION_ENABLED_FOR_RT_TOPICS =
+      "unclean_leader_election_enabled_for_rt_topics";
 
   public static final String HEARTBEAT_TIMESTAMP = "heartbeat_timestamp";
 

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

@@ -78,6 +78,7 @@
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.TARGET_SWAP_REGION_WAIT_TIME;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.TIME_LAG_TO_GO_ONLINE;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.TTL_REPUSH_ENABLED;
+import static com.linkedin.venice.controllerapi.ControllerApiConstants.UNCLEAN_LEADER_ELECTION_ENABLED_FOR_RT_TOPICS;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.UNUSED_SCHEMA_DELETION_ENABLED;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.UPDATED_CONFIGS_LIST;
 import static com.linkedin.venice.controllerapi.ControllerApiConstants.VERSION;
@@ -162,6 +163,8 @@ public UpdateStoreQueryParams(StoreInfo srcStore, boolean storeMigrating) {
             .setBlobTransferEnabled(srcStore.isBlobTransferEnabled())
             .setBlobTransferInServerEnabled(
                 ConfigCommonUtils.ActivationState.valueOf(srcStore.getBlobTransferInServerEnabled()))
+            .setUncleanLeaderElectionEnabledForRTTopics(
+                ConfigCommonUtils.ActivationState.valueOf(srcStore.getUncleanLeaderElectionEnabledForRTTopics()))
             .setMaxRecordSizeBytes(srcStore.getMaxRecordSizeBytes())
             .setMaxNearlineRecordSizeBytes(srcStore.getMaxNearlineRecordSizeBytes())
             .setTargetRegionSwap(srcStore.getTargetRegionSwap())
@@ -814,6 +817,14 @@ public Optional<String> getBlobTransferInServerEnabled() {
     return getString(BLOB_TRANSFER_IN_SERVER_ENABLED);
   }
 
+  public UpdateStoreQueryParams setUncleanLeaderElectionEnabledForRTTopics(ConfigCommonUtils.ActivationState state) {
+    return putString(UNCLEAN_LEADER_ELECTION_ENABLED_FOR_RT_TOPICS, state.name());
+  }
+
+  public Optional<String> getUncleanLeaderElectionEnabledForRTTopics() {
+    return getString(UNCLEAN_LEADER_ELECTION_ENABLED_FOR_RT_TOPICS);
+  }
+
   public UpdateStoreQueryParams setNearlineProducerCompressionEnabled(boolean compressionEnabled) {
     return putBoolean(NEARLINE_PRODUCER_COMPRESSION_ENABLED, compressionEnabled);
   }

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

@@ -1650,6 +1650,16 @@ public String getBlobTransferInServerEnabled() {
     return this.delegate.getBlobTransferInServerEnabled();
   }
 
+  @Override
+  public String getUncleanLeaderElectionEnabledForRTTopics() {
+    return this.delegate.getUncleanLeaderElectionEnabledForRTTopics();
+  }
+
+  @Override
+  public void setUncleanLeaderElectionEnabledForRTTopics(String uncleanLeaderElectionEnabledForRTTopics) {
+    throw new UnsupportedOperationException("Unclean leader election config is read-only");
+  }
+
   @Override
   public boolean isNearlineProducerCompressionEnabled() {
     return delegate.isNearlineProducerCompressionEnabled();

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

@@ -340,6 +340,10 @@ default IntSet getVersionNumbers() {
 
   String getBlobTransferInServerEnabled();
 
+  String getUncleanLeaderElectionEnabledForRTTopics();
+
+  void setUncleanLeaderElectionEnabledForRTTopics(String uncleanLeaderElectionEnabledForRTTopics);
+
   boolean isNearlineProducerCompressionEnabled();
 
   void setNearlineProducerCompressionEnabled(boolean compressionEnabled);

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

@@ -81,6 +81,7 @@ public static StoreInfo fromStore(Store store) {
     storeInfo.setUnusedSchemaDeletionEnabled(store.isUnusedSchemaDeletionEnabled());
     storeInfo.setBlobTransferEnabled(store.isBlobTransferEnabled());
     storeInfo.setBlobTransferInServerEnabled(store.getBlobTransferInServerEnabled());
+    storeInfo.setUncleanLeaderElectionEnabledForRTTopics(store.getUncleanLeaderElectionEnabledForRTTopics());
     storeInfo.setNearlineProducerCompressionEnabled(store.isNearlineProducerCompressionEnabled());
     storeInfo.setNearlineProducerCountPerWriter(store.getNearlineProducerCountPerWriter());
     storeInfo.setTargetRegionSwap(store.getTargetSwapRegion());
@@ -361,6 +362,7 @@ public static StoreInfo fromStore(Store store) {
 
   private boolean blobTransferEnabled;
   private String blobTransferInServerEnable = ActivationState.NOT_SPECIFIED.name();
+  private String uncleanLeaderElectionEnabledForRTTopics = ActivationState.NOT_SPECIFIED.name();
 
   private boolean nearlineProducerCompressionEnabled;
   private int nearlineProducerCountPerWriter;
@@ -900,6 +902,14 @@ public String getBlobTransferInServerEnabled() {
     return this.blobTransferInServerEnable;
   }
 
+  public void setUncleanLeaderElectionEnabledForRTTopics(String uncleanLeaderElectionEnabledForRTTopics) {
+    this.uncleanLeaderElectionEnabledForRTTopics = uncleanLeaderElectionEnabledForRTTopics;
+  }
+
+  public String getUncleanLeaderElectionEnabledForRTTopics() {
+    return this.uncleanLeaderElectionEnabledForRTTopics;
+  }
+
   public boolean isNearlineProducerCompressionEnabled() {
     return nearlineProducerCompressionEnabled;
   }

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

@@ -707,6 +707,16 @@ public String getBlobTransferInServerEnabled() {
     return zkSharedStore.getBlobTransferInServerEnabled();
   }
 
+  @Override
+  public String getUncleanLeaderElectionEnabledForRTTopics() {
+    return zkSharedStore.getUncleanLeaderElectionEnabledForRTTopics();
+  }
+
+  @Override
+  public void setUncleanLeaderElectionEnabledForRTTopics(String uncleanLeaderElectionEnabledForRTTopics) {
+    throwUnsupportedOperationException("setUncleanLeaderElectionEnabledForRTTopics is not supported in SystemStore");
+  }
+
   @Override
   public void setMaxCompactionLagSeconds(long maxCompactionLagSeconds) {
     throwUnsupportedOperationException("setMaxCompactionLagSeconds");

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

@@ -976,6 +976,16 @@ public String getBlobTransferInServerEnabled() {
     return this.storeProperties.blobTransferInServerEnabled.toString();
   }
 
+  @Override
+  public String getUncleanLeaderElectionEnabledForRTTopics() {
+    return this.storeProperties.uncleanLeaderElectionEnabledForRTTopics.toString();
+  }
+
+  @Override
+  public void setUncleanLeaderElectionEnabledForRTTopics(String uncleanLeaderElectionEnabledForRTTopics) {
+    this.storeProperties.uncleanLeaderElectionEnabledForRTTopics = uncleanLeaderElectionEnabledForRTTopics;
+  }
+
   @Override
   public boolean isNearlineProducerCompressionEnabled() {
     return this.storeProperties.nearlineProducerCompressionEnabled;

internal/venice-common/src/main/java/com/linkedin/venice/serialization/avro/AvroProtocolDefinition.java

@@ -77,7 +77,7 @@ public enum AvroProtocolDefinition {
    *
    * TODO: Move AdminOperation to venice-common module so that we can properly reference it here.
    */
-  ADMIN_OPERATION(94, SpecificData.get().getSchema(ByteBuffer.class), "AdminOperation"),
+  ADMIN_OPERATION(96, SpecificData.get().getSchema(ByteBuffer.class), "AdminOperation"),
 
   /**
    * Single chunk of a large multi-chunk value. Just a bunch of bytes.
@@ -148,7 +148,7 @@ public enum AvroProtocolDefinition {
   /**
    * Value schema for metadata system store.
    */
-  METADATA_SYSTEM_SCHEMA_STORE(39, StoreMetaValue.class),
+  METADATA_SYSTEM_SCHEMA_STORE(41, StoreMetaValue.class),
 
   /*
     Value Schema for Parent Controller Metadata system store