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

[sushantmane]

Problem Statement

PubSubClientsFactory silently defaulted the producer/consumer/admin adapter factories to Apache Kafka
when their *.adapter.factory.class config was absent. On non-Kafka (e.g. xinfra) deployments this
masks a real misconfiguration: a Kafka client is built that can't talk to the configured backend, and
the failure surfaces later, far from the root cause.

Solution

Make the fallback config-driven and fail-fast by default. When a factory-class config is missing or
blank and the fallback is disabled, PubSubClientsFactory throws a VeniceException naming the missing
key instead of constructing the Apache Kafka factory.

Config keys

Key	Default	Purpose
pubsub.adapter.factory.kafka.fallback.enabled	false	New. true restores the legacy implicit Apache Kafka default for any factory-class config that is missing/blank; false = fail fast.
pubsub.producer.adapter.factory.class	—	Producer adapter factory FQCN.
pubsub.consumer.adapter.factory.class	—	Consumer adapter factory FQCN.
pubsub.admin.adapter.factory.class	—	Admin adapter factory FQCN.
pubsub.source.of.truth.admin.adapter.factory.class	—	Source-of-truth admin adapter factory FQCN.

Because fail-fast is now the default, every place that builds a pub-sub client supplies these classes.
Integration tests resolve them from the pub-sub backend under test via
ServiceFactory.getPubSubClientConfigs() (driven by the pubSubBrokerFactory system property) instead
of hard-coding Apache Kafka, so the suite runs against whatever backend is configured.
KafkaBrokerFactory.getAdditionalConfig advertises all four to cluster components; forked JVMs
(DaVinciUserApp, changelog user app, forked servers), docker server/controller configs,
VenicePulsarSink, and admin-tool commands supply them; and two production paths that dropped the class
when deriving a config are fixed — VeniceChangelogConsumerImpl (RocksDB-buffer VeniceServerConfig)
and DefaultPushJobHeartbeatSenderFactory (heartbeat VeniceWriterFactory). venice-test-common main
helpers and pure unit tests keep the Apache Kafka default because the pluggable backend lives in the
integrationTest source set and no broker is started there.

Code changes

• Added new code behind a config: pubsub.adapter.factory.kafka.fallback.enabled (default false).
• Introduced new log lines (debug-level only).

How was this PR tested?

• PubSubClientsFactoryTest: config resolution (new + legacy keys), fail-fast-by-default,
  present-but-blank value handling, and the explicit fallback-enabled path.
• CI (unit + integration + Pulsar container test), exercising every fixed call site.

Does this PR introduce any user-facing or breaking changes?

• Yes. Fail-fast is now the default. A deployment relying on the implicit Apache Kafka default must
  set the four pubsub.*.adapter.factory.class configs above, or set
  pubsub.adapter.factory.kafka.fallback.enabled=true. The bundled docker sample configs are updated.

[Copilot]

Pull request overview

This PR makes pub-sub adapter factory selection fail fast when required factory-class configs are missing, preventing implicit fallback to Apache Kafka in non-Kafka deployments unless explicitly enabled via a new config flag.

Changes:

• Added pubsub.adapter.factory.kafka.fallback.enabled (default false) and updated PubSubClientsFactory to throw a descriptive VeniceException when factory-class configs are missing and fallback is disabled.
• Updated integration test broker config wiring to explicitly propagate the Apache Kafka adapter factory class names.
• Added/updated unit tests to cover both fail-fast default behavior and the opt-in Kafka fallback behavior.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.

File	Description
internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/KafkaBrokerFactory.java	Propagates explicit pub-sub adapter factory class configs to keep integration/e2e wiring working without implicit Kafka fallback.
internal/venice-common/src/test/java/com/linkedin/venice/pubsub/PubSubClientsFactoryTest.java	Adds unit coverage for fail-fast behavior by default and legacy behavior when fallback is explicitly enabled.
internal/venice-common/src/main/java/com/linkedin/venice/pubsub/PubSubClientsFactory.java	Introduces config-driven, default-disabled Kafka fallback and throws descriptive exceptions when factory configs are missing.
internal/venice-common/src/main/java/com/linkedin/venice/ConfigKeys.java	Adds the new pubsub.adapter.factory.kafka.fallback.enabled config key with documentation.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 7 out of 7 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (2)

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

