From 817073d07d01781f083d4728efbbe97fd2201542 Mon Sep 17 00:00:00 2001 From: Anton Vinogradov Date: Sun, 9 Aug 2026 02:55:07 +0300 Subject: [PATCH 01/10] IGNITE-27977 Refactor bytes serialization for DataStreamerRequest The stream receiver was marshalled by hand: the streamer produced the blob, and the processor unmarshalled it with a class loader it had just built. The pair is now an @Marshalled field, so the generated marshaller owns both directions. The class is marked @UseBinaryMarshaller: the receiver is a user class, and the hand-written call used ctx.marshaller(), which is the same schema-aware marshaller the annotation selects. The wire format is unchanged - updaterBytes stays @Order(3). The message stays a DeferredUnmarshalMessage. Its class loader does not come from a carried deployment alone: with forced local deployment it is the grid class loader. The processor therefore passes the loader explicitly and keeps the read inside its own try, so a missing deployment is still answered to the sender instead of leaving it waiting for a timeout. Co-Authored-By: Claude Opus 5 --- .../datastreamer/DataStreamProcessor.java | 12 ++++------ .../datastreamer/DataStreamerImpl.java | 11 +-------- .../datastreamer/DataStreamerRequest.java | 23 ++++++++++++------- .../DataStreamerImplSelfTest.java | 2 +- 4 files changed, 22 insertions(+), 26 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java index 464a74d82ee25..b2b7f85953dbe 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java @@ -27,6 +27,7 @@ import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; import org.apache.ignite.internal.managers.communication.GridIoManager; import org.apache.ignite.internal.managers.communication.GridMessageListener; +import org.apache.ignite.internal.managers.communication.MessageMarshalling; import org.apache.ignite.internal.managers.deployment.GridDeployment; import org.apache.ignite.internal.processors.GridProcessorAdapter; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; @@ -48,7 +49,6 @@ import org.apache.ignite.lang.IgniteClosure; import org.apache.ignite.lang.IgniteFuture; import org.apache.ignite.lang.IgniteInClosure; -import org.apache.ignite.marshaller.Marshaller; import org.apache.ignite.stream.StreamReceiver; import org.jetbrains.annotations.Nullable; @@ -68,9 +68,6 @@ public class DataStreamProcessor extends GridProcessorAdapter { /** Data Streamer flusher. */ private final DataStreamerFlusher flusher = new DataStreamerFlusher(); - /** Marshaller. */ - private final Marshaller marsh; - /** * @param ctx Kernal context. */ @@ -86,8 +83,6 @@ public DataStreamProcessor(GridKernalContext ctx) { } }); } - - marsh = ctx.marshaller(); } /** {@inheritDoc} */ @@ -240,7 +235,10 @@ private void processRequest(final UUID nodeId, final DataStreamerRequest req) { StreamReceiver updater; try { - updater = U.unmarshal(marsh, req.updaterBytes(), U.resolveClassLoader(clsLdr, ctx.config())); + // The request carries user classes, so it is read here, with the deployment class loader at hand. + MessageMarshalling.unmarshal(req, ctx, null, U.resolveClassLoader(clsLdr, ctx.config())); + + updater = req.updater(); if (updater != null) ctx.resource().injectGeneric(updater); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java index 1040f06ddf83b..944193a281351 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java @@ -155,9 +155,6 @@ public class DataStreamerImpl implements IgniteDataStreamer, Delayed /** Cache receiver. */ private StreamReceiver rcvr = ISOLATED_UPDATER; - /** */ - private byte[] updaterBytes; - /** IO policy resovler for data load request. */ private IgniteClosure ioPlcRslvr; @@ -1943,12 +1940,6 @@ private void submit( if (val != null) val.marshal(cacheObjCtx); } - - if (updaterBytes == null) { - assert rcvr != null; - - updaterBytes = U.marshal(ctx, rcvr); - } } catch (IgniteCheckedException e) { U.error(log, "Failed to marshal.", e); @@ -1995,7 +1986,7 @@ private void submit( reqId, topicId, cacheName, - updaterBytes, + rcvr, entries, true, skipStore, diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java index 88af958a50373..933bdbba9417f 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java @@ -22,20 +22,24 @@ import java.util.UUID; import org.apache.ignite.configuration.DeploymentMode; import org.apache.ignite.internal.DeferredUnmarshalMessage; +import org.apache.ignite.internal.Marshalled; import org.apache.ignite.internal.Order; import org.apache.ignite.internal.StripedMessage; +import org.apache.ignite.internal.UseBinaryMarshaller; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; import org.apache.ignite.internal.processors.cache.GridCacheUtils; import org.apache.ignite.internal.util.tostring.GridToStringInclude; import org.apache.ignite.internal.util.typedef.internal.S; import org.apache.ignite.lang.IgniteUuid; import org.apache.ignite.plugin.extensions.communication.CacheIdAware; +import org.apache.ignite.stream.StreamReceiver; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import static org.apache.ignite.internal.GridTopic.TOPIC_DATASTREAM; -/** */ +/** A batch of streamed entries. The receiver is a user class, hence the deferred unmarshalling. */ +@UseBinaryMarshaller public class DataStreamerRequest implements DeferredUnmarshalMessage, CacheIdAware, StripedMessage { /** */ @Order(0) @@ -49,8 +53,11 @@ public class DataStreamerRequest implements DeferredUnmarshalMessage, CacheIdAwa @Order(2) String cacheName; - /** */ - // TODO: Refactor bytes serialization - IGNITE-27977 + /** Cache receiver. */ + @Marshalled("updaterBytes") + StreamReceiver updater; + + /** Serialized cache receiver. */ @Order(3) byte[] updaterBytes; @@ -112,7 +119,7 @@ public DataStreamerRequest() { * @param reqId Request ID. * @param resTopicId Response topic ID. * @param cacheName Cache name. - * @param updaterBytes Cache receiver. + * @param updater Cache receiver. * @param entries Entries to put. * @param ignoreDepOwnership Ignore ownership. * @param skipStore Skip store flag. @@ -130,7 +137,7 @@ public DataStreamerRequest( long reqId, IgniteUuid resTopicId, @Nullable String cacheName, - byte[] updaterBytes, + StreamReceiver updater, Collection entries, boolean ignoreDepOwnership, boolean skipStore, @@ -149,7 +156,7 @@ public DataStreamerRequest( this.reqId = reqId; this.resTopicId = resTopicId; this.cacheName = cacheName; - this.updaterBytes = updaterBytes; + this.updater = updater; this.entries = entries; this.ignoreDepOwnership = ignoreDepOwnership; this.skipStore = skipStore; @@ -180,8 +187,8 @@ String cacheName() { } /** @return Updater. */ - byte[] updaterBytes() { - return updaterBytes; + StreamReceiver updater() { + return updater; } /** @return Entries to update. */ diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java index 2bc777180e0ff..f8ade7da47394 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java @@ -692,7 +692,7 @@ private static class StaleTopologyCommunicationSpi extends TcpCommunicationSpi { req.requestId(), req.resTopicId, req.cacheName(), - req.updaterBytes(), + req.updater(), req.entries(), req.ignoreDeploymentOwnership(), req.skipStore(), From e3ba9b097bc8ef1f73bf8bfffc8de5fcc4cbc97d Mon Sep 17 00:00:00 2001 From: Anton Vinogradov Date: Sun, 9 Aug 2026 03:15:35 +0300 Subject: [PATCH 02/10] IGNITE-27977 Simplify the wording around the deferred read The class javadoc and the comment at the read said the same thing twice. The javadoc now states what the message is, and the comment states why the read waits for this point. Co-Authored-By: Claude Opus 5 --- .../internal/processors/datastreamer/DataStreamProcessor.java | 2 +- .../internal/processors/datastreamer/DataStreamerRequest.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java index b2b7f85953dbe..6e2d787a893b0 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java @@ -235,7 +235,7 @@ private void processRequest(final UUID nodeId, final DataStreamerRequest req) { StreamReceiver updater; try { - // The request carries user classes, so it is read here, with the deployment class loader at hand. + // Read here, not on the inbound pass: the deployment class loader is known only at this point. MessageMarshalling.unmarshal(req, ctx, null, U.resolveClassLoader(clsLdr, ctx.config())); updater = req.updater(); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java index 933bdbba9417f..a5f056e5be852 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java @@ -38,7 +38,7 @@ import static org.apache.ignite.internal.GridTopic.TOPIC_DATASTREAM; -/** A batch of streamed entries. The receiver is a user class, hence the deferred unmarshalling. */ +/** Batch of streamed entries. The receiver is a user class. */ @UseBinaryMarshaller public class DataStreamerRequest implements DeferredUnmarshalMessage, CacheIdAware, StripedMessage { /** */ From bf633f0da4df8537480dd68f6b3f24797355af87 Mon Sep 17 00:00:00 2001 From: Anton Vinogradov Date: Sun, 9 Aug 2026 03:26:42 +0300 Subject: [PATCH 03/10] IGNITE-27977 Keep the receiver out of the message toString The field holds a user object, and the request is printed on the sending side under debug logging. The blob it replaced printed as bytes, and GridToStringBuilder rethrows whatever a field toString throws, so a user toString could now break the logging path. GridJobExecuteRequest excludes its user objects the same way. Co-Authored-By: Claude Opus 5 --- .../internal/processors/datastreamer/DataStreamerRequest.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java index a5f056e5be852..c9b54dcce8bd4 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java @@ -28,6 +28,7 @@ import org.apache.ignite.internal.UseBinaryMarshaller; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; import org.apache.ignite.internal.processors.cache.GridCacheUtils; +import org.apache.ignite.internal.util.tostring.GridToStringExclude; import org.apache.ignite.internal.util.tostring.GridToStringInclude; import org.apache.ignite.internal.util.typedef.internal.S; import org.apache.ignite.lang.IgniteUuid; @@ -53,7 +54,8 @@ public class DataStreamerRequest implements DeferredUnmarshalMessage, CacheIdAwa @Order(2) String cacheName; - /** Cache receiver. */ + /** Cache receiver. A user object, kept out of the message {@code toString()}. */ + @GridToStringExclude @Marshalled("updaterBytes") StreamReceiver updater; From d92c10517a1b090995dc9c298b04ec5895f8d66d Mon Sep 17 00:00:00 2001 From: Anton Vinogradov Date: Sun, 9 Aug 2026 03:51:50 +0300 Subject: [PATCH 04/10] IGNITE-27977 Marshal the receiver once per streamer Handing the message the receiver instead of its bytes cost a marshal per batch, where the streamer used to marshal once and reuse the result. Measured on one node: 281 ns for IsolatedUpdater, 810 ns for StreamTransformer.from(ep), 9.3 us for a receiver holding 10K of state, against 76 us to build and marshal a batch of 512 entries - up to 12% of a batch, and a larger share of a small one. The streamer keeps the bytes the generated marshaller produced for the first request and hands them to the next one, which the marshaller then keeps instead of producing its own. It reuses a result rather than deciding how to obtain it, so the marshaller stays the one codegen picks. The bytes are paired with the receiver they belong to, so a receiver replaced mid-stream invalidates them by itself - no separate cache reset that a concurrent send could race with. This also closes the older mismatch, where the cache was never invalidated at all and a replaced receiver took effect locally but not remotely. Co-Authored-By: Claude Opus 5 --- .../datastreamer/DataStreamerImpl.java | 21 ++++++++- .../datastreamer/DataStreamerRequest.java | 16 +++++++ .../DataStreamerImplSelfTest.java | 47 +++++++++++++++++++ 3 files changed, 82 insertions(+), 2 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java index 944193a281351..c21bd94a98474 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java @@ -104,6 +104,7 @@ import org.apache.ignite.internal.util.tostring.GridToStringInclude; import org.apache.ignite.internal.util.typedef.CI1; import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.T2; import org.apache.ignite.internal.util.typedef.X; import org.apache.ignite.internal.util.typedef.internal.A; import org.apache.ignite.internal.util.typedef.internal.CU; @@ -155,6 +156,9 @@ public class DataStreamerImpl implements IgniteDataStreamer, Delayed /** Cache receiver. */ private StreamReceiver rcvr = ISOLATED_UPDATER; + /** A receiver paired with the bytes the marshaller produced for it; a replaced receiver invalidates the pair. */ + private volatile T2, byte[]> marshalledRcvr; + /** IO policy resovler for data load request. */ private IgniteClosure ioPlcRslvr; @@ -1982,11 +1986,13 @@ private void submit( if (topVer == null) topVer = ctx.cache().context().exchange().readyAffinityVersion(); + StreamReceiver rcvr0 = rcvr; + DataStreamerRequest req = new DataStreamerRequest( reqId, topicId, cacheName, - rcvr, + rcvr0, entries, true, skipStore, @@ -1998,11 +2004,22 @@ private void submit( dep != null ? dep.classLoaderId() : null, dep == null, topVer, - (rcvr == ISOLATED_UPDATER) ? partId : NO_STRIPE); + (rcvr0 == ISOLATED_UPDATER) ? partId : NO_STRIPE); + + // Every batch carries the same receiver, so it is marshalled once: the message is handed the bytes of + // an earlier request, and yields its own when it is the first to be marshalled. + T2, byte[]> marshalled = marshalledRcvr; + + byte[] rcvrBytes = marshalled != null && marshalled.get1() == rcvr0 ? marshalled.get2() : null; + + req.updaterBytes(rcvrBytes); try { ctx.io().sendToGridTopic(node, TOPIC_DATASTREAM, req, plc); + if (rcvrBytes == null) + marshalledRcvr = new T2<>(rcvr0, req.updaterBytes()); + if (log.isDebugEnabled()) log.debug("Sent request to node [nodeId=" + node.id() + ", req=" + req + ']'); } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java index c9b54dcce8bd4..6538752a23481 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java @@ -193,6 +193,22 @@ String cacheName() { return updater; } + /** @return Serialized updater, {@code null} until the message is marshalled. */ + byte[] updaterBytes() { + return updaterBytes; + } + + /** + * Hands the message the serialized form of its updater, so the marshaller keeps it instead of producing its own. + * The receiver of a streamer does not change between batches, hence the sender marshals it once and passes the + * result on. + * + * @param updaterBytes Serialized updater taken from an already marshalled request. + */ + void updaterBytes(byte[] updaterBytes) { + this.updaterBytes = updaterBytes; + } + /** @return Entries to update. */ Collection entries() { return entries; diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java index f8ade7da47394..d68b9150ec8ff 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java @@ -24,6 +24,7 @@ import java.util.Map; import java.util.Random; import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.TimeUnit; @@ -88,6 +89,9 @@ public class DataStreamerImplSelfTest extends GridCommonAbstractTest { /** Indicates whether we need to make the topology stale */ private static boolean needStaleTop = false; + /** Collects the serialized receiver of every streamer request sent, when set. */ + private static volatile Collection sentUpdaterBytes; + /** {@inheritDoc} */ @Override protected void afterTest() throws Exception { super.afterTest(); @@ -142,6 +146,42 @@ public void testCloseWithCancellation() throws Exception { assertTrue(fut.isDone()); } + /** + * The receiver does not change between batches, so it is marshalled once: every request carries the very bytes + * produced for the first one. + * + * @throws Exception If failed. + */ + @Test + public void testReceiverMarshalledOncePerStreamer() throws Exception { + cnt = 0; + + startGrids(2); + + Collection sent = new ConcurrentLinkedQueue<>(); + + sentUpdaterBytes = sent; + + try (IgniteDataStreamer ldr = grid(0).dataStreamer(DEFAULT_CACHE_NAME)) { + ldr.perNodeBufferSize(1); + + for (int i = 0; i < KEYS_COUNT; i++) + ldr.addData(i, i); + } + finally { + sentUpdaterBytes = null; + } + + assertTrue("Expected more than one request to a remote node, got " + sent.size(), sent.size() > 1); + + byte[] first = F.first(sent); + + assertNotNull(first); + + for (byte[] bytes : sent) + assertTrue("The receiver was marshalled more than once", first == bytes); + } + /** * Test inconsistency log warning of the streamer. Default receiver goes first and is set again after a consistent * receiver. The warning must appear only once. @@ -670,6 +710,13 @@ private CacheConfiguration cacheConfiguration() { private static class StaleTopologyCommunicationSpi extends TcpCommunicationSpi { /** {@inheritDoc} */ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure ackC) { + Collection updaterBytes = sentUpdaterBytes; + + // The message is already marshalled at this point, so the serialized receiver is in place. + if (updaterBytes != null && msg instanceof GridIoMessage + && ((GridIoMessage)msg).message() instanceof DataStreamerRequest) + updaterBytes.add(((DataStreamerRequest)((GridIoMessage)msg).message()).updaterBytes()); + // Send stale topology only in the first request to avoid indefinitely getting failures. if (needStaleTop) { if (msg instanceof GridIoMessage) { From 29cf297d09b742eb595fdc35b92bdfe9056a1aca Mon Sep 17 00:00:00 2001 From: Anton Vinogradov Date: Sun, 9 Aug 2026 04:09:04 +0300 Subject: [PATCH 05/10] IGNITE-27977 Carry the receiver in a message of its own Holding the receiver in the request meant its serialized form belonged to the request, while the object belonged to the streamer. A request is one batch, the receiver lives for the whole stream, so keeping the bytes cost either a marshal per batch or a cache beside the streamer - a cache that had to be invalidated by hand and published safely. The receiver now travels in StreamReceiverMessage, where the object and its bytes sit together and live exactly as long as the receiver does. The streamer holds one instance and puts it into every request, so the generated marshaller fills the bytes for the first batch and the rest find them already there. Replacing the receiver builds another instance, which invalidates the old bytes by construction. This also removes an older race: the receiver field was mutated from the user thread and read by the sending ones without being volatile. The wire format of the request changes: field 3 is now a nested message rather than a byte array. Co-Authored-By: Claude Opus 5 --- .../ignite/internal/CoreMessagesProvider.java | 2 + .../datastreamer/DataStreamerImpl.java | 47 +++++++--------- .../datastreamer/DataStreamerRequest.java | 31 ++--------- .../datastreamer/StreamReceiverMessage.java | 55 +++++++++++++++++++ .../DataStreamerImplSelfTest.java | 26 ++++----- 5 files changed, 93 insertions(+), 68 deletions(-) create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/StreamReceiverMessage.java diff --git a/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java b/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java index a460ed789d24b..66be5c98df724 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java @@ -210,6 +210,7 @@ import org.apache.ignite.internal.processors.datastreamer.DataStreamerEntry; import org.apache.ignite.internal.processors.datastreamer.DataStreamerRequest; import org.apache.ignite.internal.processors.datastreamer.DataStreamerResponse; +import org.apache.ignite.internal.processors.datastreamer.StreamReceiverMessage; import org.apache.ignite.internal.processors.marshaller.MappedName; import org.apache.ignite.internal.processors.marshaller.MappingAcceptedMessage; import org.apache.ignite.internal.processors.marshaller.MappingProposedMessage; @@ -662,6 +663,7 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh) { register(DataStreamerEntry.class); register(DataStreamerRequest.class); register(DataStreamerResponse.class); + register(StreamReceiverMessage.class); // [11900 - 12000]: Metrics, monitoring messages. msgIdx = 11900; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java index c21bd94a98474..364d57c2f5f04 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java @@ -104,7 +104,6 @@ import org.apache.ignite.internal.util.tostring.GridToStringInclude; import org.apache.ignite.internal.util.typedef.CI1; import org.apache.ignite.internal.util.typedef.F; -import org.apache.ignite.internal.util.typedef.T2; import org.apache.ignite.internal.util.typedef.X; import org.apache.ignite.internal.util.typedef.internal.A; import org.apache.ignite.internal.util.typedef.internal.CU; @@ -153,11 +152,8 @@ public class DataStreamerImpl implements IgniteDataStreamer, Delayed /** Amount of permissions should be available to continue new data processing. */ private static final int REMAP_SEMAPHORE_PERMISSIONS_COUNT = Integer.MAX_VALUE; - /** Cache receiver. */ - private StreamReceiver rcvr = ISOLATED_UPDATER; - - /** A receiver paired with the bytes the marshaller produced for it; a replaced receiver invalidates the pair. */ - private volatile T2, byte[]> marshalledRcvr; + /** Cache receiver, in the message that carries it to the remote nodes. */ + private volatile StreamReceiverMessage rcvrMsg = new StreamReceiverMessage(ISOLATED_UPDATER); /** IO policy resovler for data load request. */ private IgniteClosure ioPlcRslvr; @@ -490,12 +486,18 @@ public IgniteInternalFuture internalFuture() { @Override public void receiver(StreamReceiver rcvr) { A.notNull(rcvr, "rcvr"); - this.rcvr = rcvr; + rcvrMsg = new StreamReceiverMessage(rcvr); + } + + /** @return Cache receiver. */ + @SuppressWarnings("unchecked") + private StreamReceiver rcvr() { + return (StreamReceiver)rcvrMsg.receiver(); } /** {@inheritDoc} */ @Override public boolean allowOverwrite() { - return rcvr != ISOLATED_UPDATER; + return rcvr() != ISOLATED_UPDATER; } /** {@inheritDoc} */ @@ -508,7 +510,7 @@ public IgniteInternalFuture internalFuture() { if (node == null) throw new CacheException("Failed to get node for cache: " + cacheName); - rcvr = allow ? DataStreamerCacheUpdaters.individual() : ISOLATED_UPDATER; + rcvrMsg = new StreamReceiverMessage(allow ? DataStreamerCacheUpdaters.individual() : ISOLATED_UPDATER); } /** {@inheritDoc} */ @@ -656,7 +658,7 @@ public IgniteFuture addDataInternal(Collection e lock(false); - if (rcvr instanceof IsolatedUpdater && inconsistencyWarned.compareAndSet(false, true)) + if (rcvr() instanceof IsolatedUpdater && inconsistencyWarned.compareAndSet(false, true)) log.warning(WRN_INCONSISTENT_UPDATES); try { @@ -890,9 +892,9 @@ private void load0( if (cacheObjCtx.addDeploymentInfo()) jobPda = new DataStreamerPda(key.value(cacheObjCtx, false), entry.getValue() != null ? entry.getValue().value(cacheObjCtx, false) : null, - rcvr); - else if (rcvr != null) - jobPda = new DataStreamerPda(rcvr); + rcvr()); + else if (rcvr() != null) + jobPda = new DataStreamerPda(rcvr()); initPda = false; } @@ -1851,7 +1853,7 @@ else if (!topFut.isDone()) false, skipStore, keepBinary, - rcvr), + rcvr()), plc); locFuts.add(callFut); @@ -1986,13 +1988,13 @@ private void submit( if (topVer == null) topVer = ctx.cache().context().exchange().readyAffinityVersion(); - StreamReceiver rcvr0 = rcvr; + StreamReceiverMessage rcvrMsg0 = rcvrMsg; DataStreamerRequest req = new DataStreamerRequest( reqId, topicId, cacheName, - rcvr0, + rcvrMsg0, entries, true, skipStore, @@ -2004,22 +2006,11 @@ private void submit( dep != null ? dep.classLoaderId() : null, dep == null, topVer, - (rcvr0 == ISOLATED_UPDATER) ? partId : NO_STRIPE); - - // Every batch carries the same receiver, so it is marshalled once: the message is handed the bytes of - // an earlier request, and yields its own when it is the first to be marshalled. - T2, byte[]> marshalled = marshalledRcvr; - - byte[] rcvrBytes = marshalled != null && marshalled.get1() == rcvr0 ? marshalled.get2() : null; - - req.updaterBytes(rcvrBytes); + (rcvrMsg0.receiver() == ISOLATED_UPDATER) ? partId : NO_STRIPE); try { ctx.io().sendToGridTopic(node, TOPIC_DATASTREAM, req, plc); - if (rcvrBytes == null) - marshalledRcvr = new T2<>(rcvr0, req.updaterBytes()); - if (log.isDebugEnabled()) log.debug("Sent request to node [nodeId=" + node.id() + ", req=" + req + ']'); } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java index 6538752a23481..abea26da3d1b0 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java @@ -22,10 +22,8 @@ import java.util.UUID; import org.apache.ignite.configuration.DeploymentMode; import org.apache.ignite.internal.DeferredUnmarshalMessage; -import org.apache.ignite.internal.Marshalled; import org.apache.ignite.internal.Order; import org.apache.ignite.internal.StripedMessage; -import org.apache.ignite.internal.UseBinaryMarshaller; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; import org.apache.ignite.internal.processors.cache.GridCacheUtils; import org.apache.ignite.internal.util.tostring.GridToStringExclude; @@ -39,8 +37,7 @@ import static org.apache.ignite.internal.GridTopic.TOPIC_DATASTREAM; -/** Batch of streamed entries. The receiver is a user class. */ -@UseBinaryMarshaller +/** Batch of streamed entries. The receiver it carries is a user class, hence the deferred unmarshalling. */ public class DataStreamerRequest implements DeferredUnmarshalMessage, CacheIdAware, StripedMessage { /** */ @Order(0) @@ -56,12 +53,8 @@ public class DataStreamerRequest implements DeferredUnmarshalMessage, CacheIdAwa /** Cache receiver. A user object, kept out of the message {@code toString()}. */ @GridToStringExclude - @Marshalled("updaterBytes") - StreamReceiver updater; - - /** Serialized cache receiver. */ @Order(3) - byte[] updaterBytes; + StreamReceiverMessage updater; /** Entries to update. */ @Order(4) @@ -139,7 +132,7 @@ public DataStreamerRequest( long reqId, IgniteUuid resTopicId, @Nullable String cacheName, - StreamReceiver updater, + StreamReceiverMessage updater, Collection entries, boolean ignoreDepOwnership, boolean skipStore, @@ -190,23 +183,7 @@ String cacheName() { /** @return Updater. */ StreamReceiver updater() { - return updater; - } - - /** @return Serialized updater, {@code null} until the message is marshalled. */ - byte[] updaterBytes() { - return updaterBytes; - } - - /** - * Hands the message the serialized form of its updater, so the marshaller keeps it instead of producing its own. - * The receiver of a streamer does not change between batches, hence the sender marshals it once and passes the - * result on. - * - * @param updaterBytes Serialized updater taken from an already marshalled request. - */ - void updaterBytes(byte[] updaterBytes) { - this.updaterBytes = updaterBytes; + return updater != null ? updater.receiver() : null; } /** @return Entries to update. */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/StreamReceiverMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/StreamReceiverMessage.java new file mode 100644 index 0000000000000..08423747a3da2 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/StreamReceiverMessage.java @@ -0,0 +1,55 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.datastreamer; + +import org.apache.ignite.internal.Marshalled; +import org.apache.ignite.internal.Order; +import org.apache.ignite.internal.UseBinaryMarshaller; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.stream.StreamReceiver; + +/** + * The receiver of a streamer on its way to the nodes that own the data: a user object here, its serialized form on + * the wire. One instance serves every batch of a streamer, so the receiver is marshalled once and the batches share + * the result; a streamer given another receiver builds another instance. + */ +@UseBinaryMarshaller +public class StreamReceiverMessage implements Message { + /** */ + @Marshalled("rcvrBytes") + StreamReceiver rcvr; + + /** Serialized {@link #rcvr}, written by whichever batch is marshalled first and read by the rest. */ + @Order(0) + volatile byte[] rcvrBytes; + + /** Empty constructor. */ + public StreamReceiverMessage() { + // No-op. + } + + /** @param rcvr Receiver. */ + StreamReceiverMessage(StreamReceiver rcvr) { + this.rcvr = rcvr; + } + + /** @return Receiver. */ + StreamReceiver receiver() { + return rcvr; + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java index d68b9150ec8ff..86453aa9e9ccb 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java @@ -89,8 +89,8 @@ public class DataStreamerImplSelfTest extends GridCommonAbstractTest { /** Indicates whether we need to make the topology stale */ private static boolean needStaleTop = false; - /** Collects the serialized receiver of every streamer request sent, when set. */ - private static volatile Collection sentUpdaterBytes; + /** Collects the receiver carrier of every streamer request sent, when set. */ + private static volatile Collection sentReceivers; /** {@inheritDoc} */ @Override protected void afterTest() throws Exception { @@ -158,9 +158,9 @@ public void testReceiverMarshalledOncePerStreamer() throws Exception { startGrids(2); - Collection sent = new ConcurrentLinkedQueue<>(); + Collection sent = new ConcurrentLinkedQueue<>(); - sentUpdaterBytes = sent; + sentReceivers = sent; try (IgniteDataStreamer ldr = grid(0).dataStreamer(DEFAULT_CACHE_NAME)) { ldr.perNodeBufferSize(1); @@ -169,17 +169,17 @@ public void testReceiverMarshalledOncePerStreamer() throws Exception { ldr.addData(i, i); } finally { - sentUpdaterBytes = null; + sentReceivers = null; } assertTrue("Expected more than one request to a remote node, got " + sent.size(), sent.size() > 1); - byte[] first = F.first(sent); + StreamReceiverMessage first = F.first(sent); - assertNotNull(first); + assertNotNull(first.rcvrBytes); - for (byte[] bytes : sent) - assertTrue("The receiver was marshalled more than once", first == bytes); + for (StreamReceiverMessage rcvr : sent) + assertTrue("The receiver was marshalled more than once", first.rcvrBytes == rcvr.rcvrBytes); } /** @@ -710,12 +710,12 @@ private CacheConfiguration cacheConfiguration() { private static class StaleTopologyCommunicationSpi extends TcpCommunicationSpi { /** {@inheritDoc} */ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure ackC) { - Collection updaterBytes = sentUpdaterBytes; + Collection rcvrs = sentReceivers; // The message is already marshalled at this point, so the serialized receiver is in place. - if (updaterBytes != null && msg instanceof GridIoMessage + if (rcvrs != null && msg instanceof GridIoMessage && ((GridIoMessage)msg).message() instanceof DataStreamerRequest) - updaterBytes.add(((DataStreamerRequest)((GridIoMessage)msg).message()).updaterBytes()); + rcvrs.add(((DataStreamerRequest)((GridIoMessage)msg).message()).updater); // Send stale topology only in the first request to avoid indefinitely getting failures. if (needStaleTop) { @@ -739,7 +739,7 @@ private static class StaleTopologyCommunicationSpi extends TcpCommunicationSpi { req.requestId(), req.resTopicId, req.cacheName(), - req.updater(), + new StreamReceiverMessage(req.updater()), req.entries(), req.ignoreDeploymentOwnership(), req.skipStore(), From d3d15aeb023ad6067d8f75ec408d4082baab78f3 Mon Sep 17 00:00:00 2001 From: Anton Vinogradov Date: Sun, 9 Aug 2026 04:19:41 +0300 Subject: [PATCH 06/10] IGNITE-27977 Name the receiver accessors after the setter The streamer exposes receiver(StreamReceiver) yet read the field back through rcvr(), and the request named its carrier field after what the getter returns rather than after what it holds. Paired accessors in these classes share a name - allowOverwrite(), skipStore(), keepBinary() - so the getter is receiver() now, and the carrier is updaterMsg. Co-Authored-By: Claude Opus 5 --- .../processors/datastreamer/DataStreamerImpl.java | 14 +++++++------- .../datastreamer/DataStreamerRequest.java | 12 ++++++------ .../datastreamer/DataStreamerImplSelfTest.java | 2 +- 3 files changed, 14 insertions(+), 14 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java index 364d57c2f5f04..106357af751ca 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java @@ -491,13 +491,13 @@ public IgniteInternalFuture internalFuture() { /** @return Cache receiver. */ @SuppressWarnings("unchecked") - private StreamReceiver rcvr() { + private StreamReceiver receiver() { return (StreamReceiver)rcvrMsg.receiver(); } /** {@inheritDoc} */ @Override public boolean allowOverwrite() { - return rcvr() != ISOLATED_UPDATER; + return receiver() != ISOLATED_UPDATER; } /** {@inheritDoc} */ @@ -658,7 +658,7 @@ public IgniteFuture addDataInternal(Collection e lock(false); - if (rcvr() instanceof IsolatedUpdater && inconsistencyWarned.compareAndSet(false, true)) + if (receiver() instanceof IsolatedUpdater && inconsistencyWarned.compareAndSet(false, true)) log.warning(WRN_INCONSISTENT_UPDATES); try { @@ -892,9 +892,9 @@ private void load0( if (cacheObjCtx.addDeploymentInfo()) jobPda = new DataStreamerPda(key.value(cacheObjCtx, false), entry.getValue() != null ? entry.getValue().value(cacheObjCtx, false) : null, - rcvr()); - else if (rcvr() != null) - jobPda = new DataStreamerPda(rcvr()); + receiver()); + else if (receiver() != null) + jobPda = new DataStreamerPda(receiver()); initPda = false; } @@ -1853,7 +1853,7 @@ else if (!topFut.isDone()) false, skipStore, keepBinary, - rcvr()), + receiver()), plc); locFuts.add(callFut); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java index abea26da3d1b0..9de9ea1af43ea 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java @@ -51,10 +51,10 @@ public class DataStreamerRequest implements DeferredUnmarshalMessage, CacheIdAwa @Order(2) String cacheName; - /** Cache receiver. A user object, kept out of the message {@code toString()}. */ + /** Cache receiver, in the message that carries it. Holds a user object, hence out of {@code toString()}. */ @GridToStringExclude @Order(3) - StreamReceiverMessage updater; + StreamReceiverMessage updaterMsg; /** Entries to update. */ @Order(4) @@ -114,7 +114,7 @@ public DataStreamerRequest() { * @param reqId Request ID. * @param resTopicId Response topic ID. * @param cacheName Cache name. - * @param updater Cache receiver. + * @param updaterMsg Cache receiver, in the message that carries it. * @param entries Entries to put. * @param ignoreDepOwnership Ignore ownership. * @param skipStore Skip store flag. @@ -132,7 +132,7 @@ public DataStreamerRequest( long reqId, IgniteUuid resTopicId, @Nullable String cacheName, - StreamReceiverMessage updater, + StreamReceiverMessage updaterMsg, Collection entries, boolean ignoreDepOwnership, boolean skipStore, @@ -151,7 +151,7 @@ public DataStreamerRequest( this.reqId = reqId; this.resTopicId = resTopicId; this.cacheName = cacheName; - this.updater = updater; + this.updaterMsg = updaterMsg; this.entries = entries; this.ignoreDepOwnership = ignoreDepOwnership; this.skipStore = skipStore; @@ -183,7 +183,7 @@ String cacheName() { /** @return Updater. */ StreamReceiver updater() { - return updater != null ? updater.receiver() : null; + return updaterMsg != null ? updaterMsg.receiver() : null; } /** @return Entries to update. */ diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java index 86453aa9e9ccb..7cde572f5bf87 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java @@ -715,7 +715,7 @@ private static class StaleTopologyCommunicationSpi extends TcpCommunicationSpi { // The message is already marshalled at this point, so the serialized receiver is in place. if (rcvrs != null && msg instanceof GridIoMessage && ((GridIoMessage)msg).message() instanceof DataStreamerRequest) - rcvrs.add(((DataStreamerRequest)((GridIoMessage)msg).message()).updater); + rcvrs.add(((DataStreamerRequest)((GridIoMessage)msg).message()).updaterMsg); // Send stale topology only in the first request to avoid indefinitely getting failures. if (needStaleTop) { From 7b82b613132bfcf37a365e8da42ad622d6b31470 Mon Sep 17 00:00:00 2001 From: Anton Vinogradov Date: Sun, 9 Aug 2026 04:24:34 +0300 Subject: [PATCH 07/10] IGNITE-27977 Say why the receiver bytes are volatile The field is written by the batch that is marshalled first and read by the rest, and those batches leave on different threads. Without the keyword a reader could see the reference before the contents, skip the marshalling and send a half-written array. Co-Authored-By: Claude Opus 5 --- .../processors/datastreamer/StreamReceiverMessage.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/StreamReceiverMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/StreamReceiverMessage.java index 08423747a3da2..076bd8fe384e5 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/StreamReceiverMessage.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/StreamReceiverMessage.java @@ -34,7 +34,11 @@ public class StreamReceiverMessage implements Message { @Marshalled("rcvrBytes") StreamReceiver rcvr; - /** Serialized {@link #rcvr}, written by whichever batch is marshalled first and read by the rest. */ + /** + * Serialized {@link #rcvr}, written by whichever batch is marshalled first and read by the rest. Those batches + * leave on different threads, hence the {@code volatile}: a reader seeing the reference before the contents would + * skip the marshalling and send a half-written array. + */ @Order(0) volatile byte[] rcvrBytes; From 491bdcecf9489a85364aad19a735f9c96e9d88b6 Mon Sep 17 00:00:00 2001 From: Anton Vinogradov Date: Sun, 9 Aug 2026 04:43:29 +0300 Subject: [PATCH 08/10] IGNITE-27977 Drop the local copy the test did not need The collection was held in a local only because the test cleared the static field in a finally block, before the assertions ran. Clearing it in afterTest, next to the other static cleanup of this class, removes both the local and the try/finally. The local inside the SPI stays and is now explained: it reads the volatile field once, since the field is cleared while nodes that are still stopping keep sending through it. Co-Authored-By: Claude Opus 5 --- .../datastreamer/DataStreamerImplSelfTest.java | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java index 7cde572f5bf87..6b8b73ad82c07 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java @@ -98,6 +98,8 @@ public class DataStreamerImplSelfTest extends GridCommonAbstractTest { stopAllGrids(); + sentReceivers = null; + // Unbinds the log listeners from single static log instance. U.>field(DataStreamerImpl.class, "logRef").set(null); GridTestUtils.setFieldValue(null, DataStreamerImpl.class, "log", null); @@ -158,9 +160,7 @@ public void testReceiverMarshalledOncePerStreamer() throws Exception { startGrids(2); - Collection sent = new ConcurrentLinkedQueue<>(); - - sentReceivers = sent; + sentReceivers = new ConcurrentLinkedQueue<>(); try (IgniteDataStreamer ldr = grid(0).dataStreamer(DEFAULT_CACHE_NAME)) { ldr.perNodeBufferSize(1); @@ -168,17 +168,15 @@ public void testReceiverMarshalledOncePerStreamer() throws Exception { for (int i = 0; i < KEYS_COUNT; i++) ldr.addData(i, i); } - finally { - sentReceivers = null; - } - assertTrue("Expected more than one request to a remote node, got " + sent.size(), sent.size() > 1); + assertTrue("Expected more than one request to a remote node, got " + sentReceivers.size(), + sentReceivers.size() > 1); - StreamReceiverMessage first = F.first(sent); + StreamReceiverMessage first = F.first(sentReceivers); assertNotNull(first.rcvrBytes); - for (StreamReceiverMessage rcvr : sent) + for (StreamReceiverMessage rcvr : sentReceivers) assertTrue("The receiver was marshalled more than once", first.rcvrBytes == rcvr.rcvrBytes); } @@ -710,6 +708,7 @@ private CacheConfiguration cacheConfiguration() { private static class StaleTopologyCommunicationSpi extends TcpCommunicationSpi { /** {@inheritDoc} */ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure ackC) { + // Read once: the field is cleared after the test, while nodes still stopping send through this SPI. Collection rcvrs = sentReceivers; // The message is already marshalled at this point, so the serialized receiver is in place. From 0b9c412f5ba349c12d95d1cf25134bf68e679d21 Mon Sep 17 00:00:00 2001 From: Anton Vinogradov Date: Sun, 9 Aug 2026 04:46:55 +0300 Subject: [PATCH 09/10] IGNITE-27977 Clear the collected receivers before the test, not after Clearing after the test made the field nullable, which cost a null check and a local copy in the SPI, and my comment there claimed a race that did not exist - afterTest clears the field once the grids are already stopped. A final collection cleared in beforeTest gives each test the same clean start with none of that. Co-Authored-By: Claude Opus 5 --- .../DataStreamerImplSelfTest.java | 23 +++++++++---------- 1 file changed, 11 insertions(+), 12 deletions(-) diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java index 6b8b73ad82c07..8f2a737b835c7 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java @@ -89,8 +89,15 @@ public class DataStreamerImplSelfTest extends GridCommonAbstractTest { /** Indicates whether we need to make the topology stale */ private static boolean needStaleTop = false; - /** Collects the receiver carrier of every streamer request sent, when set. */ - private static volatile Collection sentReceivers; + /** Receiver carriers of the streamer requests sent since the current test started. */ + private static final Collection sentReceivers = new ConcurrentLinkedQueue<>(); + + /** {@inheritDoc} */ + @Override protected void beforeTest() throws Exception { + super.beforeTest(); + + sentReceivers.clear(); + } /** {@inheritDoc} */ @Override protected void afterTest() throws Exception { @@ -98,8 +105,6 @@ public class DataStreamerImplSelfTest extends GridCommonAbstractTest { stopAllGrids(); - sentReceivers = null; - // Unbinds the log listeners from single static log instance. U.>field(DataStreamerImpl.class, "logRef").set(null); GridTestUtils.setFieldValue(null, DataStreamerImpl.class, "log", null); @@ -160,8 +165,6 @@ public void testReceiverMarshalledOncePerStreamer() throws Exception { startGrids(2); - sentReceivers = new ConcurrentLinkedQueue<>(); - try (IgniteDataStreamer ldr = grid(0).dataStreamer(DEFAULT_CACHE_NAME)) { ldr.perNodeBufferSize(1); @@ -708,13 +711,9 @@ private CacheConfiguration cacheConfiguration() { private static class StaleTopologyCommunicationSpi extends TcpCommunicationSpi { /** {@inheritDoc} */ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure ackC) { - // Read once: the field is cleared after the test, while nodes still stopping send through this SPI. - Collection rcvrs = sentReceivers; - // The message is already marshalled at this point, so the serialized receiver is in place. - if (rcvrs != null && msg instanceof GridIoMessage - && ((GridIoMessage)msg).message() instanceof DataStreamerRequest) - rcvrs.add(((DataStreamerRequest)((GridIoMessage)msg).message()).updaterMsg); + if (msg instanceof GridIoMessage && ((GridIoMessage)msg).message() instanceof DataStreamerRequest) + sentReceivers.add(((DataStreamerRequest)((GridIoMessage)msg).message()).updaterMsg); // Send stale topology only in the first request to avoid indefinitely getting failures. if (needStaleTop) { From 36c67f23ff823ea40870d3dd84fb7cd92a62334a Mon Sep 17 00:00:00 2001 From: Anton Vinogradov Date: Sun, 9 Aug 2026 05:04:20 +0300 Subject: [PATCH 10/10] IGNITE-27977 Self-review touch-ups Read the receiver once where the deployment aware is built, instead of calling the getter three times in a row; keep the explicit type argument on individual() that the rewrite had dropped; unwrap the sent message once in the test SPI; and say in the request javadoc that the excluded field carries a user object rather than being one. Co-Authored-By: Claude Opus 5 --- .../processors/datastreamer/DataStreamerImpl.java | 10 ++++++---- .../processors/datastreamer/DataStreamerRequest.java | 2 +- .../datastreamer/DataStreamerImplSelfTest.java | 6 ++++-- 3 files changed, 11 insertions(+), 7 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java index 106357af751ca..a8bc307096e4b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java @@ -510,7 +510,7 @@ private StreamReceiver receiver() { if (node == null) throw new CacheException("Failed to get node for cache: " + cacheName); - rcvrMsg = new StreamReceiverMessage(allow ? DataStreamerCacheUpdaters.individual() : ISOLATED_UPDATER); + rcvrMsg = new StreamReceiverMessage(allow ? DataStreamerCacheUpdaters.individual() : ISOLATED_UPDATER); } /** {@inheritDoc} */ @@ -889,12 +889,14 @@ private void load0( assert key != null; if (initPda) { + StreamReceiver rcvr = receiver(); + if (cacheObjCtx.addDeploymentInfo()) jobPda = new DataStreamerPda(key.value(cacheObjCtx, false), entry.getValue() != null ? entry.getValue().value(cacheObjCtx, false) : null, - receiver()); - else if (receiver() != null) - jobPda = new DataStreamerPda(receiver()); + rcvr); + else if (rcvr != null) + jobPda = new DataStreamerPda(rcvr); initPda = false; } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java index 9de9ea1af43ea..f846c465599a3 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java @@ -51,7 +51,7 @@ public class DataStreamerRequest implements DeferredUnmarshalMessage, CacheIdAwa @Order(2) String cacheName; - /** Cache receiver, in the message that carries it. Holds a user object, hence out of {@code toString()}. */ + /** Cache receiver, in the message that carries it. Out of {@code toString()}: it is a user object. */ @GridToStringExclude @Order(3) StreamReceiverMessage updaterMsg; diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java index 8f2a737b835c7..5e350aafc1f95 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java @@ -711,9 +711,11 @@ private CacheConfiguration cacheConfiguration() { private static class StaleTopologyCommunicationSpi extends TcpCommunicationSpi { /** {@inheritDoc} */ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure ackC) { + Message sentMsg = msg instanceof GridIoMessage ? ((GridIoMessage)msg).message() : null; + // The message is already marshalled at this point, so the serialized receiver is in place. - if (msg instanceof GridIoMessage && ((GridIoMessage)msg).message() instanceof DataStreamerRequest) - sentReceivers.add(((DataStreamerRequest)((GridIoMessage)msg).message()).updaterMsg); + if (sentMsg instanceof DataStreamerRequest) + sentReceivers.add(((DataStreamerRequest)sentMsg).updaterMsg); // Send stale topology only in the first request to avoid indefinitely getting failures. if (needStaleTop) {