EricJoy2048 commented on code in PR #3503:
URL:
https://github.com/apache/incubator-seatunnel/pull/3503#discussion_r1033482470
##########
seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/ClusterFaultToleranceIT.java:
##########
@@ -611,4 +611,116 @@ public void testStreamJobRestoreIn3NodeMasterDown()
throws ExecutionException, I
}
}
}
+
+ @SuppressWarnings("checkstyle:RegexpSingleline")
+ @Test
+ public void testStreamJobRestoreInAllNodeDown() throws ExecutionException,
InterruptedException {
+ String testCaseName = "testStreamJobRestoreInAllNodeDown";
+ String testClusterName =
"ClusterFaultToleranceIT_testStreamJobRestoreInAllNodeDown";
+ long testRowNumber = 1000;
+ int testParallelism = 6;
+ HazelcastInstanceImpl node1 = null;
+ HazelcastInstanceImpl node2 = null;
+ HazelcastInstanceImpl node3 = null;
+ SeaTunnelClient engineClient = null;
+
+ try {
+ node1 = SeaTunnelServerStarter.createHazelcastInstance(
+ TestUtils.getClusterName(testClusterName));
+
+ node2 = SeaTunnelServerStarter.createHazelcastInstance(
+ TestUtils.getClusterName(testClusterName));
+
+ node3 = SeaTunnelServerStarter.createHazelcastInstance(
+ TestUtils.getClusterName(testClusterName));
+
+ // waiting all node added to cluster
+ HazelcastInstanceImpl finalNode = node1;
+ Awaitility.await().atMost(10000, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> Assertions.assertEquals(3,
finalNode.getCluster().getMembers().size()));
+
+ Common.setDeployMode(DeployMode.CLIENT);
+ ImmutablePair<String, String> testResources =
+ createTestResources(testCaseName, JobMode.STREAMING,
testRowNumber, testParallelism);
+ JobConfig jobConfig = new JobConfig();
+ jobConfig.setName(testCaseName);
+
+ ClientConfig clientConfig =
ConfigProvider.locateAndGetClientConfig();
+ clientConfig.setClusterName(
+ TestUtils.getClusterName(testClusterName));
+ engineClient = new SeaTunnelClient(clientConfig);
+ JobExecutionEnvironment jobExecutionEnv =
+
engineClient.createExecutionContext(testResources.getRight(), jobConfig);
+ ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
+
+ CompletableFuture<JobStatus> objectCompletableFuture =
CompletableFuture.supplyAsync(() -> {
+ return clientJobProxy.waitForJobComplete();
+ });
+
+ Awaitility.await().atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> {
+ // Wait some tasks commit finished, and we can get
rows from the sink target dir
+ Thread.sleep(2000);
+
System.out.println(FileUtils.getFileLineNumberFromDir(testResources.getLeft()));
+
Assertions.assertTrue(JobStatus.RUNNING.equals(clientJobProxy.getJobStatus()) &&
+
FileUtils.getFileLineNumberFromDir(testResources.getLeft()) > 1);
+ });
+
+ Thread.sleep(5000);
+ // shutdown all node
+ node1.shutdown();
+ node2.shutdown();
+ node3.shutdown();
+
+ Thread.sleep(10000);
+
+ node1 = SeaTunnelServerStarter.createHazelcastInstance(
+ TestUtils.getClusterName(testClusterName));
+
+ node2 = SeaTunnelServerStarter.createHazelcastInstance(
+ TestUtils.getClusterName(testClusterName));
+
+ node3 = SeaTunnelServerStarter.createHazelcastInstance(
+ TestUtils.getClusterName(testClusterName));
+
+ Thread.sleep(360000);
Review Comment:
> I have modified the code, but I cannot pass the CI. Is the code I modified
wrong?
Please wait https://github.com/apache/incubator-seatunnel/pull/3418 merge.
And then you need config to enable `IMap storage` for this test case.
--
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]