• getPubSubApacheKafkaAdapterFactoryConfigs() is used by tests that construct VeniceControllerClusterConfig, which also creates a source-of-truth admin adapter factory (PubSubClientsFactory.createSourceOfTruthAdminFactory). With the new fail-fast default, omitting ConfigKeys.PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS will cause controller-config construction to throw in many tests. The helper and its Javadoc should include this key as well (it currently says producer/consumer/admin only).

   * 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

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

• testKafkaFallbackWhenExplicitlyEnabled() verifies the fallback for the three factories created by the PubSubClientsFactory constructor, but it doesn't verify the same fallback behavior for createSourceOfTruthAdminFactory(), which is part of the fail-fast change surface area described in the PR.

  @Test
  public void testKafkaFallbackWhenExplicitlyEnabled() {
    Properties fallbackEnabled = new Properties();
    fallbackEnabled.put(PUBSUB_ADAPTER_FACTORY_KAFKA_FALLBACK_ENABLED, "true");
    verifyFactoryClasses(

[Copilot]

Pull request overview

Copilot reviewed 9 out of 9 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (1)

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

• createFactory treats an empty/blank factory-class value as "configured" (because the key exists) and then fails later with a generic Failed to create instance of class: ... error. Since this PR is aiming for fail-fast misconfiguration detection, it should also treat blank values as missing and throw the same targeted exception. While updating that logic, consider rewording the exception to reference the config key(s) directly instead of %s = FactoryType, because createSourceOfTruthAdminFactory currently passes FactoryType.ADMIN, which makes the message less specific even though the key differs.

    String className;
    if (properties.containsKey(preferredConfigKey) || properties.containsKey(alternateConfigKey)) {
      className = properties.getStringWithAlternative(preferredConfigKey, alternateConfigKey);
      LOGGER.debug("Creating pub-sub {} adapter factory instance for class: {}", factoryType, className);
    } else {

[Copilot]

Pull request overview

Copilot reviewed 12 out of 12 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (2)

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

• The Javadoc says this helper returns configs for "producer, consumer, admin", but the method also sets PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS. Please update the comment to match the actual returned properties so callers understand what they get.

   * Returns the Apache Kafka pub-sub adapter factory-class configs (producer, consumer, admin).

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

• This config key controls the implicit Kafka fallback not only for producer/consumer/admin factories but also for the source-of-truth admin factory (PubSubClientsFactory.createSourceOfTruthAdminFactory uses the same check). The Javadoc should mention source-of-truth admin to avoid misleading readers.

   * 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.

[Copilot]

Pull request overview

Copilot reviewed 16 out of 16 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (3)

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

• The Javadoc says this returns configs for producer/consumer/admin, but the method also sets PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS. Please update the Javadoc to match the actual returned properties.

   * Returns the Apache Kafka pub-sub adapter factory-class configs (producer, consumer, admin).

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

• getAdditionalConfig() now advertises producer/consumer/admin adapter factory classes, but it does not advertise PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS. VeniceControllerClusterConfig eagerly constructs the source-of-truth admin adapter factory, so tests wiring configs via getBrokerDetailsForClients() can still fail fast due to this missing key when the Kafka fallback is disabled by default.

      // 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;

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

• This config flag also affects the source-of-truth admin adapter factory (via PubSubClientsFactory.createSourceOfTruthAdminFactory), but the Javadoc only mentions producer/consumer/admin. Please include source-of-truth admin here to avoid misleading readers about the scope of the breaking change.

   * Configuration key that controls whether the PubSub producer/consumer/admin adapter factories

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (2)

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

• The PR description states pubsub.adapter.factory.kafka.fallback.enabled should default to false (fail-fast by default), but the implementation currently sets DEFAULT_KAFKA_FALLBACK_ENABLED = true, meaning Apache Kafka fallback remains enabled unless explicitly disabled. Please align the default behavior with the PR description (flip the default and update tests/docs), or update the PR description if the default is intentionally backward-compatible.

   * {@code false} to opt into fail-fast, where a missing factory-class config raises an exception
   * instead of silently defaulting to Apache Kafka (recommended for non-Kafka deployments).
   */
  public static final boolean DEFAULT_KAFKA_FALLBACK_ENABLED = true;

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

• This new config key’s Javadoc says it defaults to true (Kafka fallback), but the PR description says the new flag should default to false (fail-fast by default). Please reconcile the documented default with the intended behavior and the implementation in PubSubClientsFactory so operators know what happens when the key is omitted.

   * Defaults to {@code true} (fall back to Apache Kafka) for backward compatibility. Set this to
   * {@code false} to opt into fail-fast: the {@code PubSubClientsFactory} then throws when a
   * factory-class config is missing, surfacing misconfiguration early instead of masking it behind
   * an implicit Kafka default. Recommended for non-Kafka (e.g. xinfra) deployments.
   */

[Copilot]

Pull request overview

Copilot reviewed 63 out of 63 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (2)

clients/venice-push-job/src/test/java/com/linkedin/venice/heartbeat/TestPushJobHeartbeatSender.java:34

• The test passes pub-sub adapter factory configs via the sslProperties parameter, but sslProperties is intended for Kafka SSL client configs and is blindly merged into veniceWriterProperties in DefaultPushJobHeartbeatSenderFactory. This is confusing and couples non-SSL config propagation to the SSL path; consider updating DefaultPushJobHeartbeatSenderFactory#getVeniceWriterProperties(...) to merge the needed pub-sub factory configs from the properties argument instead (and keep sslProperties truly SSL-only).

    VeniceProperties properties = new VeniceProperties(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
    Optional<Properties> sslProperties = Optional.of(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
    DefaultPushJobHeartbeatSenderFactory pushJobHeartbeatSenderFactory = new DefaultPushJobHeartbeatSenderFactory();

integrations/venice-pulsar/src/main/java/com/linkedin/venice/pulsar/sink/VenicePulsarSink.java:250

• This hard-coded class name string is brittle (won’t be caught by refactors/relocations). Prefer using the class literal to keep it compile-time safe.

    config.put(
        ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS,
        "com.linkedin.venice.pubsub.adapter.kafka.producer.ApacheKafkaProducerAdapterFactory");

[Copilot]

Pull request overview

Copilot reviewed 64 out of 64 changed files in this pull request and generated no new comments.

Suppressed comments (4)

clients/venice-push-job/src/test/java/com/linkedin/venice/heartbeat/TestPushJobHeartbeatSender.java:34

• This test passes pub-sub adapter factory configs via the sslProperties parameter, but DefaultPushJobHeartbeatSenderFactory only merges sslProperties into the VeniceWriter properties (it ignores the main VeniceProperties argument when building writer props). That makes sslProperties a confusing carrier for non-SSL configs and suggests the production path may fail to configure PubSubClientsFactory when SSL is disabled. Consider updating DefaultPushJobHeartbeatSenderFactory to also merge the needed pub-sub factory configs from the main properties argument, then keep sslProperties strictly SSL-related.

    VeniceProperties properties = new VeniceProperties(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
    Optional<Properties> sslProperties = Optional.of(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
    DefaultPushJobHeartbeatSenderFactory pushJobHeartbeatSenderFactory = new DefaultPushJobHeartbeatSenderFactory();

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

• createFactory() treats an empty-string factory class as "configured" because it only checks containsKey(). If a key is present but blank, this falls through to createInstance("") and throws a generic "Failed to create instance" error instead of the intended fail-fast message naming the missing config key. Consider treating blank values as missing (and then applying the fallback-enabled logic / fail-fast message).

    String className;
    if (properties.containsKey(preferredConfigKey) || properties.containsKey(alternateConfigKey)) {
      className = properties.getStringWithAlternative(preferredConfigKey, alternateConfigKey);
      LOGGER.debug("Creating pub-sub {} adapter factory instance for class: {}", factoryType, className);
    } else {

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

• Javadoc says this helper returns factory-class configs for producer/consumer/admin, but the method also sets PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS. Update the comment to match the actual configs returned so callers know it covers source-of-truth admin as well.
  internal/venice-test-common/src/main/java/com/linkedin/venice/utils/TestWriteUtils.java:993
• PASS_THROUGH_CONFIG_PREFIXES_LIST_KEY is a comma-separated list; using putIfAbsent() means if a test already sets a non-empty pass-through prefix list that doesn't include "pubsub.", the pub-sub factory-class configs still won't be passed through and PubSubClientsFactory will fail fast later. Consider appending PUBSUB_CLIENT_CONFIG_PREFIX when the key is present but missing that prefix.

[Copilot]

Pull request overview

Copilot reviewed 66 out of 66 changed files in this pull request and generated no new comments.

Suppressed comments (2)

clients/venice-push-job/src/test/java/com/linkedin/venice/heartbeat/TestPushJobHeartbeatSender.java:34

• sslProperties is being used here to smuggle pub-sub adapter factory configs into DefaultPushJobHeartbeatSenderFactory.getVeniceWriterProperties(). That masks a real runtime issue: when sslProperties is empty (SSL disabled), the heartbeat sender will build a VeniceWriterFactory with only KAFKA_BOOTSTRAP_SERVERS, and PubSubClientsFactory.createProducerFactory() will now fail-fast because pubsub.producer.adapter.factory.class is missing. The implementation should merge the adapter-factory configs from the main properties argument into the writer properties (independent of SSL), and this test should avoid coupling adapter factory config to SSL config.

    String heartbeatStoreName = AvroProtocolDefinition.BATCH_JOB_HEARTBEAT.getSystemStoreName();
    VeniceProperties properties = new VeniceProperties(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
    Optional<Properties> sslProperties = Optional.of(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
    DefaultPushJobHeartbeatSenderFactory pushJobHeartbeatSenderFactory = new DefaultPushJobHeartbeatSenderFactory();

integrations/venice-pulsar/src/main/java/com/linkedin/venice/pulsar/sink/VenicePulsarSink.java:250

• Hard-coding the producer adapter factory class name as a string is brittle (renames/refactors won’t be caught at compile time). Prefer referencing the class and using .class.getName() (and/or reuse the shared TestUtils helper pattern) to avoid typos and keep this in sync with the actual implementation.

    // The sink's Venice producer writes to the (Kafka-backed) real-time topic; set the pub-sub producer
    // adapter factory explicitly so it does not fail fast when the factory class is unconfigured.
    config.put(
        ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS,
        "com.linkedin.venice.pubsub.adapter.kafka.producer.ApacheKafkaProducerAdapterFactory");

[Copilot]

Pull request overview

Copilot reviewed 68 out of 68 changed files in this pull request and generated no new comments.

Suppressed comments (2)

clients/venice-push-job/src/test/java/com/linkedin/venice/heartbeat/TestPushJobHeartbeatSender.java:33

• This test now passes pub-sub adapter factory configs via sslProperties, but sslProperties is only forwarded into VeniceWriterFactory when SSL is enabled in production (VenicePushJob#createPushJobHeartbeatSender passes Optional.empty() when SSL is off). With the new fail-fast default, the heartbeat sender’s VeniceWriterFactory will still be missing pubsub.producer.adapter.factory.class when SSL is disabled, and can fail at runtime. Consider updating DefaultPushJobHeartbeatSenderFactory#getVeniceWriterProperties(...) (or the caller) to always include the required pub-sub producer factory-class config (and keep sslProperties for SSL-only), then this test can avoid overloading the SSL parameter.

    VeniceProperties properties = new VeniceProperties(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
    Optional<Properties> sslProperties = Optional.of(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());

internal/venice-test-common/src/main/java/com/linkedin/venice/utils/TestWriteUtils.java:993

• PASS_THROUGH_CONFIG_PREFIXES_LIST_KEY is only set if absent. If a caller pre-populates it (e.g., with other prefixes), this helper won’t ensure pubsub. configs are passed through, which can still trigger the new fail-fast behavior when VPJ constructs pub-sub clients. Safer to append PUBSUB_CLIENT_CONFIG_PREFIX when the list is present but doesn’t include it.

[Copilot]

Pull request overview

Copilot reviewed 70 out of 70 changed files in this pull request and generated no new comments.

Suppressed comments (1)

integrations/venice-pulsar/src/main/java/com/linkedin/venice/pulsar/sink/VenicePulsarSink.java:250

• Avoid hard-coding the Apache Kafka producer adapter factory class name as a string. This module already depends on :internal:venice-common, so you can use the class literal to make refactors safe and avoid typos.

    // The sink's Venice producer writes to the (Kafka-backed) real-time topic; set the pub-sub producer
    // adapter factory explicitly so it does not fail fast when the factory class is unconfigured.
    config.put(
        ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS,
        "com.linkedin.venice.pubsub.adapter.kafka.producer.ApacheKafkaProducerAdapterFactory");

[Copilot]

Pull request overview

Copilot reviewed 71 out of 71 changed files in this pull request and generated no new comments.

Suppressed comments (3)

clients/venice-push-job/src/test/java/com/linkedin/venice/heartbeat/TestPushJobHeartbeatSender.java:34

• This test passes the pub-sub adapter factory configs via sslProperties. That can mask regressions in DefaultPushJobHeartbeatSenderFactory, because getVeniceWriterProperties() merges sslProperties before the new forwarding logic, so the test may still pass even if forwarding from jobProperties breaks. Keep sslProperties empty (or limited to actual SSL keys) so the test exercises the intended code path.

    VeniceProperties properties = new VeniceProperties(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
    Optional<Properties> sslProperties = Optional.of(TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs());
    DefaultPushJobHeartbeatSenderFactory pushJobHeartbeatSenderFactory = new DefaultPushJobHeartbeatSenderFactory();

clients/venice-push-job/src/main/java/com/linkedin/venice/heartbeat/DefaultPushJobHeartbeatSenderFactory.java:110

• The heartbeat writer props only forward the new 'pubsub.producer.adapter.factory.class' key. If a job is configured using the legacy 'pub.sub.producer.adapter.factory.class' key (still supported by PubSubClientsFactory via getStringWithAlternative), the heartbeat VeniceWriterFactory will fail-fast because the legacy key is dropped here. Forward whichever key is set, and write it back under the preferred key for downstream resolution.

    // Forward the pub-sub producer adapter factory class so the heartbeat VeniceWriterFactory can resolve
    // it; the factory fails fast otherwise (there is no implicit default).
    if (jobProperties.containsKey(PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS)) {
      veniceWriterProperties
          .put(PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS, jobProperties.getString(PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS));

clients/venice-admin-tool/src/test/java/com/linkedin/venice/TestAdminTool.java:112

• The @BeforeClass/@afterclass hooks duplicate the system-property save/restore logic that now exists in TestUtils. Using the shared helper avoids drift and keeps the managed key set consistent across tests.

    Properties factoryConfigs = TestUtils.getPubSubApacheKafkaAdapterFactoryConfigs();
    for (String key: PUBSUB_ADAPTER_FACTORY_CONFIG_KEYS) {
      String originalValue = System.getProperty(key);
      if (originalValue != null) {
        ORIGINAL_PUBSUB_ADAPTER_FACTORY_SYSTEM_PROPERTIES.setProperty(key, originalValue);

[Copilot]

Pull request overview

Copilot reviewed 71 out of 71 changed files in this pull request and generated no new comments.

Suppressed comments (2)

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

• Current resolution uses getStringWithAlternative(preferred, alternate, null), which picks the preferred key whenever it is present—even if it is blank. That means a configuration with a blank preferred key but a valid legacy key will be treated as “not configured” and fail fast / fall back incorrectly. Prefer a non-blank value between preferred and legacy keys.

    String className = properties.getStringWithAlternative(preferredConfigKey, alternateConfigKey, null);
    if (className == null || className.trim().isEmpty()) {

clients/venice-push-job/src/main/java/com/linkedin/venice/heartbeat/DefaultPushJobHeartbeatSenderFactory.java:110

• Only the new PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS key is forwarded into veniceWriterProperties. If a job is configured with the legacy pub.sub.producer.adapter.factory.class key (which PubSubClientsFactory still supports), the heartbeat writer will fail fast because the derived VeniceWriterFactory properties drop that value. Forward whichever of the new/legacy keys is set (and ignore blank values).

    // Forward the pub-sub producer adapter factory class so the heartbeat VeniceWriterFactory can resolve
    // it; the factory fails fast otherwise (there is no implicit default).
    if (jobProperties.containsKey(PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS)) {
      veniceWriterProperties
          .put(PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS, jobProperties.getString(PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS));
    }

[Copilot]

Pull request overview

Copilot reviewed 73 out of 73 changed files in this pull request and generated no new comments.

Suppressed comments (3)

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

• This config key also governs source-of-truth admin adapter factory resolution (PubSubClientsFactory.createSourceOfTruthAdminFactory). The Javadoc currently lists only producer/consumer/admin, which is misleading for readers troubleshooting missing PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS.

   * Configuration key that controls whether the PubSub producer/consumer/admin adapter factories

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

• Javadoc is slightly inaccurate: this helper also sets the source-of-truth admin adapter factory class (PUBSUB_SOURCE_OF_TRUTH_ADMIN_ADAPTER_FACTORY_CLASS), not just producer/consumer/admin.
  integrations/venice-pulsar/src/main/java/com/linkedin/venice/pulsar/sink/VenicePulsarSink.java:250
• Avoid hard-coding the factory class name as a string literal; using .class.getName() keeps this correct if the class is ever relocated/renamed and aligns with other call sites in the repo.

    config.put(
        ConfigKeys.PUBSUB_PRODUCER_ADAPTER_FACTORY_CLASS,
        "com.linkedin.venice.pubsub.adapter.kafka.producer.ApacheKafkaProducerAdapterFactory");