[protocol][build] Stage PushJobDetails v6 - #2971
Conversation
Registers PushJobDetails v6, which appends externalStorageWriteTimeMs and veniceWriteTimeMs, both typed long with a -1 default meaning "not reported". The follow-up PR makes the push job aggregate these durations and the controller emit metrics from them. This PR only registers the schema, it does not activate it. Controllers register every PushJobDetails schema version found in the resources of the venice-common they were built with, so the registration has to be deployed to the controller fleet before any push job serializes a v6 payload, otherwise the controller rejects the write. Following the staged protocol convention documented in the root build.gradle, the generated PushJobDetails classes are pinned to v5 via versionOverrides, so this build registers v6 while the code still serializes v5. The follow-up PR removes the override and bumps AvroProtocolDefinition.PUSH_JOB_DETAILS to v6. Testing: TestPushJobDetailsSchemaCompatibility now asserts that v6 is staged and not activated (protocol definition still on v5, compiled class still v5), that v6 only appends the two long fields with -1 defaults, that v5 and v6 are compatible in both directions as schema registration requires, and that a v6 reader resolves a v5 writer's record to the -1 defaults. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
There was a problem hiding this comment.
Pull request overview
Stages a new PushJobDetails Avro schema version (v6) intended to add wall-clock timing fields so push-job status reporting can later attribute time to external storage vs Venice write paths, while keeping the active/serialized protocol pinned to v5 during rollout.
Changes:
- Adds
PushJobDetailsv6 schema that appends two defaulted timing fields (externalStorageWriteTimeMs,veniceWriteTimeMs). - Pins generated
PushJobDetailsclasses to v5 viaversionOverridesin the rootbuild.gradle(staged protocol pattern). - Extends schema compatibility tests to validate staged-v6 shape and v5-writer → v6-reader default resolution.
Reviewed changes
Copilot reviewed 2 out of 3 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| internal/venice-common/src/main/resources/avro/PushJobDetails/v6/PushJobDetails.avsc | Introduces v6 schema with two new defaulted timing fields. |
| internal/venice-common/src/test/java/com/linkedin/venice/status/protocol/TestPushJobDetailsSchemaCompatibility.java | Adds tests for staged v6 registration/activation assumptions and compatibility/default resolution. |
| build.gradle | Pins PushJobDetails compilation to v5 via versionOverrides while shipping v6 resources. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Review feedback pointed out that this PR claims controllers register v6 while the code still serializes v5, and that no such registration path exists. That is correct, the claim was wrong. With the versionOverrides pin the compiled schema is v5, and Utils.getAllSchemasFromResources breaks out of its resource walk as soon as it matches the compiled schema, so v6 is never even loaded into the schema map. Every registration path is then additionally bounded by AvroProtocolDefinition.currentProtocolVersion, which stays 5 here: SystemStoreInitializationHelper, SystemSchemaInitializationRoutine and ControllerClientBackedSystemSchemaInitializer.registerLocalSchemaResources all loop version = 1..getCurrentProtocolVersion(). InternalAvroSpecificSerializer builds its reader cache from the same map, so a build with this pin cannot serialize or deserialize v6 either. Registration only starts when the follow-up PR removes the pin and bumps the protocol definition to v6, so the deploy-controllers-before-push-jobs ordering is what keeps that rollout safe, not this PR. Making a staged schema register early was rejected: registering schemas above the current protocol version is deliberately excluded, and changing the shared initialization code would leak experimental schemas for every protocol. No behavior changes, this only corrects the build.gradle comment and the test naming and documentation so they describe what staging actually does, which is to freeze the schema in the resources while keeping it inert. Testing: testStagedV6IsRegisteredButNotActivated is renamed to testStagedV6IsNotActivatedAndNotVisibleAtRuntime and now asserts the invariant the reviewer was asking about, that the runtime serializer knows v5 and does not know v6. ./gradlew :internal:venice-common:test --tests "com.linkedin.venice.status.protocol.TestPushJobDetailsSchemaCompatibility" BUILD SUCCESSFUL - 3 tests, 3 passed Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 2 out of 3 changed files in this pull request and generated no new comments.
Suppressed comments (1)
internal/venice-common/src/test/java/com/linkedin/venice/status/protocol/TestPushJobDetailsSchemaCompatibility.java:61
- PR description says this change lets controllers register PushJobDetails v6 while code still serializes v5, but this test explicitly asserts the opposite: v6 is staged/inert and not visible to runtime discovery/serializer (so controllers won’t register v6 until the follow-up PR bumps PUSH_JOB_DETAILS to 6). Please update the PR description/rollout section to match the actual behavior to avoid incorrect deployment assumptions.
* v6 is staged, which means it is inert: the {@code versionOverrides} pin in the root build.gradle keeps the
* generated classes on v5, and {@link Utils#getAllSchemasFromResources(AvroProtocolDefinition)} stops at the
* compiled version, so no runtime path in this build reads, writes or registers v6. Controllers register value
* schemas only up to {@link AvroProtocolDefinition#currentProtocolVersion}, and the serializer only caches
* readers for the versions resource discovery returned, so both stop at v5 too.
Carries the correction from PR linkedin#2971 (0119fd1) into this stacked branch so PR 2 stays logically based on the updated PR 1 content. Merged rather than rebased because both branches are already pushed. Two conflicts, both in the files PR 1 touched and this PR then changes: - build.gradle: kept this branch's empty versionOverrides. PR 1 only reworded the comment attached to the pin, and this PR removes the pin entirely to activate v6, so there is nothing left to carry over. - TestPushJobDetailsSchemaCompatibility: kept this branch's activated test and carried PR 1's substance into it. The substance was that PR 1 does not register v6. Utils.getAllSchemasFromResources breaks out of its resource walk at the compiled schema, and every registration loop is bounded by AvroProtocolDefinition.currentProtocolVersion, so a build carrying the pin can neither register, serialize nor deserialize v6. This PR's javadoc repeated the same wrong claim ("v6 was registered by a previous PR"), so it is corrected here too: v6 was staged and inert, and this PR is what actually activates it. testV6IsActivatedAndOnlyAppendsTheTimingFields now asserts the mirror of the invariant PR 1 asserts. PR 1 asserts knownProtocols() does not contain v6 while staged; this asserts it does contain v6 once activated, and still contains v5 for in-flight push jobs. It also keeps PR 1's reverse-direction compatibility check, that a v5 reader can read a v6 record, which testPushJobStatusValueSchemaCompatibility does not cover. Testing: ./gradlew :internal:venice-common:test --tests "com.linkedin.venice.status.protocol.TestPushJobDetailsSchemaCompatibility" BUILD SUCCESSFUL - 3 tests, 3 passed Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Restores TestPushJobDetailsSchemaCompatibility to the base version, so this PR no longer touches it at all. PR scope is now exactly the v6 schema resource plus the versionOverrides pin in build.gradle. The tests this PR had been carrying all described the staged state: that the protocol definition and compiled class are still v5, that v6 is absent from the resolved schema map, and that the runtime serializer does not know v6. All of that asserts the absence of behavior, and it has to be deleted again by the follow-up PR the moment v6 is activated. The assertions that are actually worth keeping, that v6 only appends the two defaulted long fields and that a v6 reader resolves a v5 record's missing fields to -1, belong with the PR that activates v6, which is where they now live. The correction from the review feedback is kept where it is load-bearing: the build.gradle comment no longer claims that this PR makes controllers register v6. It does not. Utils.getAllSchemasFromResources stops at the compiled schema and every registration loop is bounded by AvroProtocolDefinition.currentProtocolVersion, so with the pin in place v6 is inert. Registration happens when the follow-up PR removes the pin. Testing: the existing TestPushJobDetailsSchemaCompatibility is unmodified and still passes with the v6 resource and the pin present, which is the regression that matters here, that adding the resource does not disturb the active v5 protocol. ./gradlew :internal:venice-common:test --tests "com.linkedin.venice.status.protocol.TestPushJobDetailsSchemaCompatibility" BUILD SUCCESSFUL - 1 test, 1 passed Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Carries 7943dc1 from PR linkedin#2971, which restores TestPushJobDetailsSchemaCompatibility to the base version so that PR is limited to the v6 schema resource and the build.gradle pin. One conflict, in that test file, resolved in favor of this branch. PR 1 reverted the file to the base version, this branch keeps its activated tests, so the file is byte-for-byte unchanged from e34abd2. That is the right outcome: with PR 1 no longer touching the file, every assertion about v6 now lands in the PR that actually activates v6, and the file shows one coherent diff against main instead of being added, rewritten and partly reverted across two PRs. The tests that were dropped from PR 1 were all assertions about the staged state, which PR 1 would have had to delete again here. The ones worth keeping already live on this branch: testV6IsActivatedAndOnlyAppendsTheTimingFields covers the appended fields, their -1 defaults and the reverse-direction v5-reads-v6 compatibility, and testV6ReaderResolvesMissingDurationsFromV5WriterToDefaults covers a v6 reader resolving a v5 record. Testing: ./gradlew :internal:venice-common:test --tests "com.linkedin.venice.status.protocol.TestPushJobDetailsSchemaCompatibility" --rerun-tasks BUILD SUCCESSFUL - 3 tests, 3 passed Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 1 out of 2 changed files in this pull request and generated no new comments.
Suppressed comments (1)
internal/venice-common/src/main/resources/avro/PushJobDetails/v6/PushJobDetails.avsc:93
- This PR introduces PushJobDetails v6 with two new fields and relies on a build pin to keep v6 staged/inert, but there are no unit tests in the repo asserting the staging behavior (v6 not visible via schema discovery/serializer) or v5↔v6 compatibility for the new fields. Add/extend schema compatibility tests so the staging contract and default resolution (-1) are enforced.
{
"name": "externalStorageWriteTimeMs",
"doc": "Total elapsed time in milliseconds, summed across all successful data writer task outputs of this push, spent in the external storage (Spaniel/TiKV) write path: throttling wait, batchPut calls including retries and retry backoff, external flush and external close. This is a sum of per-task wall-clock durations, NOT the push job's overall wall-clock duration, so with N parallel tasks it can be up to N times the push duration. -1 when the push did not report it (no dual write configured, or an older push job version).",
"type": "long",
"default": -1
},
{
"name": "veniceWriteTimeMs",
"doc": "Total elapsed time in milliseconds, summed across all successful data writer task outputs of this push, spent invoking the Venice/Kafka writes and flushing/closing the Venice writer. This is a sum of per-task wall-clock durations, NOT the push job's overall wall-clock duration. -1 when the push did not report it.",
"type": "long",
"default": -1
…ash conflicts PR linkedin#2967 was squash-merged to main as 5b45502, while this branch still carried it as two unsquashed commits (6e3dff2, c4c0e42). The two histories touch the same files, so GitHub reported 19 conflicting paths and the PR was unmergeable. Rebasing to drop the duplicates would need a force push, so this is a normal merge instead. 17 paths conflicted. Main's copy of linkedin#2967 is treated as authoritative throughout, since it is the version that was actually reviewed and merged and it has drifted slightly from the copy on this branch (a fifth ControllerRoute param, extra dimension-test coverage, wider DualWriteVeniceWriter and integration-test changes). Each conflicted file was resolved by taking main's content and reapplying only this branch's timing delta on top, computed as the diff between c4c0e42 (this branch's pre-timing state) and f620848 (its post-timing state). 16 of 17 merged cleanly that way. The remaining one, DualWriteVeniceWriterTest, conflicted only on style: main imports Set and HashSet while this branch had spelled them java.util.Set and java.util.HashSet. Main's imported form is kept and the four timing counters are retained alongside failedExternalStorageRegions. Verified afterwards that for 27 of the 29 files in the final diff, the diff against main is byte-identical to the isolated timing delta, so no linkedin#2967 content is duplicated or lost. The two intentional exceptions are the PushJobDetails v6 resource and TestPushJobDetailsSchemaCompatibility, which come from PR linkedin#2971 and are unchanged from this branch's previous head. build.gradle correctly drops out of the diff entirely: PR linkedin#2971 adds the versionOverrides pin and this PR removes it, so against today's main the two cancel out. The timing feature is intact: PushJobDetails v6 activation, the DualWriteVeniceWriter external and Venice leg timings covering throttling, batchPut retries, flush and close, Spark task-output aggregation without accumulators, the MR counters, and the controller metrics with Avg and Max Tehuti stats only. Testing: ./gradlew spotlessCheck BUILD SUCCESSFUL :clients:venice-push-job:test (DualWriteVeniceWriterTest, MapReduceDataWriterTaskTrackerTest, SparkDataWriterTaskTrackerTest, AbstractDataWriterSparkJobTest, VenicePushJobLifecycleTest) :internal:venice-common:test (TestPushJobDetailsSchemaCompatibility, ControllerRouteDimensionTest, ReadOnlyStoreTest) :internal:venice-client-common:test (VeniceMetricsDimensionsTest, VenicePushJobDataWriterSinkTest) :services:venice-controller:test (TestPushJobStatusStats, PushJobStatusStatsOtelTest, PushJobOtelMetricEntityTest, PushJobTehutiMetricNameEnumTest, StoresRoutesTest) 49 passed All green, including the linkedin#2967 fail-open tests that now run against main's implementation. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
…nternal storage-system names from OSS doc Replace the two flat fields (externalStorageWriteTimeMs, veniceWriteTimeMs) with a single additionalPushMetrics map<string, long>, so future push-job metrics (retry counts, per-batch latency, etc.) can be added without evolving this schema again. Also drop the 'Spaniel/TiKV' external-storage-system names from the field doc, since internal system names shouldn't be referenced in the OSS repo. ./gradlew :internal:venice-common:compileJava -- BUILD SUCCESSFUL ./gradlew :internal:venice-common:test --tests "*TestPushJobDetailsSchemaCompatibility*" -- 1 test, 1 passed Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 1 out of 2 changed files in this pull request and generated no new comments.
Suppressed comments (1)
internal/venice-common/src/main/resources/avro/PushJobDetails/v6/PushJobDetails.avsc:87
- The PR description says PushJobDetails v6 appends two top-level fields (
externalStorageWriteTimeMs/veniceWriteTimeMs, default -1). However, the v6 schema being added here appends a single nullable map field (additionalPushMetrics) and encodes those metrics as map keys (with “missing key” semantics), and the two top-level fields do not exist. Please align the implementation and the stated rollout/compatibility story by either (a) changing the schema to add the two explicit fields as described, or (b) updating the PR description (and any related docs) to match the map-based schema evolution and its defaults/"not reported" semantics.
{
"name": "additionalPushMetrics",
"doc": "Extensible map of additional push-job metrics keyed by metric name, so new metrics (e.g. per-destination write timings, error-retry counts, per-batch latency) can be added without evolving this schema again. Values reported so far: 'externalStorageWriteTimeMs' - total elapsed time in milliseconds, summed across all successful data writer task outputs of this push, spent in the external storage write path (throttling wait, batch write calls including retries and retry backoff, flush and close); 'veniceWriteTimeMs' - total elapsed time in milliseconds, summed across all successful data writer task outputs of this push, spent invoking the Venice/Kafka writes and flushing/closing the Venice writer. Both are sums of per-task wall-clock durations, NOT the push job's overall wall-clock duration, so with N parallel tasks they can be up to N times the push duration. A missing key means the push did not report that metric (no dual write configured, or an older push job version). Null when no additional metrics were reported.",
"type": ["null", {"type": "map", "values": "long"}],
"default": null
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Summary
Stage
PushJobDetailsv6 with an extensibleadditionalPushMetricsmap.The map stores optional
longmetrics by name. The first consumers are:externalStorageWriteTimeMs: summed task time for external writes, including throttling, retries, flush, and close.veniceWriteTimeMs: summed task time for Venice writes, flush, and close.These values are sums across successful data-writer task outputs, not the push job's overall duration. Missing keys mean the metric was not reported, and
nullmeans no additional metrics were reported.Staged rollout
This PR adds the v6 schema and pins generated
PushJobDetailsclasses to v5 throughversionOverrides. It does not activate v6 or change runtime serialization.Follow-up PR #2972 removes the pin, activates v6, and adds VPJ collection and controller metric emission. Controllers must deploy with v6 support before upgraded push jobs send v6 payloads.
Testing Done
No existing tests were changed.