Vladsz83 commented on code in PR #13454:
URL: https://github.com/apache/ignite/pull/13454#discussion_r3751655506
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java:
##########
@@ -142,6 +147,89 @@ public void testCloseWithCancellation() throws Exception {
assertTrue(fut.isDone());
}
+ /**
+ * Streams {@link #KEYS_COUNT} entries from the first node, one entry per
request, and collects the receiver
+ * carrier of every request that leaves the node.
+ *
+ * @param rcvr Receiver to set, or {@code null} to keep the default one.
+ * @throws Exception If failed.
+ */
+ private void streamToRemoteNode(@Nullable StreamReceiver<Object, Object>
rcvr) throws Exception {
Review Comment:
Let's rename to smth like `startGridsAndStream`
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java:
##########
@@ -142,6 +147,89 @@ public void testCloseWithCancellation() throws Exception {
assertTrue(fut.isDone());
}
+ /**
+ * Streams {@link #KEYS_COUNT} entries from the first node, one entry per
request, and collects the receiver
+ * carrier of every request that leaves the node.
+ *
+ * @param rcvr Receiver to set, or {@code null} to keep the default one.
+ * @throws Exception If failed.
+ */
+ private void streamToRemoteNode(@Nullable StreamReceiver<Object, Object>
rcvr) throws Exception {
Review Comment:
let's put it belowe the test where it is used.
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java:
##########
@@ -88,6 +90,9 @@ public class DataStreamerImplSelfTest extends
GridCommonAbstractTest {
/** Indicates whether we need to make the topology stale */
private static boolean needStaleTop = false;
+ /** Receiver carriers of the streamer requests sent since the current test
started. */
+ private static final List<DataStreamerReceiverMessage> sentReceivers =
Collections.synchronizedList(new ArrayList<>());
Review Comment:
upper case fot static final. But let's use a test-instance variable.
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java:
##########
@@ -142,6 +147,89 @@ public void testCloseWithCancellation() throws Exception {
assertTrue(fut.isDone());
}
+ /**
+ * Streams {@link #KEYS_COUNT} entries from the first node, one entry per
request, and collects the receiver
+ * carrier of every request that leaves the node.
+ *
+ * @param rcvr Receiver to set, or {@code null} to keep the default one.
+ * @throws Exception If failed.
+ */
+ private void streamToRemoteNode(@Nullable StreamReceiver<Object, Object>
rcvr) throws Exception {
+ cnt = 0;
+
+ startGrids(2);
+
+ awaitPartitionMapExchange();
Review Comment:
Do we need to wait for PME here?
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java:
##########
@@ -88,6 +90,9 @@ public class DataStreamerImplSelfTest extends
GridCommonAbstractTest {
/** Indicates whether we need to make the topology stale */
private static boolean needStaleTop = false;
+ /** Receiver carriers of the streamer requests sent since the current test
started. */
+ private static final List<DataStreamerReceiverMessage> sentReceivers =
Collections.synchronizedList(new ArrayList<>());
Review Comment:
Can we hold only different receivers and use them in the following tests?
Like Set with a dedicated comparator. We could compare
`DataStreamerReceiverMessage#rcvrBytes` and
`DataStreamerReceiverMessage#builtIn`
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java:
##########
@@ -142,6 +147,89 @@ public void testCloseWithCancellation() throws Exception {
assertTrue(fut.isDone());
}
+ /**
+ * Streams {@link #KEYS_COUNT} entries from the first node, one entry per
request, and collects the receiver
+ * carrier of every request that leaves the node.
+ *
+ * @param rcvr Receiver to set, or {@code null} to keep the default one.
+ * @throws Exception If failed.
+ */
+ private void streamToRemoteNode(@Nullable StreamReceiver<Object, Object>
rcvr) throws Exception {
+ cnt = 0;
+
+ startGrids(2);
+
+ awaitPartitionMapExchange();
+
+ sentReceivers.clear();
+
+ try (IgniteDataStreamer<Object, Object> ldr =
grid(0).dataStreamer(DEFAULT_CACHE_NAME)) {
+ if (rcvr != null)
+ ldr.receiver(rcvr);
+
+ ldr.perNodeBufferSize(1);
+
+ for (int i = 0; i < KEYS_COUNT; i++)
+ ldr.addData(i, i);
+ }
+ }
+
+ /**
+ * 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 {
+ streamToRemoteNode(new TestReceiver());
+
+ assertTrue("Expected more than one request to a remote node, got " +
sentReceivers.size(),
+ sentReceivers.size() > 1);
+
+ DataStreamerReceiverMessage first = F.first(sentReceivers);
+
+ assertNotNull(first.rcvrBytes);
+
+ for (DataStreamerReceiverMessage rcvr : sentReceivers)
+ assertTrue("The receiver was marshalled more than once",
first.rcvrBytes == rcvr.rcvrBytes);
+ }
+
+ /**
+ * The updaters the streamer ships with are named rather than sent, and
the data still lands.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testBuiltInUpdaterIsNotSent() throws Exception {
Review Comment:
Collades with name `testBuiltInReceiverIsNotSent`. What differs?
`Receiver/Updater`?
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java:
##########
@@ -142,6 +147,89 @@ public void testCloseWithCancellation() throws Exception {
assertTrue(fut.isDone());
}
+ /**
+ * Streams {@link #KEYS_COUNT} entries from the first node, one entry per
request, and collects the receiver
+ * carrier of every request that leaves the node.
+ *
+ * @param rcvr Receiver to set, or {@code null} to keep the default one.
+ * @throws Exception If failed.
+ */
+ private void streamToRemoteNode(@Nullable StreamReceiver<Object, Object>
rcvr) throws Exception {
+ cnt = 0;
+
+ startGrids(2);
+
+ awaitPartitionMapExchange();
+
+ sentReceivers.clear();
+
+ try (IgniteDataStreamer<Object, Object> ldr =
grid(0).dataStreamer(DEFAULT_CACHE_NAME)) {
+ if (rcvr != null)
+ ldr.receiver(rcvr);
+
+ ldr.perNodeBufferSize(1);
+
+ for (int i = 0; i < KEYS_COUNT; i++)
+ ldr.addData(i, i);
+ }
+ }
+
+ /**
+ * 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 {
+ streamToRemoteNode(new TestReceiver());
+
+ assertTrue("Expected more than one request to a remote node, got " +
sentReceivers.size(),
+ sentReceivers.size() > 1);
+
+ DataStreamerReceiverMessage first = F.first(sentReceivers);
+
+ assertNotNull(first.rcvrBytes);
+
+ for (DataStreamerReceiverMessage rcvr : sentReceivers)
+ assertTrue("The receiver was marshalled more than once",
first.rcvrBytes == rcvr.rcvrBytes);
+ }
+
+ /**
+ * The updaters the streamer ships with are named rather than sent, and
the data still lands.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testBuiltInUpdaterIsNotSent() throws Exception {
+ streamToRemoteNode(null);
+
+ assertTrue("Expected requests to a remote node, got " +
sentReceivers.size(), !sentReceivers.isEmpty());
+
+ for (DataStreamerReceiverMessage rcvr : sentReceivers)
+ assertFalse("A built-in updater was sent with a request",
rcvr.user());
+
+ IgniteCache<Object, Object> cache = grid(1).cache(DEFAULT_CACHE_NAME);
+
+ for (int i = 0; i < KEYS_COUNT; i++)
+ assertEquals(i, cache.get(i));
+ }
+
+ /**
+ * A built-in receiver set explicitly is named rather than sent, just like
the default one.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testBuiltInReceiverIsNotSent() throws Exception {
Review Comment:
Lets use `testBuiltInUpdaterIsNotSent` and an internal cycle for
`DataStreamerBuiltinUpdater values`.
--
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]
