Vladsz83 commented on code in PR #12488:
URL: https://github.com/apache/ignite/pull/12488#discussion_r2483371235
##########
modules/core/src/test/java/org/apache/ignite/internal/client/thin/ServiceAwarenessTest.java:
##########
@@ -612,6 +611,11 @@ private IgniteClient startClient(@Nullable
Collection<UUID> requestedServerNodes
getClientConfiguration(grid(0)));
}
+ /** */
+ private Set<UUID> nodeIds(int... nodeIdxs) {
Review Comment:
Might be static if we do not use something like `grid(idx)`.
Also, let's rename to `serviceNodeIds`
##########
modules/core/src/test/java/org/apache/ignite/internal/client/thin/ServiceAwarenessTest.java:
##########
@@ -350,94 +368,75 @@ public void testForcedServiceRedeployWhileClientIsIdle()
throws Exception {
}
/** */
- private void doTestClusterTopChangesWhileServiceCalling(
- int nodesCnt,
- boolean addNodes,
- boolean multiThreaded)
- throws Exception {
- assert nodesCnt > 0;
-
- Set<UUID> newNodesUUIDs = new GridConcurrentHashSet<>();
+ private void doTestClusterTopChangesWhileServiceCalling(boolean shrinkTop,
int svcInvokeThreads) throws Exception {
+ Set<UUID> expInitSvcTop = shrinkTop ? nodeIds(1, 2, 4, 5, 6) :
nodeIds(1, 2);
+ Set<UUID> expUpdSvcTop = shrinkTop ? nodeIds(1, 2) : nodeIds(1, 2, 4,
5, 6);
// Start additional nodes to stop them.
- if (!addNodes) {
- startGridsMultiThreaded(GRIDS, nodesCnt);
-
- for (int i = GRIDS; i < GRIDS + nodesCnt; ++i)
- newNodesUUIDs.add(grid(i).localNode().id());
- }
+ if (shrinkTop)
+ startGridsMultiThreaded(BASE_NODES_CNT, TOP_UPD_NODES_CNT);
// Service topology on the clients.
- Set<UUID> srvcTopOnClient = new GridConcurrentHashSet<>();
+ AtomicReference<Set<UUID>> actualSvcTop = new AtomicReference<>();
- addSrvcTopUpdateClientLogLsnr(srvcTopOnClient::addAll);
+ registerServiceTopologyUpdateListener(actualSvcTop::set);
- AtomicBoolean changeClusterTop = new AtomicBoolean();
- AtomicBoolean stopFlag = new AtomicBoolean();
+ AtomicBoolean isTestStopped = new AtomicBoolean();
Review Comment:
Trivial. `is` for `bool` of bool-return-methods isn't by the code style.
Let's use smth. more conventional like 'stopFlag'
##########
modules/core/src/test/java/org/apache/ignite/internal/client/thin/ServiceAwarenessTest.java:
##########
@@ -350,94 +368,75 @@ public void testForcedServiceRedeployWhileClientIsIdle()
throws Exception {
}
/** */
- private void doTestClusterTopChangesWhileServiceCalling(
- int nodesCnt,
- boolean addNodes,
- boolean multiThreaded)
- throws Exception {
- assert nodesCnt > 0;
-
- Set<UUID> newNodesUUIDs = new GridConcurrentHashSet<>();
+ private void doTestClusterTopChangesWhileServiceCalling(boolean shrinkTop,
int svcInvokeThreads) throws Exception {
+ Set<UUID> expInitSvcTop = shrinkTop ? nodeIds(1, 2, 4, 5, 6) :
nodeIds(1, 2);
+ Set<UUID> expUpdSvcTop = shrinkTop ? nodeIds(1, 2) : nodeIds(1, 2, 4,
5, 6);
// Start additional nodes to stop them.
- if (!addNodes) {
- startGridsMultiThreaded(GRIDS, nodesCnt);
-
- for (int i = GRIDS; i < GRIDS + nodesCnt; ++i)
- newNodesUUIDs.add(grid(i).localNode().id());
- }
+ if (shrinkTop)
+ startGridsMultiThreaded(BASE_NODES_CNT, TOP_UPD_NODES_CNT);
// Service topology on the clients.
- Set<UUID> srvcTopOnClient = new GridConcurrentHashSet<>();
+ AtomicReference<Set<UUID>> actualSvcTop = new AtomicReference<>();
Review Comment:
Minor. Suggestion: 'actual' is a bit confusing. It hints of real topology,
not detected by a service. let's use just 'scvTop'
##########
modules/core/src/test/java/org/apache/ignite/internal/client/thin/ServiceAwarenessTest.java:
##########
@@ -350,94 +368,75 @@ public void testForcedServiceRedeployWhileClientIsIdle()
throws Exception {
}
/** */
- private void doTestClusterTopChangesWhileServiceCalling(
- int nodesCnt,
- boolean addNodes,
- boolean multiThreaded)
- throws Exception {
- assert nodesCnt > 0;
-
- Set<UUID> newNodesUUIDs = new GridConcurrentHashSet<>();
+ private void doTestClusterTopChangesWhileServiceCalling(boolean shrinkTop,
int svcInvokeThreads) throws Exception {
+ Set<UUID> expInitSvcTop = shrinkTop ? nodeIds(1, 2, 4, 5, 6) :
nodeIds(1, 2);
+ Set<UUID> expUpdSvcTop = shrinkTop ? nodeIds(1, 2) : nodeIds(1, 2, 4,
5, 6);
// Start additional nodes to stop them.
- if (!addNodes) {
- startGridsMultiThreaded(GRIDS, nodesCnt);
-
- for (int i = GRIDS; i < GRIDS + nodesCnt; ++i)
- newNodesUUIDs.add(grid(i).localNode().id());
- }
+ if (shrinkTop)
+ startGridsMultiThreaded(BASE_NODES_CNT, TOP_UPD_NODES_CNT);
// Service topology on the clients.
- Set<UUID> srvcTopOnClient = new GridConcurrentHashSet<>();
+ AtomicReference<Set<UUID>> actualSvcTop = new AtomicReference<>();
- addSrvcTopUpdateClientLogLsnr(srvcTopOnClient::addAll);
+ registerServiceTopologyUpdateListener(actualSvcTop::set);
- AtomicBoolean changeClusterTop = new AtomicBoolean();
- AtomicBoolean stopFlag = new AtomicBoolean();
+ AtomicBoolean isTestStopped = new AtomicBoolean();
try (IgniteClient client = startClient()) {
ServicesTest.TestServiceInterface svc =
client.services().serviceProxy(SRV_NAME,
ServicesTest.TestServiceInterface.class);
((GridTestLog4jLogger)log).setLevel(Level.DEBUG);
- IgniteInternalFuture<?> runFut = runMultiThreadedAsync(() -> {
- do {
- try {
- svc.testMethod();
- }
- catch (ClientException e) {
- String m = e.getMessage();
-
- // TODO: IGNITE-20802 : Exception should not occur.
- // Client doesn't retry service invocation if the
redirected-to service instance node leaves cluster.
- if (addNodes || (!m.contains("Node has left grid") &&
!m.contains("Failed to send job due to node failure"))
- || newNodesUUIDs.stream().noneMatch(nid ->
m.contains(nid.toString())))
- throw e;
- }
- }
- while (!stopFlag.get());
- }, multiThreaded ? 4 : 1, "ServiceTestLoader");
-
- while (!stopFlag.get()) {
- // Wait until the initial topology is received.
- if (srvcTopOnClient.size() == (addNodes ? INIT_SRVC_NODES_CNT
: INIT_SRVC_NODES_CNT + nodesCnt)
- && changeClusterTop.compareAndSet(false, true)) {
- srvcTopOnClient.clear();
-
- for (int i = 0; i < nodesCnt; ++i) {
- int nodeIdx = GRIDS + i;
-
- runAsync(() -> {
- try {
- if (addNodes)
-
newNodesUUIDs.add(startGrid(nodeIdx).localNode().id());
- else
- stopGrid(nodeIdx);
- }
- catch (Exception e) {
- log.error("Unable to start or stop test
grid.", e);
-
- stopFlag.set(true);
- }
- });
+ IgniteInternalFuture<?> svcInvokeFut = runMultiThreadedAsync(
+ () -> {
+ do {
+ try {
+ svc.testMethod();
+ }
+ catch (Exception e) {
+ String errMsg = e.getMessage();
+
+ // TODO: IGNITE-20802 : Exception should not occur.
+ // Client doesn't retry service invocation if the
redirected-to service instance node leaves cluster.
+ assertTrue(shrinkTop
+ && (errMsg.contains("Node has left grid") ||
errMsg.contains("Failed to send job due to node failure"))
+ && expInitSvcTop.stream().anyMatch(id ->
errMsg.contains(id.toString())));
+ }
}
- }
+ while (!isTestStopped.get());
+ },
+ svcInvokeThreads,
+ "ServiceTestLoader"
+ );
+
+ assertTrue(waitForCondition(() ->
expInitSvcTop.equals(actualSvcTop.get()), getTestTimeout()));
- // Stop if new excepted service topology received.
- if (srvcTopOnClient.size() == (addNodes ? INIT_SRVC_NODES_CNT
+ nodesCnt : INIT_SRVC_NODES_CNT))
- stopFlag.set(true);
+ Collection<IgniteInternalFuture<?>> topUpdFuts =
ConcurrentHashMap.newKeySet();
- Thread.sleep(10);
+ for (int i = 0; i < TOP_UPD_NODES_CNT; ++i) {
+ int nodeIdx = BASE_NODES_CNT + i;
+
+ topUpdFuts.add(runAsync(() -> {
+ if (shrinkTop)
+ stopGrid(nodeIdx);
+ else
+ startGrid(nodeIdx);
+ }));
}
- runFut.get();
- }
+ assertTrue(waitForCondition(() ->
expUpdSvcTop.equals(actualSvcTop.get()), getTestTimeout()));
Review Comment:
Suggestion: let's wait for the futures first.
##########
modules/core/src/test/java/org/apache/ignite/internal/client/thin/ServiceAwarenessTest.java:
##########
@@ -90,11 +98,18 @@ public class ServiceAwarenessTest extends
AbstractThinClientTest {
/** */
private static ListeningTestLogger clientLogLsnr;
+ /** */
+ private final UUID[] nodeIds = new UUID[BASE_NODES_CNT +
TOP_UPD_NODES_CNT];
+
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+ int nodeIdx = getTestIgniteInstanceIndex(igniteInstanceName);
+
cfg.setDiscoverySpi(new TestBlockingDiscoverySpi());
+ cfg.setNodeId(nodeIds[nodeIdx]);
Review Comment:
Not sure that we should use deprecated method. Can we calculate and watch th
ids inside the test? Example in #12491
##########
modules/core/src/test/java/org/apache/ignite/internal/client/thin/ServiceAwarenessTest.java:
##########
@@ -350,94 +368,75 @@ public void testForcedServiceRedeployWhileClientIsIdle()
throws Exception {
}
/** */
- private void doTestClusterTopChangesWhileServiceCalling(
- int nodesCnt,
- boolean addNodes,
- boolean multiThreaded)
- throws Exception {
- assert nodesCnt > 0;
-
- Set<UUID> newNodesUUIDs = new GridConcurrentHashSet<>();
+ private void doTestClusterTopChangesWhileServiceCalling(boolean shrinkTop,
int svcInvokeThreads) throws Exception {
+ Set<UUID> expInitSvcTop = shrinkTop ? nodeIds(1, 2, 4, 5, 6) :
nodeIds(1, 2);
Review Comment:
Can we rely here on `BASE_NODES_CNT`, `TOP_UPD_NODES_CNT` and/or on
TestNodeFilter somehow?
Example in #12491
--
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]
