IGNITE-27977 Refactor bytes serialization for DataStreamerRequest

[anton-vinogradov]

DataStreamerRequest.updaterBytes was a hand-marshalled blob: DataStreamerImpl produced
the bytes while building the request, and DataStreamProcessor unmarshalled them with a
class loader it had just resolved.

Why this matters

A field that marshals itself picks its own marshaller, so it stays outside whatever the
transport decides — which is what IGNITE-28940 has to decide in one place. The generated
marshaller is that place. updaterBytes was the last such blob on this path.

Change

The updater travels in DataStreamerReceiverMessage, which carries it in whichever form
suits it — the same way GridTopicMessage holds either the ordinal of a known topic or a
serialized arbitrary one:

• an updater the streamer ships with is named: DataStreamerBuiltInUpdater covers
  ISOLATED, INDIVIDUAL, BATCHED and BATCHED_SORTED, and the receiving node uses its
  own instance. Every node has these classes, so serializing one into each request only to
  deserialize an identical copy was pure waste;
• a user updater is carried as an @Marshalled pair of the object and its bytes.

The carrier is marked @UseBinaryMarshaller: it is the one holding a user class, and the
hand-written call used ctx.marshaller() — the very marshaller that annotation selects, so
the bytes are produced exactly as before. DataStreamProcessor reads the message through
MessageMarshalling.unmarshal and drops its own Marshaller field.

Why the read stays with the consumer

The message remains a DeferredUnmarshalMessage. Its class loader does not follow from a
carried deployment alone: with forced local deployment it is the grid class loader, and
otherwise it comes from the global deployment of the sender. So the processor passes the
loader explicitly and keeps the read inside its existing try, where a missing deployment
is reported back to the sender instead of leaving it waiting for a timeout.

Marshalling a user updater once

Measured on one node: 281 ns to marshal IsolatedUpdater (12 B), 810 ns for
StreamTransformer.from(ep) (299 B), 9.3 µs for a receiver holding 10K of state (10270 B),
against 76 µs to build and marshal a batch of 512 entries. Marshalling per batch would cost
up to 12% of a full batch, so the streamer keeps one carrier per receiver and the generated
marshaller fills its bytes once. Nothing is allocated for the built-in updaters at all:
their carriers are constants of the enum, so flipping allowOverwrite back and forth
allocates nothing, and a streamer that never touches the setting builds no carrier either.

DataStreamerImplSelfTest covers all three: testReceiverMarshalledOncePerStreamer (every
request carries the same bytes), testBuiltInUpdaterIsNotSent (the default receiver is
named, and the data still lands) and testBuiltInReceiverIsNotSent (same for one set
explicitly).

Wire format

Field 3 of the request becomes a nested message instead of a byte array, and the carrier
takes the next free id in the datastreamer group. The carrier itself has two fields: the
serialized user updater, and the name of a built-in one.

No client sees this. DataStreamerRequest travels between Java nodes only: the thin client
has its own ClientDataStreamerRequest with its own protocol, and the platform clients go
through the embedded node. Within 2.19 the layout of this message has already been reshaped
three times — IGNITE-27862, IGNITE-27824, IGNITE-28719 — and IGNITE-28528 reshaped it again.
IGNITE-27824 is the same move on the same class: it replaced resTopicBytes, the other
hand-marshalled blob here, with a nested message, and the TODO naming this ticket sat on
that very field.

Along the way

• The receiver field of the streamer was mutated from the user thread and read by the
  sending ones without being volatile; holding it in one carrier fixes that publication.
  Its bytes are volatile for the same reason: the batch that marshals first writes them
  and the others read them.
• The serialized receiver used to be cached without ever being invalidated, so a receiver
  replaced mid-stream took effect locally, where the live field is read, but not remotely,
  where the stale bytes kept going. Both paths now see the same object.
• allowOverwrite is the receiver choice it always was: the getter compares against the
  isolated updater, the setter goes through receiver(...), and the field is empty until
  someone sets one, so "nothing set means isolated" is stated once rather than assigned up
  front.
• The carrier is @GridToStringExclude in the request: it holds a user object, the request
  is printed under debug logging on the sending side, and GridToStringBuilder rethrows
  whatever a field toString throws. GridJobExecuteRequest excludes its user objects the
  same way.

What this changes for the entries

Reading the message reads all of it, so entries now pass through the generated marshaller
on the receiving side too, with the cache object context resolved from cacheId and the
deployment class loader. DataStreamerUpdateJob still unmarshals them once more under the
global loader; that pass is now a no-op, since CacheObject.unmarshal only acts when the
value is absent. It is kept because the job also runs on the local path, where no message
was ever unmarshalled, and because the same loop carries the security permission checks.

The streamer also still marshals keys and values by hand before sending. That is not a
duplicate of the generated marshal: it uses the cache object context the streamer holds from
its creation, whereas the generated code resolves the context by cacheId and would skip the
work if the cache were destroyed in between.

Verified

