[controller] Remove topic-based concurrent push detection and clarify push-blocking messaging

internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java

@@ -343,7 +343,6 @@ private ConfigKeys() {
    */
   public static final String DEFAULT_READ_STRATEGY = "default.read.strategy";
   public static final String DEFAULT_OFFLINE_PUSH_STRATEGY = "default.offline.push.strategy";
-  public static final String CONCURRENT_PUSH_DETECTION_STRATEGY = "concurrent.push.detection.strategy";
 
   public static final String DEFAULT_ROUTING_STRATEGY = "default.routing.strategy";
   public static final String DEFAULT_REPLICA_FACTOR = "default.replica.factor";

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/controller/TestParentControllerWithMultiDataCenter.java

@@ -1301,7 +1301,7 @@ public void testCompliancePushCannotKillAnotherCompliancePush() {
           "Second compliance push should fail because system pushes cannot kill other system pushes");
       Assert.assertTrue(
           compliancePushResponse2.getError().contains("An ongoing push") && compliancePushResponse2.getError()
-              .contains("is found and it must be terminated before another push can be started"),
+              .contains("is still in progress and must complete or be terminated before another push can be started"),
           "Error should indicate an ongoing push must be terminated: " + compliancePushResponse2.getError());
     }
   }
@@ -1334,7 +1334,7 @@ public void testCompliancePushCannotKillUserPush() {
           "Compliance push should fail because system pushes cannot kill user pushes");
       Assert.assertTrue(
           compliancePushResponse.getError().contains("An ongoing push") && compliancePushResponse.getError()
-              .contains("is found and it must be terminated before another push can be started"),
+              .contains("is still in progress and must complete or be terminated before another push can be started"),
           "Error should indicate an ongoing push must be terminated: " + compliancePushResponse.getError());
     }
   }

services/venice-controller/src/main/java/com/linkedin/venice/controller/DeferredVersionSwapService.java

@@ -14,7 +14,6 @@
 import com.linkedin.venice.exceptions.VeniceNoClusterException;
 import com.linkedin.venice.hooks.StoreLifecycleHooks;
 import com.linkedin.venice.hooks.StoreVersionLifecycleEventOutcome;
-import com.linkedin.venice.meta.ConcurrentPushDetectionStrategy;
 import com.linkedin.venice.meta.LifecycleHooksRecord;
 import com.linkedin.venice.meta.ReadWriteStoreRepository;
 import com.linkedin.venice.meta.Store;
