[vpj] Add writer hook provider support

[misyel]

Problem Statement

Venice Push Job creates its primary data writer without a way to supply an optional VeniceWriterHook. This prevents extensions from using the existing pre-produce hook without changing the writer implementation.

Solution

Add a lightweight executor-side VeniceWriterHookFactory extension point with the minimal practical API:

VeniceWriterHook createWriterHook(String storeName, VeniceProperties taskProperties);

VPJ instantiates the configured factory inside each executor task JVM. The store name is parsed from the destination topic, and the executor task properties include settings forwarded through the push.job.writer.hook.* namespace. Factory loading uses Class#asSubclass, and the hook is attached only to the primary data writer. Materialized-view child writers, control-message writers, and heartbeat writers do not receive the hook. Reused Spark sessions clear stale hook settings before applying the current job configuration.

Code changes

• Added new code behind a config. push.job.writer.hook.factory.class is unset by default, which preserves existing behavior. Additional factory settings use the push.job.writer.hook.* prefix.
• 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.

The factory and hook are scoped to one partition writer, and the hook runs before the writer partition lock. No shared mutable collections or new synchronization are introduced.

How was this PR tested?

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

Ran targeted Venice Push Job tests covering configured and unconfigured factories, the two factory inputs, whitespace configuration, invalid classes, missing constructors, null hooks, materialized-view isolation, MapReduce and Spark property propagation, and stale Spark-session cleanup. Also ran Java Spotless checks and Venice Push Job SpotBugs checks for main and test code.

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.

🤖 Generated with GitHub Copilot CLI

[Copilot]

Pull request overview

Adds a new VPJ extension point that allows jobs to supply an optional VeniceWriterHook via a configurable VeniceWriterHookProvider, enabling custom pre-produce behavior without changing the writer implementation. This is wired through both MapReduce and Spark so provider configuration reaches executor task properties, and Spark sessions proactively clear stale hook-related runtime config between jobs.

Changes:

• Introduces VeniceWriterHookProvider and VPJ config keys under push.job.writer.hook.* (including provider.class).
• Loads the provider per partition writer, injects the hook only into the primary data writer, and manages provider lifecycle/cleanup.
• Forwards hook-provider configuration to Spark RuntimeConfig and MR JobConf, with Spark unsetting stale hook settings on reused sessions; adds unit tests.

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
clients/venice-push-job/src/main/java/com/linkedin/venice/vpj/VenicePushJobConstants.java	Adds new VPJ config constants for hook provider class + provider-specific settings prefix.
clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/task/datawriter/VeniceWriterHookProvider.java	Introduces the new provider interface and its immutable initialization context.
clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/task/datawriter/AbstractPartitionWriter.java	Loads provider, injects hook into main writer options, and closes provider during task shutdown.
clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/mapreduce/datawriter/jobs/DataWriterMRJob.java	Forwards push.job.writer.hook.* properties to MR task config.
clients/venice-push-job/src/main/java/com/linkedin/venice/spark/datawriter/jobs/AbstractDataWriterSparkJob.java	Clears stale hook configs from reused Spark sessions and forwards push.job.writer.hook.* to executors.
clients/venice-push-job/src/test/java/com/linkedin/venice/hadoop/task/datawriter/AbstractPartitionWriterHookProviderTest.java	Adds unit tests for provider loading, hook injection, MV isolation, and provider cleanup/failure behavior.
clients/venice-push-job/src/test/java/com/linkedin/venice/spark/datawriter/jobs/AbstractDataWriterSparkJobTest.java	Adds unit tests verifying Spark config propagation and stale-session cleanup for hook-provider settings.

---

💡 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 no new comments.

[Copilot]

Pull request overview

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

Comments suppressed due to low confidence (3)

clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/task/datawriter/AbstractPartitionWriter.java:1085

• Factory instantiation can also fail with LinkageError/ExceptionInInitializerError (e.g., static init or missing deps). Catching only Exception may leak a raw error without the configured class name, making misconfiguration harder to triage. Wrap LinkageError here as well.

    try {
      return ReflectUtils.callConstructor(factoryClass, new Class<?>[0], new Object[0]);
    } catch (Exception e) {
      throw new VeniceException(
          "Failed to instantiate " + VeniceWriterHookFactory.class.getSimpleName() + " '" + className + "'",
          e);
    }

clients/venice-push-job/src/main/java/com/linkedin/venice/hadoop/task/datawriter/AbstractPartitionWriter.java:1070

• ReflectUtils.loadClass(className) can fail with LinkageError (e.g., NoClassDefFoundError when the factory’s transitive deps are missing). The current catch (Exception e) won’t wrap those, so operators may see an unhelpful raw error without the configured class name. Catch LinkageError as well and wrap it in the same VeniceException for consistent, actionable diagnostics.

This issue also appears on line 1079 of the same file.

    try {
      loadedClass = ReflectUtils.loadClass(className);
    } catch (Exception e) {
      throw new VeniceException(
          "Failed to load " + VeniceWriterHookFactory.class.getSimpleName() + " class '" + className + "'",
          e);
    }

clients/venice-push-job/src/main/java/com/linkedin/venice/vpj/VenicePushJobConstants.java:529

• The PR description documents the opt-in key as push.job.writer.hook.provider.class, but the code introduces push.job.writer.hook.factory.class (PUSH_JOB_WRITER_HOOK_FACTORY_CLASS). This mismatch is likely to cause misconfiguration for early adopters and makes it unclear which key is supported.

  /**
   * Fully-qualified class name of the optional
   * {@code com.linkedin.venice.hadoop.task.datawriter.VeniceWriterHookFactory}. The class must have a public
   * no-arg constructor. When absent or empty, VPJ creates writers exactly as before, without a writer hook.
   */
  public static final String PUSH_JOB_WRITER_HOOK_FACTORY_CLASS = PUSH_JOB_WRITER_HOOK_PROP_PREFIX + "factory.class";

[Copilot]

Pull request overview

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

Comments suppressed due to low confidence (1)

clients/venice-push-job/src/main/java/com/linkedin/venice/spark/datawriter/jobs/AbstractDataWriterSparkJob.java:243

• This cleanup copies all Spark RuntimeConfig keys into a new ArrayList before filtering by the hook prefix. On large Spark sessions this creates unnecessary allocations; you can avoid that by collecting only the matching keys (still ensuring you don't mutate the config while iterating).

    new ArrayList<>(JavaConverters.mapAsJavaMap(jobConf.getAll()).keySet()).stream()
        .filter(key -> key.startsWith(PUSH_JOB_WRITER_HOOK_PROP_PREFIX))
        .forEach(jobConf::unset);