129 tests, no failures: the whole processors.datastreamer package — with
IgniteDataStreamerPerformanceTest left out as an endless benchmark that always times out —
plus P2PStreamingClassLoaderTest, P2PClassLoadingFailureHandlingTest,
GridP2PContinuousDeploymentSelfTest, ClassLoadingProblemExceptionTest,
MessageMarshalOnceTest, IgniteCoreMessagesSerializationTest,
DirectMarshallingMessagesTest and MessageProcessorTest. Also
mvn checkstyle:check -Pcheckstyle -pl modules/core.

🤖 Generated with Claude Code

[anton-vinogradov]

/runall

---

🚀 RunAll queued — build 9267111 · live progress & verdict: Ignite PR Checker. The verdict lands here when the run finishes.
🏁 Run finished — the verdict comment has the full story.

[anton-vinogradov]

Ignite PR Checker verdict · RunAll build 9267111 · 147 suites ran, 0 reused

⚠️ This run doesn't cover the PR fully:

• a newer run is still going — its unfinished suites can still fail

Everything below is what it did manage to say.

🔎 No blockers found — but the run above can't prove the PR is clean. 16 pre-existing/flaky tests filtered out. Re-run once the above is sorted out.

[Vladsz83]

Should StreamReceiverMessage be a MarshallableMessage?

[anton-vinogradov]

I don't think so: the marker means the opposite of what this class does. MarshallableMessage declares marshal(Marshaller) and unmarshal(Marshaller, ClassLoader) — it says "I carry a hand-written marshalling step, call it", and the generated marshaller does exactly that, on top of the fields. Implementing it here would mean writing that step by hand again, which is what this ticket removes.

StreamReceiverMessage has no such step: the @Marshalled pair is what makes the generator produce one. The other @Marshalled messages are the same — GridEventStorageRequest, GridJobExecuteRequest, StartRequestData, GenericValueMessage — none of them implements the interface.

If what you had in mind is the side effects the marker brings — the marshaller becoming mandatory at registration, and MessageUnmarshalOnceCheck covering the message — those would apply to every @Marshalled message equally, so it reads as a codegen-level decision rather than a property of this class. Happy to file it separately if you think the check should cover them.

[Vladsz83]

Serialized {@link #rcvr} is enough

[anton-vinogradov]

Cut it down, but kept one clause for the volatile:

/** Serialized {@link #rcvr}. Volatile: the batches sharing it are marshalled on different threads. */

Without it the modifier reads as unexplained — the field is written by whichever batch is marshalled first and read by the others, and dropping volatile would let a reader see the reference before the contents. If you would rather keep the javadoc to Serialized {@link #rcvr} and leave the reasoning to the reviewer of the modifier, say so and I will trim it.

[Vladsz83]

Do we need to keep instances in DataStreamerCacheUpdaters? Like DataStreamerCacheUpdaters#BATCHED_SORTED is in so nessesary? Maybe we should simplify code and return new BatcherSorted() (and others) and used the on demant as the message. WDYT?

[anton-vinogradov]

Good catch on the coupling — the lookup no longer depends on it:

// Matches by class, so an updater built anew is still recognized as built-in: these updaters hold no state.
if (builtIn.updater.getClass() == updater.getClass())

So whatever DataStreamerCacheUpdaters decides to do with its instances, a built-in updater keeps travelling as a name.

On the instances themselves I would leave them as they are, and not in this ticket:

• they are handed out by public factories, so individual() and friends returning a new object each time changes what callers get — anyone comparing receivers, ours included until this commit, would notice;
• a receiver is typically set once per streamer, but nothing stops a caller from doing it per streamer in a loop, and allocating a stateless object for that is a step back;
• it is unrelated to what this ticket does, and the gain is three objects per JVM.

Happy to file it separately if you think the factories should change.

[Vladsz83]

The only spot I see is merging somehow DataStreamerBuiltInUpdater and DataStreamerCacheUpdaters. The look too similarly.

[anton-vinogradov]

They do sit close, but they answer different questions, and I am not sure which way you want them merged — so let me lay out what I see.

DataStreamerCacheUpdaters is public API: factories a user calls to pick an updater, plus the implementations behind them. DataStreamerBuiltInUpdater is a wire-level registry: it exists so a request can name an updater instead of sending it, and it has to cover the Isolated one too, which lives in DataStreamerImpl and is not part of that public class.

So a full merge would mean moving IsolatedUpdater out of DataStreamerImpl into the public class — a bigger change than this ticket, and one that touches what the class means.

What is cheap is nesting: make the enum DataStreamerCacheUpdaters.BuiltIn, package-private, so there is one top-level name instead of two similar ones. It still points at DataStreamerImpl.ISOLATED_UPDATER for that constant.

Tell me if the nested variant is what you meant and I will do it; if you had the full merge in mind, I would rather file it separately.

[github-actions]

Possible compatibility issues. Please, check rolling upgrade cases

This PR modifies protected classes (with Order annotation).
Changes to these classes can break rolling upgrade compatibility.

Affected files:

• modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerReceiverMessage.java
• modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java