[vpj] Fix NorthGuard repush ZSTD dictionary routing

[pthirun]

Problem Statement

Kafka-input repushes read the source compression dictionary in the Spark driver when the source version uses ZSTD_WITH_DICT. The dictionary-read paths constructed new consumer properties containing only SSL configuration and a broker address. This discarded the configured pub-sub consumer adapter, Xinfra routing maps, position configuration, and other job-level pub-sub properties.

Without the configured adapter factory, PubSubClientsFactory defaults to ApacheKafkaConsumerAdapterFactory. A NorthGuard-backed source topic is then queried through legacy Kafka, which returns UNKNOWN_TOPIC_OR_PARTITION until the push job times out.

Solution

Build the source dictionary consumer properties from the complete VPJ job configuration, then apply SSL overrides and set both the pub-sub and Kafka broker properties to the repush source broker.

Both driver-side dictionary-read paths now use the same helper. The helper preserves Xinfra and pub-sub configuration without mutating the original job properties. Copying the properties occurs once during driver-side dictionary lookup and has no material performance impact.

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.

No new configuration or log lines were introduced.

Concurrency-Specific Checks

Both reviewer and PR author to verify

• Code has no race conditions or thread safety issues.
• Proper synchronization mechanisms (e.g., synchronized, RWLock) are used where needed.
• No blocking calls inside critical sections that could lead to deadlocks or performance degradation.
• Verified thread-safe collections are used (e.g., ConcurrentHashMap, CopyOnWriteArrayList).
• Validated proper exception handling in multi-threaded code to avoid silent thread termination.

This change does not alter concurrency behavior.

How was this PR tested?

• New unit tests added.
• New integration tests added.
• Modified or extended existing tests.
• Verified backward compatibility (if applicable).

./gradlew :clients:venice-push-job:test \
  --tests com.linkedin.venice.hadoop.VenicePushJobRepushTest \
  --no-daemon

All 9 tests passed.

./gradlew spotlessCheck --no-daemon

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

• No. You can skip the rest of this section.
• Yes. Clearly explain the behavior change and its impact.

[sushantmane]

Thanks for tracking this down. The root cause and the buildSourceDictionaryConsumerProperties helper look right. One gap: this fixes the two DictionaryUtils.readDictionaryFromKafka sites but not the KafkaInputDictTrainer (build-new-dict) path, which is the default for a ZSTD repush.

At head (93f668d):

• kafkaInputBuildNewDictEnabled defaults to true (VenicePushJob L434; KAFKA_INPUT_COMPRESSION_BUILD_NEW_DICT_ENABLED default true).
• So fetchOrBuildCompressionDictionary() takes the if (kafkaInputBuildNewDictEnabled) branch (L1700) → new KafkaInputDictTrainer(paramBuilder.build()) → trainDict() (L1706). The else read-branch you fixed only runs when the flag is explicitly false.
• That param builder is still fed SSL-only props: .setSslProperties(pushJobSetting.enableSSL ? sslProperties.get() : new Properties()) (L1689), and KafkaInputDictTrainer.java isn't in this PR. Inside the trainer, KafkaInputUtils.getConsumerProperties(jobConf) → PubSubClientsFactory.createConsumerFactory(...) then defaults to ApacheKafkaConsumerAdapterFactory, the same fallback you're fixing.
• The empty-push/hybrid branch at L1757 hits the same trainer with the same SSL-only param.

Net: a standard ZSTD_WITH_DICT repush with default config still builds its dictionary consumer (and the split-planner TopicManager it spawns via getSplits) through Kafka, so a NorthGuard source still returns UNKNOWN_TOPIC_OR_PARTITION on that path.

I reproduced this with a xinfra e2e repush test on a fully-xinfra cluster: the trainer's consumer is constructed as ApacheKafkaConsumerAdapter with clientId CONSUMER-KafkaInputDictTrainer-for-<store>_v1, alongside DictionaryUtilsConsumer (x2) and TopicManager-0.

Suggested extension: thread the same full-config properties into the trainer. Rename KafkaInputDictTrainer.ParamBuilder.setSslProperties → setConsumerProperties (the builder merges these into the full consumer config, so the SSL-only name undersells it) and pass buildSourceDictionaryConsumerProperties(props, sslProperties, repushSourcePubsubBroker).toProperties() at both trainer sites (L1689 and the L1757 path). I can push that follow-up if you'd like.

[sushantmane]

Opened #2947 with the trainer-path follow-up — it routes the default build-new-dict path (KafkaInputDictTrainer) through the configured adapter too, so a standard ZSTD repush no longer falls back to Kafka.

[sushantmane]

Verified this now routes all four driver-side dictionary paths through the configured pub-sub adapter: the run() source read, the read branch, the default build-new-dict trainer branch, and the empty-push/hybrid rebuild (with the trainer broker set per path via the parameterized overload). The KafkaInputDictTrainer rename and the constructor ordering (consumer props first, then per-path broker/topic overrides) look correct.

This matches a fix I built independently and verified end-to-end on a fully-xinfra cluster: a ZSTD repush went from 4 ApacheKafkaConsumerAdapter constructions (dict trainer, 2 dictionary reads, split-planner TopicManager) down to 0, all now using the configured adapter. CI is green. LGTM.