[compat] Make maxNearlineRecordSizeBytes a per-version config

[KaiSernLim]

Stacked on #2991 ([protocol][build] Stage StoreMetaValue v48). This PR activates the staged schema: it removes the compileAvro versionOverrides pin and bumps AvroProtocolDefinition.METADATA_SYSTEM_SCHEMA_STORE 47→48, then adds the read/snapshot logic. It is self-contained and green on its own; #2991 is the isolated schema-only precursor for reviewers who want to look at the wire change in isolation. Merge #2991 first, then this rebases to a pin-removal + logic diff.

Why

Nearline (realtime-topic partial-update) ingestion is continuous. Reading the oversized-record limit live from the store would retroactively change enforcement for already in-flight versions. This makes maxNearlineRecordSizeBytes a per-version config, snapshotted onto the Version at version-creation time (mirroring chunkingEnabled / storageMode), so a later store-config change only affects future versions.

Batch maxRecordSizeBytes is intentionally left as a live store-level read — a batch push writes a version exactly once, so its value is already de-facto frozen per version.

Wire contract

StoreVersion.maxNearlineRecordSizeBytes is a nullable union ["null","int"], default null:

• null = pre-v48 version, no snapshot → the server falls back to the live store-level value at runtime (preserves behavior for metadata written before v48).
• non-null, including -1 = an immutable per-version snapshot; later store-level changes never affect it (-1 = store was unset at creation, backfilled to the server default at runtime).

Changes

