anton-vinogradov commented on code in PR #13454:
URL: https://github.com/apache/ignite/pull/13454#discussion_r3751829121
##########
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:
Moved below both tests that use it.
##########
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:
Renamed to `startGridsAndStream` — it does start the grids, so the name
should say so.
##########
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:
Made it a test instance field, as you prefer, so the constant-case question
goes away. The SPI became an inner class to reach it.
##########
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:
Yes. Until the partition map is ready every partition is primary on the node
we stream from, so a streamer that overwrites sends nothing to the remote node
and the test has nothing to look at — that is exactly how it failed before the
wait was added. Said so in the helper javadoc.
##########
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:
Done, and it made the assertions shorter. The SPI now collects the distinct
updaters that left the node — the serialized bytes for a custom receiver, the
constant for a built-in one:
```java
sentUpdaters.add(updaterMsg.custom() ? updaterMsg.rcvrBytes :
updaterMsg.builtIn);
```
A `byte[]` compares by identity in a `HashSet`, so "marshalled once" is just
`assertEquals(1, sentUpdaters.size())`, and "named, not sent" is
`assertEquals(singleton(builtIn), sentUpdaters)`.
##########
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:
Gone — the two collapsed into one test.
##########
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:
Done:
```java
for (DataStreamerBuiltInUpdater builtIn :
DataStreamerBuiltInUpdater.values()) {
startGridsAndStream(builtIn.updater());
assertEquals("Expected " + builtIn + " to be named, not sent",
Collections.singleton(builtIn), sentUpdaters);
...
}
```
So a new built-in updater is covered the moment it is added to the enum.
--
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]
