aliehsaeedii opened a new pull request, #22966:
URL: https://github.com/apache/kafka/pull/22966
## Summary
Wires the stream-stream outer join up to the headers-aware `ListValueStore`,
and carries
per-record headers through the non-joined (spurious-result) emit.
**This is the only PR of the three that changes behaviour.** It is the
switch that makes the
storage and changelog layers from PR 1 reachable; it cannot compile without
them.
> ⚠️ **Stacked on #22961.** Until that merges, the diff shown here against
`trunk` also contains
> #22961's changes. The changes belonging to *this* PR are only the 5 files
listed below.
## Changes
- **`OuterStreamJoinStoreFactory`** — in HEADERS mode, wrap the value serde
in
`AggregationWithHeadersSerde` and build through the headers-aware
`ListValueStoreBuilder`.
For a persistent RocksDB store, pick
`RocksDBListValueHeadersBytesStoreSupplier` so an existing
store written by a pre-headers (PLAIN) version upgrades in place instead
of being misread.
In-memory and user-supplied stores keep their supplier; their upgrade is
handled on restore by
the list-aware `RecordConverter`.
The `DslStoreSuppliers` request stays `DslStoreFormat.PLAIN` even in
HEADERS mode. That is
deliberate: the value-with-headers stores put a single headers section in
front of the whole
value, which is not the shape of a list — here each element carries its
own headers inline, so
the generic whole-value converter would corrupt the blob.
- **`KStreamKStreamJoin`** / **`OuterJoinStoreWrapper`** — read the
per-element headers back out of
the store and attach them to the record emitted for a non-joined record.
Without this, headers
are silently dropped for exactly the records that spent time in the
outer-join store.
## Testing
- `KStreamKStreamOuterJoinTest` — non-joined emits carry their original
headers, for both store
formats.
- `OuterJoinListValueStoreRestorationTest` (integration) — 8 cases across
both processing
guarantees:
- restoration with multiple records, PLAIN and HEADERS;
- headers surviving a wipe-and-rebuild-from-changelog cycle;
- **`testHeadersStoreChangelogIsReadableByAPlainStore`** — the downgrade
direction: write in
HEADERS mode, wipe local state, restart as PLAIN, and assert the
restored records are still
left records with byte-identical values. Verified to be a real
regression test: with the
changelog strip from PR 1 disabled, this fails with
`expected: <[key0..key9]> but was: <[]>` — not one of the ten left
records survives, every one
comes back as a right record, silently.
Checkstyle and spotbugs pass.
## Relationship to the other PRs
| PR | scope | changes behaviour? |
|---|---|---|
| **#22961** | storage + changelog format | no — nothing constructs it |
| **this one** | DSL wiring + non-joined emit | **yes** |
| **PR 3** | tests-only: list-store format and headers round-trip | no |
There is no "must ship together" hazard: #22961 is inert on its own, and
this PR cannot compile
without it, so no release can contain the switch without the format support.
## Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
Reviewers: Matthias J. Sax <[email protected]>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