@@ -1278,17 +1277,6 @@ public void updateStore(String clusterName, String storeName, VersionStatus stat
       store.updateVersionStatus(targetVersionNum, status);
       if (status == ONLINE || status == PARTIALLY_ONLINE) {
         store.setCurrentVersion(targetVersionNum);
-
-        // For jobs that stop polling early or for pushes that don't poll (empty push), we need to truncate the parent
-        // VT here to unblock the next push
-        String kafkaTopicName = Version.composeKafkaTopic(storeName, targetVersionNum);
-        ConcurrentPushDetectionStrategy strategy =
-            veniceControllerMultiClusterConfig.getControllerConfig(clusterName).getConcurrentPushDetectionStrategy();
-        // skip truncating if the topic was not created based on ConcurrentPushDetectionStrategy
-        if (strategy.isTopicWriteNeeded() && !veniceParentHelixAdmin.isTopicTruncated(kafkaTopicName)) {
-          LOGGER.info("Truncating parent VT for {}", kafkaTopicName);
-          veniceParentHelixAdmin.truncateKafkaTopic(Version.composeKafkaTopic(storeName, targetVersionNum));
-        }
       }
       repository.updateStore(store);
     } catch (Exception e) {

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceController.java

@@ -15,7 +15,6 @@
 import com.linkedin.venice.controller.grpc.server.interceptor.ControllerGrpcSslSessionInterceptor;
 import com.linkedin.venice.controller.grpc.server.interceptor.ParentControllerRegionValidationInterceptor;
 import com.linkedin.venice.controller.kafka.TopicCleanupService;
-import com.linkedin.venice.controller.kafka.TopicCleanupServiceForParentController;
 import com.linkedin.venice.controller.lingeringjob.HeartbeatCheckerOtelMetricEntity;
 import com.linkedin.venice.controller.server.AdminSparkServer;
 import com.linkedin.venice.controller.server.VeniceControllerGrpcServiceImpl;
@@ -294,23 +293,12 @@ AdminSparkServer createAdminServer(boolean secure) {
 
   private TopicCleanupService createTopicCleanupService() {
     Admin admin = controllerService.getVeniceHelixAdmin();
-
-    if (multiClusterConfigs.isParent()) {
-      // TODO: Remove the following once ConcurrentPushDetectionStrategy.PARENT_VERSION_STATUS_ONLY is fully rolled out
-      return new TopicCleanupServiceForParentController(
-          admin,
-          multiClusterConfigs,
-          pubSubTopicRepository,
-          new TopicCleanupServiceStats(metricsRepository),
-          pubSubClientsFactory);
-    } else {
-      return new TopicCleanupService(
-          admin,
-          multiClusterConfigs,
-          pubSubTopicRepository,
-          new TopicCleanupServiceStats(metricsRepository),
-          pubSubClientsFactory);
-    }
+    return new TopicCleanupService(
+        admin,
+        multiClusterConfigs,
+        pubSubTopicRepository,
+        new TopicCleanupServiceStats(metricsRepository),
+        pubSubClientsFactory);
   }
 
   private Optional<StoreBackupVersionCleanupService> createStoreBackupVersionCleanupService() {

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceControllerClusterConfig.java

@@ -27,7 +27,6 @@
 import static com.linkedin.venice.ConfigKeys.CLUSTER_TO_D2;
 import static com.linkedin.venice.ConfigKeys.CLUSTER_TO_SERVER_D2;
 import static com.linkedin.venice.ConfigKeys.CONCURRENT_INIT_ROUTINES_ENABLED;
-import static com.linkedin.venice.ConfigKeys.CONCURRENT_PUSH_DETECTION_STRATEGY;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_ADMIN_GRPC_PORT;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_ADMIN_SECURE_GRPC_PORT;
 import static com.linkedin.venice.ConfigKeys.CONTROLLER_AUTO_MATERIALIZE_DAVINCI_PUSH_STATUS_SYSTEM_STORE;
@@ -249,7 +248,6 @@
 import com.linkedin.venice.controllerapi.ControllerRoute;
 import com.linkedin.venice.exceptions.ConfigurationException;
 import com.linkedin.venice.exceptions.VeniceException;
-import com.linkedin.venice.meta.ConcurrentPushDetectionStrategy;
 import com.linkedin.venice.meta.OfflinePushStrategy;
 import com.linkedin.venice.meta.PersistenceType;
 import com.linkedin.venice.meta.ReadStrategy;
@@ -512,7 +510,6 @@ public class VeniceControllerClusterConfig {
   private final PersistenceType persistenceType;
   private final ReadStrategy readStrategy;
 
-  private final ConcurrentPushDetectionStrategy concurrentPushDetectionStrategy;
   private final OfflinePushStrategy offlinePushStrategy;
   private final RoutingStrategy routingStrategy;
   private final int replicationFactor;
@@ -785,13 +782,6 @@ public VeniceControllerClusterConfig(VeniceProperties props) {
     this.isSkipHybridStoreRTTopicCompactionPolicyUpdateEnabled =
         props.getBoolean(SKIP_HYBRID_STORE_RT_TOPIC_COMPACTION_POLICY_UPDATE_ENABLED, false);
 
-    if (props.containsKey(CONCURRENT_PUSH_DETECTION_STRATEGY)) {
-      this.concurrentPushDetectionStrategy =
-          ConcurrentPushDetectionStrategy.valueOf(props.getString(CONCURRENT_PUSH_DETECTION_STRATEGY));
-    } else {
-      this.concurrentPushDetectionStrategy = ConcurrentPushDetectionStrategy.DUAL;
-    }
-
     if (props.containsKey(DEFAULT_READ_STRATEGY)) {
       this.readStrategy = ReadStrategy.valueOf(props.getString(DEFAULT_READ_STRATEGY));
     } else {
@@ -1472,10 +1462,6 @@ public ReadStrategy getReadStrategy() {
     return readStrategy;
   }
 
-  public ConcurrentPushDetectionStrategy getConcurrentPushDetectionStrategy() {
-    return concurrentPushDetectionStrategy;
-  }
-
   public OfflinePushStrategy getOfflinePushStrategy() {
     return offlinePushStrategy;
   }

services/venice-controller/src/main/java/com/linkedin/venice/controller/VeniceHelixAdmin.java

@@ -3115,7 +3115,7 @@ private Pair<Boolean, Version> addVersion(
                 version.getNumber(),
                 store.getStoreLifecycleHooks());
             long createBatchTopicStartTime = System.currentTimeMillis();
-            if (clusterConfig.getConcurrentPushDetectionStrategy().isTopicWriteNeeded() || !isParent()) {
+            if (!isParent()) {
               topicToCreationTime.computeIfAbsent(version.kafkaTopicName(), topic -> System.currentTimeMillis());
               createBatchTopics(
                   version,
@@ -4121,18 +4121,6 @@ public void deleteOneStoreVersion(String clusterName, String storeName, int vers
     deleteOneStoreVersion(clusterName, storeName, versionNumber, false);
   }
 
-  /**
-   * Check if we should skip truncating topic. If it's parent fabrics and the topic write is NOT needed, return true;
-   * Otherwise, return false.
-   * @param clusterName the cluster name to check
-   * @return true if topic truncation should be skipped, false otherwise
-   */
-  public boolean shouldSkipTruncatingTopic(String clusterName) {
-    return isParent() && !getMultiClusterConfigs().getControllerConfig(clusterName)
-        .getConcurrentPushDetectionStrategy()
-        .isTopicWriteNeeded();
-  }
-
   private void deleteOneStoreVersion(String clusterName, String storeName, int versionNumber, boolean isForcedDelete) {
     HelixVeniceClusterResources resources = getHelixVeniceClusterResources(clusterName);
     try (AutoCloseableLock ignore = resources.getClusterLockManager().createStoreWriteLock(storeName)) {
@@ -4169,8 +4157,8 @@ private void deleteOneStoreVersion(String clusterName, String storeName, int ver
           // Not using deletedVersion.get().kafkaTopicName() because it's incorrect for Zk shared stores.
           String versionTopicName = Version.composeKafkaTopic(storeName, deletedVersion.get().getNumber());
 
-          // skip truncating topic if it's parent controller and topic write is not needed
-          if (!shouldSkipTruncatingTopic(clusterName)) {
+          // skip truncating topic if it's a parent controller (parent controllers do not write version topics)
+          if (!isParent()) {
             if (fatalDataValidationFailureRetentionMs != -1 && hasFatalDataValidationError) {
               truncateKafkaTopic(versionTopicName, fatalDataValidationFailureRetentionMs);
             } else {