petrov-mg commented on code in PR #12488:
URL: https://github.com/apache/ignite/pull/12488#discussion_r2484883408
##########
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:
Done.
--
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]