• Schema/protocol — remove the v47 pin in build.gradle, bump METADATA_SYSTEM_SCHEMA_STORE 47→48 (schema itself introduced in [protocol][compat] Stage StoreMetaValue v48 (version-level maxNearlineRecordSizeBytes) #2991).
• Version / VersionImpl — Integer get/setMaxNearlineRecordSizeBytes backed by the nullable StoreVersion field; carried through cloneVersion().
• ReadOnlyStore — delegating getter (setter throws) + convertVersion reverse mapping.
• AbstractStore.addVersion — snapshot the store-level value onto the new version.
• Server read path — LeaderFollowerStoreIngestionTask.getMaxNearlineRecordSizeBytes() reads the per-version snapshot via a new StoreIngestionTask.getVersion() accessor, with the null→live-store fallback above.
• Tests — TestZKStore: snapshot + immutability, null default, cloneVersion preservation, and a cloneStoreProperties().getVersions() round-trip. LeaderFollowerStoreIngestionTaskTest: version→store→fleet-default fallback and no-retroactive-change.

VALIDATION_OVERRIDE

The enforce-lines-added check rejects a PR that touches both a src/main .avsc and a src/main .java file. This PR does, because it activates schema v48 (present in the diff via the stacked #2991 commit) in the same change that adds the consuming Java — they must be atomic so the generated StoreVersion version matches the registered METADATA_SYSTEM_SCHEMA_STORE. Totals are well under the line caps; only the schema-plus-Java coupling rule is triggered, so VALIDATION_OVERRIDE is the sanctioned path.

Co-authored-by: Copilot App 223556219+Copilot@users.noreply.github.com

[Copilot]

Pull request overview

This PR snapshots maxNearlineRecordSizeBytes onto each Version at version-creation time so nearline (realtime-topic) oversized-record enforcement remains stable for an in-flight version, while leaving batch maxRecordSizeBytes as a live store-level read.

Changes:

• Add StoreMetaValue schema v47 with StoreVersion.maxNearlineRecordSizeBytes, bump metadata system schema version, and plumb the new field through Version/VersionImpl, cloning, and store/version conversions.
• Update ingestion read/merge paths to avoid assembling chunked values when the stored manifest indicates the record is already over the nearline ceiling; emit a new OTel counter + warning log on skips.
• Add/extend unit + integration tests covering snapshotting/immutability, chunk read veto behavior, ingestion skipping semantics, and metric entity registration.

Reviewed changes

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

Show a summary per file

File	Description
internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/endToEnd/TestNearlineLargeRecordSkipping.java	New multi-region E2E test validating nearline partial-update skipping and reset behavior.
internal/venice-common/src/test/java/com/linkedin/venice/meta/TestZKStore.java	Unit tests for version-level snapshotting/default/clone persistence.
internal/venice-common/src/main/resources/avro/StoreMetaValue/v47/StoreMetaValue.avsc	New schema version adding StoreVersion.maxNearlineRecordSizeBytes.
internal/venice-common/src/main/java/com/linkedin/venice/serialization/avro/AvroProtocolDefinition.java	Bump metadata system store schema version to v47.
internal/venice-common/src/main/java/com/linkedin/venice/meta/VersionImpl.java	Back get/setMaxNearlineRecordSizeBytes with Avro StoreVersion field; preserve in clone.
internal/venice-common/src/main/java/com/linkedin/venice/meta/Version.java	Add version-level getter/setter for nearline record size ceiling.
internal/venice-common/src/main/java/com/linkedin/venice/meta/ReadOnlyStore.java	Delegate version-level getter in ReadOnlyVersion; include field in version conversion.
internal/venice-common/src/main/java/com/linkedin/venice/meta/AbstractStore.java	Snapshot store-level nearline limit onto newly-added versions.
clients/da-vinci-client/src/test/java/com/linkedin/davinci/storage/chunking/ChunkingTest.java	Tests ensuring chunked reads are vetoed before chunk fetch/assembly when over ceiling.
clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ServerMetricEntityTest.java	Update expected metric entity count for new ingestion metric.
clients/da-vinci-client/src/test/java/com/linkedin/davinci/stats/ingestion/IngestionOtelMetricEntityTest.java	Register expectation for new nearline large-record skipped counter metric.
clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTaskTest.java	Unit tests for inclusive boundary, per-record behavior, and unlimited ceiling cases.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/storage/chunking/ChunkingUtils.java	Short-circuit chunk assembly when manifest indicates size ceiling exceeded.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/storage/chunking/ChunkedValueManifestContainer.java	Add ceiling + isSizeLimitExceeded() signaling to disambiguate null results.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/NoOpIngestionOtelStats.java	No-op override for the new large-record skipped metric.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/IngestionOtelStats.java	Add OTel metric entity state + recording method for large-record skipped count.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/ingestion/IngestionOtelMetricEntity.java	Define ingestion.partial_update.large_record_skipped_count metric entity.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/stats/AggVersionedIngestionStats.java	Add aggregator method to record the new nearline skip counter.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/StoreIngestionTask.java	Expose version snapshot via getVersion() for version-frozen config reads.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTask.java	Read nearline limit from version snapshot; enforce skip before assembly; emit metric/log.
clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/ActiveActiveStoreIngestionTask.java	Apply ceiling only to partial updates; discard merge output when oversized-record skip occurs.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

Pull request overview

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

Suppressed comments (1)

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTask.java:2998

• Switching nearline max-record enforcement to read from Version means versions created before the new StoreMetaValue/v48 schema will deserialize StoreVersion.maxNearlineRecordSizeBytes as the Avro default (-1). With the current backfill logic, that effectively forces those in-flight versions to use serverConfig.getDefaultMaxRecordSizeBytes(), ignoring any previously configured store-level maxNearlineRecordSizeBytes.

If this is meant to only affect versions created after the snapshot feature, consider adding a migration/backfill path keyed off the metadata system store value schema id (<48) so pre-v48 versions get initialized (once) with the store-level maxNearlineRecordSizeBytes at upgrade time (or introduce an explicit migration flag like the other store→version config migrations).

  protected int getMaxRecordSizeBytes() {
    return backfillRecordSizeLimit(storeRepository.getStore(storeName).getMaxRecordSizeBytes());
  }

  protected int getMaxNearlineRecordSizeBytes() {
    return backfillRecordSizeLimit(getVersion().getMaxNearlineRecordSizeBytes());
  }

[Copilot]

Pull request overview

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

[Copilot]

Pull request overview

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

Suppressed comments (2)

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTask.java:3003

• The version-level snapshot uses the same sentinel (-1 / VeniceWriter.UNLIMITED_MAX_RECORD_SIZE) as both (a) an explicitly snapshotted "unset" value for versions created after v48 (AbstractStore.addVersion copies the store-level value, which is often -1), and (b) the Avro default used when the field is missing on pre-v48 versions. With the current check, any v48+ version whose snapshot is -1 will fall back to the live store-level value, so changing the store config later can retroactively change enforcement for already-created versions—contradicting the per-version snapshot goal.

    // the upgrade instead of silently reverting to the fleet default. Versions created after the snapshot feature
    // always carry an explicit value, so this fallback never overrides an intentional per-version limit.
    int recordSizeLimit = (versionLevelLimit != VeniceWriter.UNLIMITED_MAX_RECORD_SIZE)
        ? versionLevelLimit
        : storeRepository.getStore(storeName).getMaxNearlineRecordSizeBytes();

internal/venice-common/src/main/resources/avro/StoreMetaValue/v48/StoreMetaValue.avsc:271

• Using a default of -1 for StoreVersion.maxNearlineRecordSizeBytes makes pre-v48 versions (field absent) indistinguishable from v48+ versions that explicitly snapshot an unset value (-1). Since the server logic treats -1 as "unset" and falls back to the live store-level config, this can reintroduce retroactive behavior for v48+ versions created while the store-level value was unset.

                    {"name": "maxNearlineRecordSizeBytes", "type": "int", "default": -1, "doc": "Version-level max record size in bytes for nearline jobs with partial updates, snapshotted from the store-level config when this version is created. If not specified (-1), the server config 'default.max.record.size.bytes' (100MB default) will be backfilled at runtime."}

[Copilot]

Pull request overview

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

Suppressed comments (1)

internal/venice-common/src/main/resources/avro/StoreMetaValue/v48/StoreMetaValue.avsc:98

• The field doc for clientDecompressionEnabled contains a typo ("le/Disable"). Since this v48 schema is newly introduced, consider correcting it here to improve generated-schema docs (the same typo also exists in the immutable v47 schema).

            {"name": "clientDecompressionEnabled", "type": "boolean", "default": true, "doc": "le/Disable client-side record decompression (default: true)"},

[Copilot]

Pull request overview

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

Suppressed comments (1)

internal/venice-common/src/main/resources/avro/StoreMetaValue/v48/StoreMetaValue.avsc:171

• This new v48 store-level doc says editing the value only affects versions created afterward, but the implementation intentionally treats a null version snapshot (pre-v48 versions) as a signal to read this live store value. Therefore a store-level edit does affect those pre-v48 versions retroactively. Please qualify the statement to distinguish non-null snapshots from versions without a snapshot.

            {"name": "maxNearlineRecordSizeBytes", "type": "int", "default": -1, "doc": "Store-level source for the nearline max record size (bytes) for partial-update jobs. This value is snapshotted onto each StoreVersion.maxNearlineRecordSizeBytes at version-creation time, so editing it only affects versions created afterward; the server enforces the per-version snapshot, not this store-level value directly. If not specified (-1), the server config 'default.max.record.size.bytes' (100MB default) is backfilled. This may converge with maxRecordSizeBytes in the future"},

[Copilot]

Pull request overview

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

Suppressed comments (2)

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTask.java:3000

• This new per-version getter is not used by the production ingestion path: StoreIngestionTask.recordAssembledRecordSize() calls calculateAssembledRecordSizeRatio(), whose implementation below still uses getMaxRecordSizeBytes(), and there are no other production callers of getMaxNearlineRecordSizeBytes(). As a result, changing the version snapshot has no effect on nearline record-size handling despite the fallback logic being unit-tested. Please wire this value into the actual nearline path while retaining the batch path's live limit.

    Integer versionLevelLimit = getVersion().getMaxNearlineRecordSizeBytes();
    // A null snapshot means the version was created before this field existed (StoreMetaValue < v48); fall back to
    // the live store-level value so a previously configured nearline limit survives the upgrade instead of reverting
    // to the fleet default. A non-null snapshot (including -1) is used as-is, so a later store-level change never
    // retroactively alters enforcement for an already-created version.

internal/venice-common/src/main/java/com/linkedin/venice/meta/ReadOnlyStore.java:2090

• The new serialization mapping is not covered by the added tests. TestZKStore only reads the field from the in-memory Version, while ReadOnlyStoreTest currently asserts version count but not version fields; a regression in this line would silently drop the snapshot when ReadOnlyStore.cloneStoreProperties() writes metadata. Add a round-trip assertion for a non-null value (and ideally -1/null) on cloned.getVersions().get(0), following the existing version-field tests in ReadOnlyStoreTest around lines 330–348.

    storeVersion.setMaxNearlineRecordSizeBytes(version.getMaxNearlineRecordSizeBytes());

[Copilot]

Pull request overview

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

Suppressed comments (1)

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/LeaderFollowerStoreIngestionTask.java:2999

• This snapshot-aware accessor is not used by any production ingestion path: StoreIngestionTask.recordAssembledRecordSize() calls calculateAssembledRecordSizeRatio(), and this class still implements that with getMaxRecordSizeBytes() (and the ratio reporter repeats the same live store-level lookup). As a result, changing StoreVersion.maxNearlineRecordSizeBytes has no runtime effect, so the per-version nearline limit described by this PR is not actually enforced. Wire this accessor into the hybrid/partial-update path while retaining the batch path's maxRecordSizeBytes behavior, and cover that path rather than only testing the accessor directly.

  protected int getMaxNearlineRecordSizeBytes() {
    Integer versionLevelLimit = getVersion().getMaxNearlineRecordSizeBytes();
    // A null snapshot means the version was created before this field existed (StoreMetaValue < v48); fall back to
    // the live store-level value so a previously configured nearline limit survives the upgrade instead of reverting
    // to the fleet default. A non-null snapshot (including -1) is used as-is, so a later store-level change never

[Copilot]

Pull request overview

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

[sixpluszero]

lgtm, thank you for the changes!