Vladsz83 commented on code in PR #13454: URL: https://github.com/apache/ignite/pull/13454#discussion_r3751956004
########## modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerReceiverMessage.java: ########## @@ -0,0 +1,68 @@ +/* + * 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; +import org.jetbrains.annotations.Nullable; + +/** DataStreamer cache receiver/updater message. */ +@UseBinaryMarshaller +public class DataStreamerReceiverMessage implements Message { + /** Custom cache receiver/updater; {@code null} when {@link #builtIn} is effective. */ + @Marshalled("rcvrBytes") + @Nullable StreamReceiver<?, ?> rcvr; + + /** Serialized {@link #rcvr}. */ + @Order(0) + volatile @Nullable byte[] rcvrBytes; Review Comment: Is `volatile` is needed only here? ########## modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerReceiverMessage.java: ########## @@ -0,0 +1,68 @@ +/* + * 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; +import org.jetbrains.annotations.Nullable; + +/** DataStreamer cache receiver/updater message. */ +@UseBinaryMarshaller +public class DataStreamerReceiverMessage implements Message { + /** Custom cache receiver/updater; {@code null} when {@link #builtIn} is effective. */ + @Marshalled("rcvrBytes") + @Nullable StreamReceiver<?, ?> rcvr; + + /** Serialized {@link #rcvr}. */ Review Comment: + `{@code null} when {@link #builtIn} is effective.` ########## modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java: ########## @@ -88,6 +91,9 @@ public class DataStreamerImplSelfTest extends GridCommonAbstractTest { /** Indicates whether we need to make the topology stale */ private static boolean needStaleTop = false; + /** Distinct updaters sent since the current test started: the serialized bytes, or the built-in constant. */ + private final Set<Object> sentUpdaters = Collections.synchronizedSet(new HashSet<>()); Review Comment: Do we need a comparator here? ########## modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java: ########## @@ -489,12 +486,29 @@ public IgniteInternalFuture<?> internalFuture() { @Override public void receiver(StreamReceiver<K, V> rcvr) { A.notNull(rcvr, "rcvr"); - this.rcvr = rcvr; + DataStreamerBuiltInUpdater builtIn = DataStreamerBuiltInUpdater.of(rcvr); + + rcvrMsg = builtIn != null ? builtIn.message() : new DataStreamerReceiverMessage(rcvr); + } + + /** @return Message of the receiver in use, a new one naming the Isolated updater until a receiver is set. */ Review Comment: Let's use `built-in`/`custom everywhere instead of `named`, `carried`. ########## modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerReceiverMessage.java: ########## @@ -0,0 +1,68 @@ +/* + * 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; +import org.jetbrains.annotations.Nullable; + +/** DataStreamer cache receiver/updater message. */ +@UseBinaryMarshaller +public class DataStreamerReceiverMessage implements Message { + /** Custom cache receiver/updater; {@code null} when {@link #builtIn} is effective. */ + @Marshalled("rcvrBytes") + @Nullable StreamReceiver<?, ?> rcvr; + + /** Serialized {@link #rcvr}. */ + @Order(0) + volatile @Nullable byte[] rcvrBytes; + + /** A built-in updater every node has; {@code null} when {@link #rcvr} is effective. */ + @Order(1) + @Nullable DataStreamerBuiltInUpdater builtIn; + + /** Empty constructor for serialization purposes. */ + public DataStreamerReceiverMessage() { + // No-op. + } + + /** @param rcvr Custom receiver. */ + DataStreamerReceiverMessage(StreamReceiver<?, ?> rcvr) { + assert DataStreamerBuiltInUpdater.of(rcvr) == null : "A built-in updater travels by name: " + rcvr; + + this.rcvr = rcvr; + } + + /** @param builtIn Built-in updater every node has, named rather than sent. */ Review Comment: Same notations `named rather than sent` -> smth. like `cutom/built-in` ########## modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java: ########## @@ -146,17 +146,14 @@ public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K, V>, Delayed */ private final Map<Long, ThreadBuffer> threadBufMap = new ConcurrentHashMap<>(); - /** Isolated receiver. */ - private static final StreamReceiver ISOLATED_UPDATER = new IsolatedUpdater(); + /** Default, Isolated receiver. */ + static final StreamReceiver ISOLATED_UPDATER = new IsolatedUpdater(); /** 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<K, V> rcvr = ISOLATED_UPDATER; - - /** */ - private byte[] updaterBytes; + /** Cache receiver in the message that carries it; {@code null} while none is set. */ + private volatile DataStreamerReceiverMessage rcvrMsg; Review Comment: `{@code null}`? Can it be null? ########## modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java: ########## @@ -142,6 +148,71 @@ 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 { + startGridsAndStream(new TestReceiver()); + + assertEquals("The receiver was marshalled more than once", 1, sentUpdaters.size()); + + assertTrue("The receiver was named instead of sent", F.first(sentUpdaters) instanceof byte[]); Review Comment: The same. Let's use `custom`/`builtin` instead of `named` everywhere. ########## modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java: ########## @@ -142,6 +148,71 @@ 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 { + startGridsAndStream(new TestReceiver()); + + assertEquals("The receiver was marshalled more than once", 1, sentUpdaters.size()); + + assertTrue("The receiver was named instead of sent", F.first(sentUpdaters) instanceof byte[]); + } + + /** + * Every built-in updater is named rather than sent, and the data still lands. + * + * @throws Exception If failed. + */ + @Test + public void testBuiltInUpdaterIsNotSent() throws Exception { + for (DataStreamerBuiltInUpdater builtIn : DataStreamerBuiltInUpdater.values()) { + startGridsAndStream(builtIn.updater()); + + assertEquals("Expected " + builtIn + " to be named, not sent", Collections.singleton(builtIn), + sentUpdaters); + + IgniteCache<Object, Object> cache = grid(1).cache(DEFAULT_CACHE_NAME); + + for (int i = 0; i < KEYS_COUNT; i++) + assertEquals(i, cache.get(i)); + + stopAllGrids(); + } + } + + /** + * Starts two nodes and streams {@link #KEYS_COUNT} entries from the first one, a request per entry, collecting + * the updaters they carry. Waits for the partition map first: until it is ready every partition is primary here, + * and a streamer that overwrites sends nothing to the remote node. + * + * @param rcvr Receiver to stream with. + * @throws Exception If failed. + */ + @SuppressWarnings("unchecked") + private void startGridsAndStream(StreamReceiver<?, ?> rcvr) throws Exception { + cnt = 0; + + startGrids(2); + + awaitPartitionMapExchange(); Review Comment: Do we need to wait for PME here? -- 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]
