[vpj] Materialize executor-side SSL before Spark TTL filter and chunk assembly

[pthirun]

Problem Statement

SparkPubSubPartitionReaderFactory.createReader() materializes SSL properties from the Hadoop token file before creating its PubSub consumer, but two other executor-side code paths that can also create a PubSub consumer do not:

• The TTL filter's mapPartitions closure in AbstractDataWriterSparkJob.applyTTLFilter() constructs SparkKafkaInputTTLFilter directly from broadcast job properties.
• The chunk assembly flatMapGroups closure in AbstractDataWriterSparkJob.applyChunkAssembly() constructs SparkChunkAssembler directly from broadcast job properties.

When TTL filtering is enabled together with dictionary compression (ZSTD_WITH_DICT), both of these components create a second, internal PubSub (dictionary) consumer on the executor. Since SSL was never materialized for these paths, consumer creation fails with errors like "Missing required property ssl.keystore.type".

This is a simplified, minimal fix for the same gap addressed more broadly in #2955. It intentionally does not include that PR's per-task caching refactor for SparkChunkAssembler (avoiding repeated SSL materialization / dictionary-consumer creation per key group) or the UserCredentialsFactory fail-fast hardening. Those are separate, deferrable follow-ups tracked independently.

Solution

• Extracted the existing SSL materialization logic out of SparkPubSubPartitionReaderFactory into a shared, public VPJSSLUtils.setupSSLForExecutor(VeniceProperties) utility (no-op when no SSL configurator is configured; wraps failures in VeniceException).
• SparkPubSubPartitionReaderFactory.createReader() now calls the shared utility instead of its own private copy.
• AbstractDataWriterSparkJob.applyTTLFilter()'s mapPartitions closure now calls VPJSSLUtils.setupSSLForExecutor(...) before constructing SparkKafkaInputTTLFilter.
• AbstractDataWriterSparkJob.applyChunkAssembly()'s flatMapGroups closure now calls VPJSSLUtils.setupSSLForExecutor(...) (when TTL is enabled) before constructing SparkChunkAssembler, keeping the existing per-key-group instantiation unchanged.

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

How was this PR tested?

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

