PakhomovAlexander commented on code in PR #5136:
URL: https://github.com/apache/ignite-3/pull/5136#discussion_r1939332863
##########
modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItWorkerShutdownTest.java:
##########
@@ -266,6 +271,59 @@ void broadcastExecutionWorkerShutdown() {
AllInteractiveJobsApi.assertEachCalledOnce();
}
+ @Test
+ void partitionedBroadcastExecutionWorkerShutdown() {
+ // Prepare communication channels.
+ InteractiveJobs.initChannels(allNodeNames());
+
+ // Given table with replicas == 3 and partitions == 1.
+ createReplicatedTestTableWithOneRow();
+ // And partition leader for partition 1.
+ ClusterNode primaryReplica = getPrimaryReplica(node(0));
+ String firstWorkerName = primaryReplica.name();
+
+ // When start broadcast job on any node that is not primary replica.
+ Ignite entryNode = anyNodeExcept(primaryReplica);
+ CompletableFuture<BroadcastExecution<String>> executionFut =
compute(entryNode).submitAsync(
+ BroadcastJobTarget.table(TABLE_NAME),
+ InteractiveJobs.interactiveJobDescriptor(),
+ null
+ );
+
+ assertThat(executionFut, willCompleteSuccessfully());
+ BroadcastExecution<String> broadcastExecution = executionFut.join();
+ Collection<JobExecution<String>> executions =
broadcastExecution.executions();
+
+ // Then single job is alive.
+ assertThat(executions, hasSize(1));
+
+ JobExecution<String> execution =
executions.stream().findFirst().orElseThrow();
+
+ InteractiveJobs.byNode(primaryReplica).assertAlive();
+ TestingJobExecution<String> testingJobExecution = new
TestingJobExecution<>(execution);
+ testingJobExecution.assertExecuting();
+
+ // And it is running on primary replica node.
+ assertThat(execution.node().name(), equalTo(firstWorkerName));
+
+ // When stop worker node.
+ stopNode(primaryReplica);
+
+ // Get new primary replica
+ primaryReplica = getPrimaryReplica(entryNode);
+ String failoverNodeName = primaryReplica.name();
+ // Which is not the same node as before.
+ assertThat(failoverNodeName, not(equalTo(firstWorkerName)));
+
+ // And execution is running on the new primary replica. This will
implicitly wait for the job to actually run on the new node.
+ InteractiveJobs.byNode(primaryReplica).assertAlive();
+ testingJobExecution.assertExecuting();
+
+ // And the same execution object points to the new job
+ // TODO https://issues.apache.org/jira/browse/IGNITE-24353
+ // assertThat(execution.node().name(), equalTo(failoverNodeName));
Review Comment:
It would be useful to check after it if the job can finish and the result is
as expected.
--
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]
