[vpj] Route repush ZSTD dictionary reads through configured adapter

[sushantmane]

Problem Statement

Kafka-input repushes read the source compression dictionary in the VenicePushJob driver when the source version uses ZSTD_WITH_DICT. Every driver-side dictionary path constructed fresh consumer properties containing only SSL config and a broker address, discarding the configured pub-sub consumer adapter factory and its client configs (e.g. xinfra routing maps).

Without the adapter-factory class, PubSubClientsFactory.createConsumerFactory(...) defaults to ApacheKafkaConsumerAdapterFactory. On a NorthGuard-backed (non-Kafka) source the dictionary read is then attempted through legacy Kafka and returns UNKNOWN_TOPIC_OR_PARTITION until the push job times out.

An end-to-end xinfra repush test confirmed the leak. On a fully-xinfra cluster the repush constructed 4 ApacheKafkaConsumerAdapter instances: the dictionary trainer (KafkaInputDictTrainer), two DictionaryUtilsConsumer reads, and the split-planner TopicManager. The rest of the repush already used the configured adapter.

#2942 fixes the two DictionaryUtils.readDictionaryFromKafka sites but not the default build-new-dict trainer path: kafkaInputBuildNewDictEnabled defaults to true, so a standard ZSTD_WITH_DICT repush still falls back to Kafka for the dictionary.

Solution

Assemble the dictionary consumer properties from the full VPJ job configuration via one shared helper (buildRepushDictionaryConsumerProperties), then overlay SSL and point both pubsub.broker.address and kafka.bootstrap.servers at the repush source broker. The job config already carries pubsub.consumer.adapter.factory.class and any client-specific settings, so the dictionary is read with the same pub-sub client as the rest of the repush.

All four driver-side dictionary paths now use the helper:

• the source-dictionary read in run(),
• the dictionary read branch (build-new-dict disabled),
• the KafkaInputDictTrainer build-new-dict branch (the default), and
• the empty-push / hybrid dictionary rebuild.

KafkaInputDictTrainer.ParamBuilder.setSslProperties is renamed to setConsumerProperties because the bag is merged into the whole consumer config; the setSslProperties name understated what it carries.

This completes the read-path fix in #2942 by also covering the default trainer path, and complements the fail-fast guard in #2943, which turns the same misconfiguration into a hard error instead of a silent Kafka fallback.

The helper copies the job properties once during driver-side dictionary lookup and does not mutate the original job properties. No new configs or log lines.

Code changes

• Added new code behind a config. If so list the config names and their default values in the PR description.
• Introduced new log lines.
  • Confirmed if logs need to be rate limited to avoid excessive logging.

Concurrency-Specific Checks

Both reviewer and PR author to verify

• Code has no race conditions or thread safety issues. Driver-side, single-threaded dictionary read; the helper returns a fresh Properties copy.
• Proper synchronization mechanisms are used where needed. No shared mutable state introduced.
• No blocking calls inside critical sections that could lead to deadlocks or performance degradation.
• Verified thread-safe collections are used. No new shared collections.
• Validated proper exception handling in multi-threaded code. No new multi-threaded code.

How was this PR tested?

• New unit tests added. VenicePushJobRepushTest: the dictionary consumer properties retain the configured adapter factory and client configs, override both broker properties to the source broker, apply SSL overrides, and do not mutate the job properties.
• New integration tests added.
• Modified or extended existing tests. KafkaInputDictTrainer builder rename.
• Verified backward compatibility.

End-to-end verification: a xinfra ZSTD_WITH_DICT repush integration test that captures the pub-sub consumer adapter constructed during the repush. Without this change the repush created 4 ApacheKafkaConsumerAdapter instances; with it, 0, and the dictionary trainer, dictionary reader, and split-planner TopicManager all use the configured adapter.

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

• No. You can skip the rest of this section.

[Copilot]

Pull request overview

This PR ensures Kafka-input repush flows that need ZSTD dictionaries (especially ZSTD_WITH_DICT) read the source dictionary through the configured pub-sub consumer adapter (e.g., xinfra) by constructing dictionary consumer properties from the full VPJ job configuration and reusing that across all driver-side dictionary read paths.

Changes:

• Add a shared helper in VenicePushJob to build repush dictionary consumer properties seeded from the full job config, then overlay SSL and override broker settings to the repush source broker.
• Route dictionary reads and the KafkaInputDictTrainer “build-new-dict” path through the shared consumer properties.
• Update/extend unit tests to validate adapter config retention, SSL overrides, broker overrides, and non-mutation of job properties; rename the trainer builder method accordingly.

Reviewed changes

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

File	Description
clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/VenicePushJob.java	Introduces/uses shared repush dictionary consumer-properties helper and routes dictionary reads/trainer config through it.
clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/input/kafka/KafkaInputDictTrainer.java	Renames the builder input from SSL-only to full consumer properties and merges them into the trainer configuration.
clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/VenicePushJobRepushTest.java	Adds unit tests for the new helper (adapter config retention, SSL overrides, broker overrides, immutability).
clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/input/kafka/TestKafkaInputDictTrainer.java	Updates to the renamed builder method (setConsumerProperties).

Comments suppressed due to low confidence (1)

clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/input/kafka/KafkaInputDictTrainer.java:179

• param.consumerProperties is merged into properties after setting VENICE_REPUSH_SOURCE_PUBSUB_BROKER / KAFKA_INPUT_TOPIC / KAFKA_SOURCE_KEY_SCHEMA_STRING_PROP. Since the repush dictionary consumer properties are seeded from the full VPJ job config, they can already contain these keys (e.g., KAFKA_INPUT_TOPIC is read from job properties earlier in VenicePushJob). That means putAll can overwrite the trainer's intended broker/topic/schema and make the trainer read/train from the wrong source. Merge first, then set the required keys so they always win (and guard against null).

    properties.setProperty(VENICE_REPUSH_SOURCE_PUBSUB_BROKER, param.kafkaInputBroker);
    properties.setProperty(KAFKA_INPUT_TOPIC, param.topicName);
    properties.setProperty(KAFKA_SOURCE_KEY_SCHEMA_STRING_PROP, param.keySchema);
    this.sourceTopicName = param.topicName;
    properties.putAll(param.consumerProperties);

---

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

[sushantmane]

Superseded by #2942, which now covers all four driver-side dictionary paths (the trainer-path fix was added there after review) and has merged. Closing this duplicate.