Added TestVPJSSLUtils coverage for the new shared utility (no-op without an SSL configurator, and failure when the Hadoop token file can't be resolved), plus two new regression tests in DataWriterSparkJobRepushTest that force actual Spark execution (collectAsList()) of the TTL filter and chunk assembly closures with an SSL configurator configured but no resolvable token file, asserting the SSL setup failure now surfaces from both paths (proving the pre-fix code would not have attempted SSL setup there at all).

Ran the affected test classes locally: TestVPJSSLUtils, DataWriterSparkJobRepushTest, SparkPubSubPartitionReaderFactoryTest, all passing. Also verified spotlessCheck passes.

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.

[pthirun]

Verified this fix against the real behavioral tests from #2955 to confirm it resolves the same bug.

Ported from #2955 into this branch:

• SparkExecutorTestUtils (recording SSL configurator + asserting consumer factory)
• testCreateReaderMaterializesExecutorSSL
• testApplyTTLFilterMaterializesSSLBeforeReadingZstdDictionary
• testApplyChunkAssemblyMaterializesSSLBeforePostAssemblyTTL (adapted from testApplyChunkAssemblyReusesExecutorSSLForPostAssemblyTTL, with invocation-count assertions relaxed from 1 to 2 since this fix doesn't include [vpj] Materialize SSL properties for Spark TTL filters #2955's per-task assembler caching)

To confirm the tests actually catch the bug (not just pass trivially), I reverted the fix locally and reran them. Both TTL-filter and chunk-assembly tests failed with the exact real-world error from #2955:

com.linkedin.venice.exceptions.UndefinedPropertyException: Missing required property 'ssl.keystore.type'

reproduced through DictionaryUtils.readDictionaryFromKafka → KafkaInputUtils.getCompressor → VeniceRmdTTLFilter.<init>, matching #2955's root cause. Reapplying the fix, all tests pass.

Scope difference from #2955: this PR materializes SSL per key-group/task invocation rather than caching a single SparkChunkAssembler per task via TaskCompletionListener. That caching (and the UserCredentialsFactory hardening in #2955) is left as a smaller, separate follow-up.

[ymuppala]

Thanks for splitting this out — the shared VPJSSLUtils.setupSSLForExecutor() extraction is the right shape, and the SparkPubSubPartitionReaderFactory change is a faithful code move (identical logic; only the thrown type changes from RuntimeException to VeniceException, which is fine).

I verified the change is config-safe: getSslProperties() only emits ssl.*, ssl.enabled, pubsub.security.protocol, and ssl.configurator.class. It never writes a broker key. VeniceProperties.toProperties() returns a copy and the backing map is unmodifiable, so the shared Spark Broadcast<Properties> can't be mutated by concurrent task threads. The returned properties stay local to each closure and never reach the VeniceWriter. testCrossFabricRepushBrokerIsolation still passes. I compiled and ran DataWriterSparkJobRepushTest, TestVPJSSLUtils, and SparkPubSubPartitionReaderFactoryTest locally on JDK 17 — all green.

The tests are also good quality: they cover the no-op path and a real failure path, not just the happy path.

That said, I think there is one blocking issue and two things worth addressing before merge. Details are in the inline comments; summarizing here:

1. (blocking) SSL materialization and SparkChunkAssembler construction happen once per key group in applyChunkAssembly, i.e. once per unique key. See inline comment on AbstractDataWriterSparkJob.java.
2. SSL setup is unconditional on the TTL/chunk-assembly paths, which converts a currently-working configuration into one that can fail.
3. A pre-existing broker-precedence issue in KafkaInputUtils.getCompressor() becomes reachable for the first time because of this fix (below).

---

Broker precedence in KafkaInputUtils.getCompressor() (not in this diff, but newly reachable)

if (strategy.equals(CompressionStrategy.ZSTD_WITH_DICT)) {
  Properties props = properties.toProperties();
  props.setProperty(KAFKA_BOOTSTRAP_SERVERS, kafkaUrl);   // only sets kafka.bootstrap.servers
  ByteBuffer dict = DictionaryUtils.readDictionaryFromKafka(topic, new VeniceProperties(props));

PubSubUtil.getPubSubBrokerAddress() resolves pubsub.broker.address first, falling back to kafka.bootstrap.servers only when the former is absent:

public static String getPubSubBrokerAddress(VeniceProperties properties) {
  return properties.getStringWithAlternative(PUBSUB_BROKER_ADDRESS, KAFKA_BOOTSTRAP_SERVERS);
}

pubsub. is one of DataWriterComputeJob.PASS_THROUGH_CONFIG_PREFIXES, so any pubsub.broker.address present in the job properties is copied verbatim into the Spark conf, flows into the broadcast filter properties, and would silently take precedence over the source broker that getCompressor() intended to use — meaning the dictionary consumer could read from a different cluster than the repush source.

This is pre-existing, but before this PR the path failed fast on missing SSL properties, so it was never exercised. After this PR it succeeds, and a misrouted read would be silent rather than loud. KafkaInputUtils.getConsumerProperties() already guards against this at line 85 by setting PUBSUB_BROKER_ADDRESS explicitly. Suggest the same one-liner here:

props.setProperty(PUBSUB_BROKER_ADDRESS, kafkaUrl);

Even if no current job sets pubsub.broker.address, making the intended broker explicit at the point of use is cheap insurance for a failure mode that is very hard to detect after the fact.

Minor: wasted filter construction when chunking is enabled

In applyTTLFilter, SparkKafkaInputTTLFilter is constructed eagerly at the top of the mapPartitions closure, but a few lines later every row short-circuits when isChunkingEnabled is true (TTL is applied post-assembly instead). So for chunked stores the filter — including its HDFS schema fetch and, for ZSTD_WITH_DICT, a dictionary consumer — is built per partition and never used. Previously this was masked because construction failed anyway; now it will run. Worth skipping the whole mapPartitions (or constructing lazily) when isChunkingEnabled is true.

[pthirun]

Addressed the blocking issue in 0d2ea76.

• VPJSSLUtils.setupSSLForExecutor() now memoizes the materialized SSL properties (keystore/truststore temp-file locations) behind a per-JVM double-checked lock. The Hadoop token file and SSL configurator class are fixed for the executor's lifetime, so the token-file read and temp-file writes now happen once per executor instead of once per call. Failures are never cached, so a transient issue can still be retried.
• applyChunkAssembly's flatMapGroups closure is now a named ChunkAssemblyFunction that lazily builds one SparkChunkAssembler per Spark task and reuses it across all key groups in that task (Spark reuses the same deserialized function instance for every group processed by a task), instead of constructing a new one per key group. The assembler is released via a TaskContext completion listener.
• SparkChunkAssembler now implements AutoCloseable, closing its post-assembly TTL filter's schema reader when the task completes.

With this, testApplyChunkAssemblyReusesExecutorSSLForPostAssemblyTTL now asserts the SSL configurator/consumer factory are invoked exactly once across both key groups in its single-task test, matching #2955's original assertion.

Ran the full venice-push-job test module afterward: 828 tests, 0 failures, 0 errors.

[pthirun]

Also addressed the two items from the review summary that did not have a diff to comment on directly (fixed in c59bf97):

Broker precedence in KafkaInputUtils.getCompressor(): added props.setProperty(PUBSUB_BROKER_ADDRESS, kafkaUrl) right after the KAFKA_BOOTSTRAP_SERVERS line, mirroring the existing guard in getConsumerProperties(). Added testGetCompressorOverridesStalePubSubBrokerAddressForZstdWithDict to KafkaInputUtilsTest, which seeds a stale pubsub.broker.address and verifies getCompressor() overrides it with the source kafkaUrl.

Wasted filter construction when chunking is enabled: applyTTLFilter now returns immediately when isChunkingEnabled is true, before entering mapPartitions, so SparkKafkaInputTTLFilter (and its HDFS schema fetch / dictionary consumer) is never constructed for chunked stores.

Ran the full venice-push-job module afterward: 829 tests, 0 failures, 0 errors.