| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.seatunnel.engine.e2e; |
| |
| import org.apache.seatunnel.common.utils.JsonUtils; |
| import org.apache.seatunnel.e2e.common.TestSuiteBase; |
| import org.apache.seatunnel.e2e.common.container.ContainerExtendedFactory; |
| import org.apache.seatunnel.e2e.common.container.EngineType; |
| import org.apache.seatunnel.e2e.common.container.flink.AbstractTestFlinkContainer; |
| import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer; |
| import org.apache.seatunnel.e2e.common.junit.TestContainerExtension; |
| |
| import org.apache.commons.collections4.CollectionUtils; |
| |
| import org.junit.jupiter.api.Assertions; |
| import org.junit.jupiter.api.TestTemplate; |
| import org.testcontainers.containers.Container; |
| |
| import lombok.extern.slf4j.Slf4j; |
| |
| import java.io.IOException; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicReference; |
| |
| import static org.awaitility.Awaitility.await; |
| |
| @Slf4j |
| @DisabledOnContainer( |
| value = {}, |
| type = {EngineType.SEATUNNEL, EngineType.SPARK}, |
| disabledReason = "only flink adjusts the parameter configuration rules") |
| public class UnifyEnvParameterIT extends TestSuiteBase { |
| |
| @TestContainerExtension |
| protected final ContainerExtendedFactory extendedFactory = |
| container -> { |
| Container.ExecResult extraCommands = |
| container.execInContainer( |
| "bash", |
| "-c", |
| "mkdir -p /tmp/seatunnel && chown -R flink /tmp/seatunnel"); |
| Assertions.assertEquals(0, extraCommands.getExitCode(), extraCommands.getStderr()); |
| }; |
| |
| @TestTemplate |
| public void testUnifiedParam(AbstractTestFlinkContainer container) |
| throws IOException, InterruptedException { |
| genericTest( |
| "/unify-env-param-test-resource/unify_env_param_fakesource_to_localfile.conf", |
| container); |
| } |
| |
| @TestTemplate |
| public void testOutdatedParam(AbstractTestFlinkContainer container) |
| throws IOException, InterruptedException { |
| genericTest( |
| "/unify-env-param-test-resource/outdated_env_param_fakesource_to_localfile.conf", |
| container); |
| } |
| |
| @TestTemplate |
| public void testUnifiedFlinkTableEnvParam(AbstractTestFlinkContainer container) { |
| CompletableFuture.supplyAsync( |
| () -> { |
| try { |
| return container.executeJob( |
| "/unify-env-param-test-resource/unify_flink_table_env_param_fakesource_to_console.conf"); |
| } catch (Exception e) { |
| log.error("Commit task exception :" + e.getMessage()); |
| throw new RuntimeException(e); |
| } |
| }); |
| // wait obtain job id |
| AtomicReference<String> jobId = new AtomicReference<>(); |
| await().atMost(300000, TimeUnit.MILLISECONDS) |
| .untilAsserted( |
| () -> { |
| Map<String, Object> jobInfo = |
| JsonUtils.toMap( |
| container.executeJobManagerInnerCommand( |
| "curl http://localhost:8081/jobs/overview"), |
| String.class, |
| Object.class); |
| List<Map<String, Object>> jobs = |
| (List<Map<String, Object>>) jobInfo.get("jobs"); |
| if (!CollectionUtils.isEmpty(jobs)) { |
| jobId.set(jobs.get(0).get("jid").toString()); |
| } |
| Assertions.assertNotNull(jobId.get()); |
| }); |
| |
| // obtain job info |
| AtomicReference<Map<String, Object>> jobInfoReference = new AtomicReference<>(); |
| await().atMost(60000, TimeUnit.MILLISECONDS) |
| .untilAsserted( |
| () -> { |
| Map<String, Object> jobInfo = |
| JsonUtils.toMap( |
| container.executeJobManagerInnerCommand( |
| String.format( |
| "curl http://localhost:8081/jobs/%s", |
| jobId.get())), |
| String.class, |
| Object.class); |
| // wait the job initialization is complete and enters the Running state |
| if (null != jobInfo && "RUNNING".equals(jobInfo.get("state"))) { |
| jobInfoReference.set(jobInfo); |
| } |
| Assertions.assertNotNull(jobInfoReference.get()); |
| }); |
| Map<String, Object> jobInfo = jobInfoReference.get(); |
| |
| /** |
| * 'table.exec.resource.default-parallelism' has a higher priority than 'parallelism', so |
| * one of these nodes must have a parallelism of 2. |
| */ |
| Map<String, Object> plan = (Map<String, Object>) jobInfo.get("plan"); |
| List<Map<String, Object>> nodes = (List<Map<String, Object>>) plan.get("nodes"); |
| boolean tableExecParallelism = false; |
| for (Map<String, Object> node : nodes) { |
| int parallelism = (int) node.get("parallelism"); |
| if (!tableExecParallelism && parallelism == 2) { |
| tableExecParallelism = true; |
| } |
| } |
| Assertions.assertTrue(tableExecParallelism); |
| } |
| |
| public void genericTest(String configPath, AbstractTestFlinkContainer container) |
| throws IOException, InterruptedException { |
| CompletableFuture.supplyAsync( |
| () -> { |
| try { |
| return container.executeJob(configPath); |
| } catch (Exception e) { |
| log.error("Commit task exception :" + e.getMessage()); |
| throw new RuntimeException(e); |
| } |
| }); |
| // wait obtain job id |
| AtomicReference<String> jobId = new AtomicReference<>(); |
| await().atMost(300000, TimeUnit.MILLISECONDS) |
| .untilAsserted( |
| () -> { |
| Map<String, Object> jobInfo = |
| JsonUtils.toMap( |
| container.executeJobManagerInnerCommand( |
| "curl http://localhost:8081/jobs/overview"), |
| String.class, |
| Object.class); |
| List<Map<String, Object>> jobs = |
| (List<Map<String, Object>>) jobInfo.get("jobs"); |
| if (!CollectionUtils.isEmpty(jobs)) { |
| jobId.set(jobs.get(0).get("jid").toString()); |
| } |
| Assertions.assertNotNull(jobId.get()); |
| }); |
| |
| // obtain job info |
| AtomicReference<Map<String, Object>> jobInfoReference = new AtomicReference<>(); |
| await().atMost(60000, TimeUnit.MILLISECONDS) |
| .untilAsserted( |
| () -> { |
| Map<String, Object> jobInfo = |
| JsonUtils.toMap( |
| container.executeJobManagerInnerCommand( |
| String.format( |
| "curl http://localhost:8081/jobs/%s", |
| jobId.get())), |
| String.class, |
| Object.class); |
| // wait the job initialization is complete and enters the Running state |
| if (null != jobInfo && "RUNNING".equals(jobInfo.get("state"))) { |
| jobInfoReference.set(jobInfo); |
| } |
| Assertions.assertNotNull(jobInfoReference.get()); |
| }); |
| Map<String, Object> jobInfo = jobInfoReference.get(); |
| |
| // obtain execution configuration |
| Map<String, Object> jobConfig = |
| JsonUtils.toMap( |
| container.executeJobManagerInnerCommand( |
| String.format( |
| "curl http://localhost:8081/jobs/%s/config", jobId.get())), |
| String.class, |
| Object.class); |
| Map<String, Object> executionConfig = |
| (Map<String, Object>) jobConfig.get("execution-config"); |
| |
| // obtain checkpoint configuration |
| Map<String, Object> checkpointConfig = |
| JsonUtils.toMap( |
| container.executeJobManagerInnerCommand( |
| String.format( |
| "curl http://localhost:8081/jobs/%s/checkpoints/config", |
| jobId.get())), |
| String.class, |
| Object.class); |
| |
| // obtain checkpoint storage |
| AtomicReference<Map<String, Object>> completedCheckpointReference = new AtomicReference<>(); |
| await().atMost(60000, TimeUnit.MILLISECONDS) |
| .untilAsserted( |
| () -> { |
| Map<String, Object> checkpointsInfo = |
| JsonUtils.toMap( |
| container.executeJobManagerInnerCommand( |
| String.format( |
| "curl http://localhost:8081/jobs/%s/checkpoints", |
| jobId.get())), |
| String.class, |
| Object.class); |
| Map<String, Object> latestCheckpoint = |
| (Map<String, Object>) checkpointsInfo.get("latest"); |
| // waiting for at least one checkpoint trigger |
| if (null != latestCheckpoint) { |
| completedCheckpointReference.set( |
| (Map<String, Object>) latestCheckpoint.get("completed")); |
| Assertions.assertNotNull(completedCheckpointReference.get()); |
| } |
| }); |
| /** |
| * adjust the configuration of this {@link |
| * org.apache.seatunnel.core.starter.flink.utils.ConfigKeyName} to use the 'flink.' and the |
| * flink parameter name, and check whether the configuration takes effect |
| */ |
| // PARALLELISM |
| int parallelism = (int) executionConfig.get("job-parallelism"); |
| Assertions.assertEquals(1, parallelism); |
| |
| // MAX_PARALLELISM |
| int maxParallelism = (int) jobInfo.get("maxParallelism"); |
| Assertions.assertEquals(5, maxParallelism); |
| |
| // CHECKPOINT_INTERVAL |
| int interval = (int) checkpointConfig.get("interval"); |
| Assertions.assertEquals(10000, interval); |
| |
| // CHECKPOINT_MODE |
| String mode = checkpointConfig.get("mode").toString(); |
| Assertions.assertEquals("exactly_once", mode); |
| |
| // CHECKPOINT_TIMEOUT |
| int checkpointTimeout = (int) checkpointConfig.get("timeout"); |
| Assertions.assertEquals(600000, checkpointTimeout); |
| |
| // CHECKPOINT_DATA_URI |
| String externalPath = completedCheckpointReference.get().get("external_path").toString(); |
| Assertions.assertTrue(externalPath.startsWith("file:/tmp/seatunnel/flink/checkpoints")); |
| |
| // MAX_CONCURRENT_CHECKPOINTS |
| int maxConcurrent = (int) checkpointConfig.get("max_concurrent"); |
| Assertions.assertEquals(2, maxConcurrent); |
| |
| // CHECKPOINT_CLEANUP_MODE |
| Map<String, Object> externalizationMap = |
| (Map<String, Object>) checkpointConfig.get("externalization"); |
| boolean externalization = (boolean) externalizationMap.get("delete_on_cancellation"); |
| Assertions.assertTrue(externalization); |
| |
| // MIN_PAUSE_BETWEEN_CHECKPOINTS |
| int minPause = (int) checkpointConfig.get("min_pause"); |
| Assertions.assertEquals(100, minPause); |
| |
| // FAIL_ON_CHECKPOINTING_ERRORS |
| int tolerableFailedCheckpoints = (int) checkpointConfig.get("tolerable_failed_checkpoints"); |
| Assertions.assertEquals(5, tolerableFailedCheckpoints); |
| |
| // RESTART_STRATEGY / because the restart strategy is fixed-delay in config file, so don't |
| // check failure-rate |
| String restartStrategy = executionConfig.get("restart-strategy").toString(); |
| Assertions.assertTrue(restartStrategy.contains("fixed delay")); |
| |
| // RESTART_ATTEMPTS |
| Assertions.assertTrue(restartStrategy.contains("2 restart attempts")); |
| |
| // RESTART_DELAY_BETWEEN_ATTEMPTS |
| Assertions.assertTrue(restartStrategy.contains("fixed delay (1000 ms)")); |
| |
| // STATE_BACKEND |
| String stateBackend = checkpointConfig.get("state_backend").toString(); |
| Assertions.assertTrue(stateBackend.contains("RocksDBStateBackend")); |
| } |
| } |
