[common][server][controller][pulsar][test] Fail fast (config-gated) when a pub-sub adapter factory class is not configured

clients/da-vinci-client/src/test/java/com/linkedin/davinci/config/VeniceServerConfigTest.java

@@ -28,6 +28,7 @@
 import static org.testng.Assert.assertTrue;
 
 import com.linkedin.davinci.blobtransfer.client.NettyFileTransferClient;
+import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.VeniceProperties;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -42,6 +43,7 @@
 public class VeniceServerConfigTest {
   private Properties populatedBasicProperties() {
     Properties props = new Properties();
+    props.putAll(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
     props.setProperty(CLUSTER_NAME, "test_cluster");
     props.setProperty(ZOOKEEPER_ADDRESS, "fake_zk_addr");
     props.setProperty(KAFKA_BOOTSTRAP_SERVERS, "fake_kafka_addr");

clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTaskTest.java

@@ -3298,6 +3298,7 @@ public void testPartitionExceptionIsolation(AAConfig aaConfig) throws Exception
 
   private VeniceServerConfig buildVeniceServerConfig(Map<String, Object> extraProperties) {
     PropertyBuilder propertyBuilder = new PropertyBuilder();
+    propertyBuilder.put(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
     propertyBuilder.put(CLUSTER_NAME, "");
     propertyBuilder.put(ZOOKEEPER_ADDRESS, "");
     propertyBuilder.put(SERVER_PROMOTION_TO_LEADER_REPLICA_DELAY_SECONDS, 500L);

clients/da-vinci-client/src/test/java/com/linkedin/davinci/store/AbstractStorageEngineTest.java

@@ -17,6 +17,7 @@
 import com.linkedin.venice.meta.PersistenceType;
 import com.linkedin.venice.utils.PropertyBuilder;
 import com.linkedin.venice.utils.RandomGenUtils;
+import com.linkedin.venice.utils.TestUtils;
 import com.linkedin.venice.utils.Utils;
 import com.linkedin.venice.utils.VeniceProperties;
 import java.io.File;
@@ -45,6 +46,7 @@ public static VeniceProperties getServerProperties(PersistenceType persistenceTy
         .put(LISTENER_PORT, 7072)
         .put(ADMIN_PORT, 7073)
         .put(DATA_BASE_PATH, dataDirectory.getAbsolutePath())
+        .put(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs())
         .put(properties)
         .build();
   }

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

@@ -116,6 +116,19 @@ private ConfigKeys() {
   public static final String PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS =
       PUBSUB_CLIENT_CONFIG_PREFIX + "source.of.truth.admin.adapter.factory.class";
 
+  /**
+   * Configuration key that controls whether the PubSub producer/consumer/admin adapter factories
+   * silently fall back to the Apache Kafka implementation when their factory-class config keys are
+   * not explicitly provided.
+   * <p>
+   * When {@code false} (the default), the {@code PubSubClientsFactory} fails fast by throwing an
+   * exception if the corresponding factory-class config is missing. This surfaces misconfiguration
+   * early instead of masking it behind an implicit Kafka default. Set this to {@code true} to
+   * restore the legacy behavior of defaulting to the Apache Kafka adapter factories.
+   */
+  public static final String PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED =
+      PUBSUB_CLIENT_CONFIG_PREFIX + "adapter.factory.kafka.fallback.enabled";
+
   /**
    * Configuration key for specifying the address of the PubSub broker (e.g., Kafka, Pulsar).
    * <p>

internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubClientsFactory.java

@@ -1,5 +1,6 @@
 package com.linkedin.venice.pubsub;
 
+import static com.linkedin.venice.ConfigKeys.PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED;
 import static com.linkedin.venice.ConfigKeys.PUBSUB_ADMIN_ADAPTER_FACTORY_CLASS;
 import static com.linkedin.venice.ConfigKeys.PUBSUB_CONSUMER_ADAPTER_FACTORY_CLASS;
 import static com.linkedin.venice.ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS;
@@ -29,6 +30,14 @@
 public class PubSubClientsFactory {
   private static final Logger LOGGER = LogManager.getLogger(PubSubClientsFactory.class);
 
+  /**
+   * By default the adapter factories do NOT fall back to Apache Kafka when their factory-class config
+   * is missing; callers must configure the factory classes explicitly so that misconfiguration fails
+   * fast. Set {@link com.linkedin.venice.ConfigKeys#PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED} to
+   * {@code true} to restore the legacy implicit-Kafka behavior.
+   */
+  public static final boolean DEFAULT_KAFKA_FALLBACK_ENABLED = false;
+
   private enum FactoryType {
     PRODUCER, CONSUMER, ADMIN
   }
@@ -112,6 +121,20 @@ private static <T> T createFactory(
       className = properties.getStringWithAlternative(preferredConfigKey, alternateConfigKey);
       LOGGER.debug("Creating pub-sub {} adapter factory instance for class: {}", factoryType, className);
     } else {
+      boolean kafkaFallbackEnabled =
+          properties.getBoolean(PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED, DEFAULT_KAFKA_FALLBACK_ENABLED);
+      if (!kafkaFallbackEnabled) {
+        throw new VeniceException(
+            String.format(
+                "PubSub %s adapter factory class is not configured. Set '%s' (or the legacy '%s') to the "
+                    + "fully-qualified factory class name. Implicit fallback to the Apache Kafka adapter factory "
+                    + "('%s') is disabled; set '%s=true' to re-enable it.",
+                factoryType,
+                preferredConfigKey,
+                alternateConfigKey,
+                defaultClassName,
+                PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED));
+      }
       className = defaultClassName;
       LOGGER.debug("Creating pub-sub {} adapter factory instance with default class: {}", factoryType, className);
     }

internal/venice-common/src/test/java/com/linkedin/venice/pubsub/PubSubClientsFactoryTest.java

@@ -1,13 +1,16 @@
 package com.linkedin.venice.pubsub;
 
+import static com.linkedin.venice.ConfigKeys.PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED;
 import static com.linkedin.venice.ConfigKeys.PUBSUB_ADMIN_ADAPTER_FACTORY_CLASS;
 import static com.linkedin.venice.ConfigKeys.PUBSUB_CONSUMER_ADAPTER_FACTORY_CLASS;
 import static com.linkedin.venice.ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS;
+import static com.linkedin.venice.ConfigKeys.PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS;
 import static com.linkedin.venice.ConfigKeys.PUB_SUB_ADMIN_ADAPTER_FACTORY_CLASS;
 import static com.linkedin.venice.ConfigKeys.PUB_SUB_CONSUMER_ADAPTER_FACTORY_CLASS;
 import static com.linkedin.venice.ConfigKeys.PUB_SUB_PRODUCER_ADAPTER_FACTORY_CLASS;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.expectThrows;
 
 import com.linkedin.venice.exceptions.VeniceException;
@@ -26,13 +29,6 @@
 public class PubSubClientsFactoryTest {
   @Test
   public void testCreateInstanceSuccess() {
-    // default: no config provided
-    verifyFactoryClasses(
-        new Properties(),
-        ApacheKafkaProducerAdapterFactory.class,
-        ApacheKafkaConsumerAdapterFactory.class,
-        ApacheKafkaAdminAdapterFactory.class);
-
     // with legacy config names
     Properties legacyProps = new Properties();
     legacyProps.put(PUB_SUB_PRODUCER_ADAPTER_FACTORY_CLASS, TestPubSubProducerAdapterFactory.class.getName());
@@ -56,6 +52,54 @@ public void testCreateInstanceSuccess() {
         TestPubSubAdminAdapterFactory.class);
   }
 
+  /**
+   * By default (no factory-class config and no explicit fallback flag) the factory should fail fast
+   * instead of silently defaulting to the Apache Kafka adapter factories.
+   */
+  @Test
+  public void testFailFastWhenFactoryClassMissingAndFallbackDisabled() {
+    VeniceProperties emptyProps = new VeniceProperties(new Properties());
+
+    assertFailFast(() -> PubSubClientsFactory.createProducerFactory(emptyProps), PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS);
+    assertFailFast(() -> PubSubClientsFactory.createConsumerFactory(emptyProps), PUBSUB_CONSUMER_ADAPTER_FACTORY_CLASS);
+    assertFailFast(() -> PubSubClientsFactory.createAdminFactory(emptyProps), PUBSUB_ADMIN_ADAPTER_FACTORY_CLASS);
+    assertFailFast(
+        () -> PubSubClientsFactory.createSourceOfTruthAdminFactory(emptyProps),
+        PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS);
+    // The instance constructor eagerly builds all three factories, so it should fail fast as well.
+    expectThrows(VeniceException.class, () -> new PubSubClientsFactory(emptyProps));
+
+    // Explicitly disabling the fallback behaves the same as the default.
+    Properties fallbackDisabled = new Properties();
+    fallbackDisabled.put(PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED, "false");
+    expectThrows(VeniceException.class, () -> new PubSubClientsFactory(new VeniceProperties(fallbackDisabled)));
+  }
+
+  /**
+   * When the Kafka fallback is explicitly enabled, missing factory-class configs should resolve to the
+   * Apache Kafka adapter factories (the legacy behavior).
+   */
+  @Test
+  public void testKafkaFallbackWhenExplicitlyEnabled() {
+    Properties fallbackEnabled = new Properties();
+    fallbackEnabled.put(PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED, "true");
+    verifyFactoryClasses(
+        fallbackEnabled,
+        ApacheKafkaProducerAdapterFactory.class,
+        ApacheKafkaConsumerAdapterFactory.class,
+        ApacheKafkaAdminAdapterFactory.class);
+  }
+
+  private static void assertFailFast(org.testng.Assert.ThrowingRunnable runnable, String expectedConfigKeyInMessage) {
+    VeniceException e = expectThrows(VeniceException.class, runnable);
+    assertTrue(
+        e.getMessage().contains(expectedConfigKeyInMessage),
+        "Expected fail-fast message to reference '" + expectedConfigKeyInMessage + "' but was: " + e.getMessage());
+    assertTrue(
+        e.getMessage().contains(PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED),
+        "Expected fail-fast message to reference the fallback config key but was: " + e.getMessage());
+  }
+
   private void verifyFactoryClasses(
       Properties props,
       Class<?> expectedProducer,

internal/venice-common/src/test/java/com/linkedin/venice/writer/VeniceWriterFactoryTest.java

@@ -110,6 +110,7 @@ public void testVeniceWriterFactoryWithProducerCompressionDisabled() {
   public void testVeniceWriterFactoryCreatesProducerAdapterFactory() {
     Properties properties = new Properties();
     properties.put(ConfigKeys.PUBSUB_BROKER_ADDRESS, "kafka:9898");
+    properties.put(ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS, ApacheKafkaProducerAdapterFactory.class.getName());
 
     VeniceWriterFactory veniceWriterFactory = new VeniceWriterFactory(properties, null, null, null);
     assertNotNull(veniceWriterFactory.getProducerAdapterFactory());

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/KafkaBrokerFactory.java

@@ -17,7 +17,6 @@
 import com.linkedin.venice.utils.VeniceProperties;
 import java.io.File;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
@@ -286,9 +285,22 @@ public String getPubSubClusterName() {
 
     @Override
     public Map<String, String> getAdditionalConfig() {
-      return Collections.singletonMap(
+      Map<String, String> configs = new HashMap<>();
+      configs.put(
           ConfigKeys.PUBSUB_TYPE_ID_TO_POSITION_CLASS_NAME_MAP,
           VeniceProperties.mapToString(PubSubPositionTypeRegistry.RESERVED_POSITION_TYPE_ID_TO_CLASS_NAME_MAP));
+      // Explicitly advertise the Apache Kafka adapter factories so that clients relying on
+      // getBrokerDetailsForClients() do not depend on the (now disabled by default) implicit Kafka fallback.
+      configs.put(
+          ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS,
+          KAFKA_CLIENTS_FACTORY.getProducerAdapterFactory().getClass().getName());
+      configs.put(
+          ConfigKeys.PUBSUB_CONSUMER_ADAPTER_FACTORY_CLASS,
+          KAFKA_CLIENTS_FACTORY.getConsumerAdapterFactory().getClass().getName());
+      configs.put(
+          ConfigKeys.PUBSUB_ADMIN_ADAPTER_FACTORY_CLASS,
+          KAFKA_CLIENTS_FACTORY.getAdminAdapterFactory().getClass().getName());
+      return configs;
     }
 
     @Override

internal/venice-test-common/src/main/java/com/linkedin/venice/utils/TestUtils.java

@@ -79,6 +79,9 @@
 import com.linkedin.venice.pubsub.PubSubPositionTypeRegistry;
 import com.linkedin.venice.pubsub.PubSubProducerAdapterFactory;
 import com.linkedin.venice.pubsub.PubSubTopicRepository;
+import com.linkedin.venice.pubsub.adapter.kafka.admin.ApacheKafkaAdminAdapterFactory;
+import com.linkedin.venice.pubsub.adapter.kafka.consumer.ApacheKafkaConsumerAdapterFactory;
+import com.linkedin.venice.pubsub.adapter.kafka.producer.ApacheKafkaProducerAdapterFactory;
 import com.linkedin.venice.pubsub.api.PubSubPosition;
 import com.linkedin.venice.pubsub.api.PubSubTopicType;
 import com.linkedin.venice.pubsub.manager.TopicManagerRepository;
@@ -738,8 +741,33 @@ public static VeniceControllerMultiClusterConfig getMultiClusterConfigFromOneClu
     return new VeniceControllerMultiClusterConfig(configMap);
   }
 
+  /**
+   * Returns the Apache Kafka pub-sub adapter factory-class configs (producer, consumer, admin).
+   * <p>
+   * Tests that build a {@link VeniceServerConfig} or {@link VeniceControllerClusterConfig} (which
+   * eagerly construct a {@code PubSubClientsFactory}) must supply these now that the implicit Apache
+   * Kafka fallback is disabled by default. See
+   * {@code ConfigKeys#PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED}.
+   */
+  public static Properties getPubSubApacheKafkaAdapterFactoryConfigs() {
+    Properties properties = new Properties();
+    properties.setProperty(
+        ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS,
+        ApacheKafkaProducerAdapterFactory.class.getName());
+    properties.setProperty(
+        ConfigKeys.PUBSUB_CONSUMER_ADAPTER_FACTORY_CLASS,
+        ApacheKafkaConsumerAdapterFactory.class.getName());
+    properties
+        .setProperty(ConfigKeys.PUBSUB_ADMIN_ADAPTER_FACTORY_CLASS, ApacheKafkaAdminAdapterFactory.class.getName());
+    properties.setProperty(
+        ConfigKeys.PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS,
+        ApacheKafkaAdminAdapterFactory.class.getName());
+    return properties;
+  }
+
   public static Properties getPropertiesForControllerConfig() {
     Properties properties = new Properties();
+    properties.putAll(getPubSubApacheKafkaAdapterFactoryConfigs());
     properties.put(ConfigKeys.CLUSTER_NAME, "test-cluster");
     properties.put(ConfigKeys.CONTROLLER_NAME, "venice-controller");
     properties.put(ConfigKeys.DEFAULT_REPLICA_FACTOR, "1");