[controller] Add generic controller plugin framework

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

@@ -2564,6 +2564,15 @@ private ConfigKeys() {
    */
   public static final String IDENTITY_PARSER_CLASS = "identity.parser.class";
 
+  /**
+   * Comma-separated list of {@link com.linkedin.venice.controller.ControllerPlugin} implementation class names.
+   * Each class must have a public constructor taking
+   * ({@code VeniceParentHelixAdmin}, {@code AuthorizerService}, {@code VeniceControllerMultiClusterConfig}).
+   * Plugins are instantiated via reflection during {@code VeniceController} construction and started later
+   * when the controller's {@code start()} method is invoked. Only applies to parent controllers.
+   */
+  public static final String CONTROLLER_PLUGIN_CLASS_NAMES = "controller.plugin.class.names";
+
   /**
    * Specifies a list of partitioners venice supported.
    * It contains a string of concatenated partitioner class names separated by comma.

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

@@ -0,0 +1,33 @@
+package com.linkedin.venice.controller;
+
+import java.io.Closeable;
+
+
+/**
+ * A pluggable service that runs on the parent Venice controller.
+ * Implementations are provided externally (e.g., by downstream projects) and registered one of two ways:
+ * <ul>
+ *   <li>programmatically, via {@link VeniceControllerContext.Builder#setControllerPluginFactories}; or</li>
+ *   <li>by class name, via the {@code controller.plugin.class.names} config, instantiated by reflection
+ *       (the class must expose a public constructor taking {@code VeniceParentHelixAdmin},
+ *       {@code AuthorizerService}, {@code VeniceControllerMultiClusterConfig}).</li>
+ * </ul>
+ *
+ * <p>The lifecycle is:
+ * <ol>
+ *   <li>{@link ControllerPluginFactory#create} — called during controller construction</li>
+ *   <li>{@link #start()} — called when the controller starts</li>
+ *   <li>{@link #close()} — called when the controller stops</li>
+ * </ol>
+ */
+public interface ControllerPlugin extends Closeable {
+  /**
+   * Called once during controller startup. May start background threads.
+   */
+  void start();
+
+  /**
+   * Returns a human-readable name for logging.
+   */
+  String getName();
+}

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

@@ -0,0 +1,29 @@
+package com.linkedin.venice.controller;
+
+import com.linkedin.venice.authorization.AuthorizerService;
+
+
+/**
+ * Factory to create a {@link ControllerPlugin}. Receives all controller context needed to
+ * initialize the plugin: the parent admin for store enumeration and leadership checks,
+ * the authorizer service, and the multi-cluster config for reading plugin-specific properties.
+ *
+ * <p>Factories are registered via {@link VeniceControllerContext.Builder#setControllerPluginFactories}
+ * and invoked during controller construction. A factory may return {@code null} to indicate
+ * the plugin should not be created (e.g., when disabled by config).
+ */
+@FunctionalInterface
+public interface ControllerPluginFactory {
+  /**
+   * @param admin the parent controller admin
+   * @param authorizerService the controller's authorizer, or {@code null} when controller authorization is
+   *                           disabled. Implementations that require it must null-check and return {@code null}
+   *                           (skip the plugin) rather than dereferencing it.
+   * @param config the multi-cluster config, for reading plugin-specific properties
+   * @return the plugin to run, or {@code null} to skip creation
+   */
+  ControllerPlugin create(
+      VeniceParentHelixAdmin admin,
+      AuthorizerService authorizerService,
+      VeniceControllerMultiClusterConfig config);
+}

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

@@ -1,5 +1,6 @@
 package com.linkedin.venice.controller;
 
+import static com.linkedin.venice.ConfigKeys.CONTROLLER_PLUGIN_CLASS_NAMES;
 import static com.linkedin.venice.ConfigKeys.ZOOKEEPER_ADDRESS;
 
 import com.linkedin.d2.balancer.D2Client;
@@ -55,6 +56,7 @@
 import com.linkedin.venice.system.store.ControllerClientBackedSystemSchemaInitializer;
 import com.linkedin.venice.utils.LogContext;
 import com.linkedin.venice.utils.PropertyBuilder;
+import com.linkedin.venice.utils.ReflectUtils;
 import com.linkedin.venice.utils.RegionUtils;
 import com.linkedin.venice.utils.SslUtils;
 import com.linkedin.venice.utils.Utils;
@@ -150,6 +152,7 @@ public static List<Class<? extends ModuleMetricEntityInterface>> getMetricEntity
   private final PubSubPositionTypeRegistry pubSubPositionTypeRegistry;
   private final Optional<List<VeniceVersionLifecycleEventListener>> versionLifecycleEventListeners;
   private final Optional<ExternalETLService> externalETLService;
+  private final List<ControllerPlugin> controllerPlugins;
 
   /**
    * Allocates a new {@code VeniceController} object.
@@ -220,6 +223,7 @@ public VeniceController(VeniceControllerContext ctx) {
     this.versionLifecycleEventListeners = Optional.ofNullable(ctx.getVersionLifecycleEventListeners());
     this.externalETLService = Optional.ofNullable(ctx.getExternalETLService());
     this.controllerService = createControllerService();
+    this.controllerPlugins = createControllerPlugins(ctx);
     this.adminServer = createAdminServer(false);
     this.secureAdminServer = sslEnabled ? createAdminServer(true) : null;
     this.topicCleanupService = createTopicCleanupService();
@@ -267,6 +271,77 @@ private VeniceControllerService createControllerService() {
     return veniceControllerService;
   }
 
+  private List<ControllerPlugin> createControllerPlugins(VeniceControllerContext ctx) {
+    if (!multiClusterConfigs.isParent()) {
+      return Collections.emptyList();
+    }
+    Admin admin = controllerService.getVeniceHelixAdmin();
+    if (!(admin instanceof VeniceParentHelixAdmin)) {
+      return Collections.emptyList();
+    }
+    VeniceParentHelixAdmin parentAdmin = (VeniceParentHelixAdmin) admin;
+    AuthorizerService authService = ctx.getAuthorizerService();
+    List<ControllerPlugin> plugins = new ArrayList<>();
+
+    // Path 1: Programmatic factories from VeniceControllerContext
+    List<ControllerPluginFactory> factories = ctx.getControllerPluginFactories();
+    if (factories != null) {
+      for (ControllerPluginFactory factory: factories) {
+        try {
+          ControllerPlugin plugin = factory.create(parentAdmin, authService, multiClusterConfigs);
+          if (plugin != null) {
+            plugins.add(plugin);
+            LOGGER.info("Created controller plugin from factory: {}", plugin.getName());
+          }
+        } catch (Exception e) {
+          // Fail fast: a registered factory that throws is a wiring bug. Surface it at startup rather
+          // than letting the controller come up with the plugin silently absent.
+          throw new VeniceException("Failed to create controller plugin from factory.", e);
+        }
+      }
+    }
+
+    // Path 2: Reflection-based discovery from config
+    String pluginClassNames =
+        multiClusterConfigs.getCommonConfig().getProps().getString(CONTROLLER_PLUGIN_CLASS_NAMES, "");
+    if (!pluginClassNames.isEmpty()) {
+      for (String className: pluginClassNames.split(",")) {
+        className = className.trim();
+        if (className.isEmpty()) {
+          continue;
+        }
+        try {
+          Class<?> loadedClass = ReflectUtils.loadClass(className);
+          if (!ControllerPlugin.class.isAssignableFrom(loadedClass)) {
+            throw new VeniceException(
+                "Configured controller plugin class does not implement " + ControllerPlugin.class.getName() + ": "
+                    + className);
+          }
+          Class<? extends ControllerPlugin> pluginClass = loadedClass.asSubclass(ControllerPlugin.class);
+          ControllerPlugin plugin = ReflectUtils.callConstructor(
+              pluginClass,
+              new Class[] { VeniceParentHelixAdmin.class, AuthorizerService.class,
+                  VeniceControllerMultiClusterConfig.class },
+              new Object[] { parentAdmin, authService, multiClusterConfigs });
+          plugins.add(plugin);
+          LOGGER.info("Created controller plugin from config: {} ({})", plugin.getName(), className);
+        } catch (VeniceException e) {
+          // Already specific (e.g. wrong type); surface as-is without re-wrapping.
+          throw e;
+        } catch (Exception e) {
+          // Fail fast: this class name was set explicitly by an operator. A typo, missing class, or
+          // missing constructor must abort startup rather than silently no-op the configured plugin.
+          throw new VeniceException(
+              "Failed to create controller plugin from configured class: " + className + ". Check the "
+                  + CONTROLLER_PLUGIN_CLASS_NAMES + " config.",
+              e);
+        }
+      }
+    }
+
+    return plugins;
+  }
+
   AdminSparkServer createAdminServer(boolean secure) {
     return new AdminSparkServer(
         secure ? multiClusterConfigs.getAdminSecurePort() : multiClusterConfigs.getAdminPort(),
@@ -463,6 +538,17 @@ public void start() {
     systemStoreRepairService.ifPresent(AbstractVeniceService::start);
     disabledPartitionEnablerService.ifPresent(AbstractVeniceService::start);
     deferredVersionSwapService.ifPresent(AbstractVeniceService::start);
+    for (ControllerPlugin plugin: controllerPlugins) {
+      LOGGER.info("Starting controller plugin: {}", plugin.getName());
+      try {
+        plugin.start();
+      } catch (Exception e) {
+        // Fail fast: a configured plugin that cannot start is a deployment problem and must surface
+        // immediately. Abort controller startup (before service discovery registration below) rather
+        // than letting the controller come up healthy while the plugin is silently not running.
+        throw new VeniceException("Failed to start controller plugin: " + plugin.getName(), e);
+      }
+    }
     // register with service discovery at the end
     asyncRetryingServiceDiscoveryAnnouncer.register();
     if (adminGrpcServer != null) {
@@ -531,6 +617,9 @@ public void stop() {
     storeBackupVersionCleanupService.ifPresent(Utils::closeQuietlyWithErrorLogged);
     disabledPartitionEnablerService.ifPresent(Utils::closeQuietlyWithErrorLogged);
     deferredVersionSwapService.ifPresent(Utils::closeQuietlyWithErrorLogged);
+    for (ControllerPlugin plugin: controllerPlugins) {
+      Utils.closeQuietlyWithErrorLogged(plugin);
+    }
     if (adminGrpcServer != null) {
       adminGrpcServer.stop();
     }

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

@@ -38,6 +38,7 @@ public class VeniceControllerContext {
   private PubSubClientsFactory pubSubClientsFactory;
   private List<VeniceVersionLifecycleEventListener> versionLifecycleEventListeners;
   private ExternalETLService externalETLService;
+  private List<ControllerPluginFactory> controllerPluginFactories;
 
   public List<VeniceProperties> getPropertiesList() {
     return propertiesList;
@@ -91,6 +92,10 @@ public ExternalETLService getExternalETLService() {
     return externalETLService;
   }
 
+  public List<ControllerPluginFactory> getControllerPluginFactories() {
+    return controllerPluginFactories;
+  }
+
   public VeniceControllerContext(Builder builder) {
     this.propertiesList = builder.propertiesList;
     this.metricsRepository = builder.metricsRepository;
@@ -105,6 +110,7 @@ public VeniceControllerContext(Builder builder) {
     this.d2Clients = builder.d2Clients;
     this.versionLifecycleEventListeners = builder.versionLifecycleEventListeners;
     this.externalETLService = builder.externalETLService;
+    this.controllerPluginFactories = builder.controllerPluginFactories;
   }
 
   public static class Builder {
@@ -124,6 +130,7 @@ public static class Builder {
     private boolean isServiceDiscoveryAnnouncerSet;
     private List<VeniceVersionLifecycleEventListener> versionLifecycleEventListeners;
     private ExternalETLService externalETLService;
+    private List<ControllerPluginFactory> controllerPluginFactories = Collections.emptyList();
 
     public Builder setPropertiesList(List<VeniceProperties> propertiesList) {
       this.propertiesList = propertiesList;
@@ -194,6 +201,11 @@ public Builder setExternalETLService(ExternalETLService externalETLService) {
       return this;
     }
 
+    public Builder setControllerPluginFactories(List<ControllerPluginFactory> controllerPluginFactories) {
+      this.controllerPluginFactories = controllerPluginFactories;
+      return this;
+    }
+
     private void addDefaultValues() {
       if (metricsRepository == null && !isMetricsRepositorySet) {
 

services/venice-controller/src/test/java/com/linkedin/venice/controller/VeniceControllerContextTest.java

@@ -24,6 +24,7 @@ public void testVeniceServerContextCanSetDefaults() {
     assertNotNull(veniceControllerContext.getMetricsRepository());
     assertNotNull(veniceControllerContext.getServiceDiscoveryAnnouncers());
     assertEquals(veniceControllerContext.getServiceDiscoveryAnnouncers(), Collections.emptyList());
+    assertEquals(veniceControllerContext.getControllerPluginFactories(), Collections.emptyList());
   }
 
   @Test
@@ -36,6 +37,8 @@ public void testVeniceServerContextCanSetValues() {
     ClientConfig routerClientConfig = mock(ClientConfig.class);
     ICProvider icProvider = mock(ICProvider.class);
     SupersetSchemaGenerator externalSupersetSchemaGenerator = mock(SupersetSchemaGenerator.class);
+    List<ControllerPluginFactory> controllerPluginFactories =
+        Collections.singletonList(mock(ControllerPluginFactory.class));
 
     VeniceControllerContext veniceControllerContext =
         new VeniceControllerContext.Builder().setPropertiesList(propertiesList)
@@ -47,6 +50,7 @@ public void testVeniceServerContextCanSetValues() {
             .setExternalSupersetSchemaGenerator(externalSupersetSchemaGenerator)
             .setMetricsRepository(null)
             .setServiceDiscoveryAnnouncers(null)
+            .setControllerPluginFactories(controllerPluginFactories)
             .build();
 
     assertEquals(veniceControllerContext.getPropertiesList(), propertiesList);
@@ -57,5 +61,6 @@ public void testVeniceServerContextCanSetValues() {
     assertEquals(veniceControllerContext.getD2Client(), d2Client);
     assertEquals(veniceControllerContext.getRouterClientConfig(), routerClientConfig);
     assertEquals(veniceControllerContext.getIcProvider(), icProvider);
+    assertEquals(veniceControllerContext.getControllerPluginFactories(), controllerPluginFactories);
   }
 }