rmetzger commented on code in PR #29379: URL: https://github.com/apache/flink/pull/29379#discussion_r4262679263
########## flink-container/src/test/java/org/apache/flink/container/entrypoint/MiniClusterJobEntryPointITCase.java: ########## @@ -0,0 +1,928 @@ +/* + * 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.flink.container.entrypoint; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.api.common.eventtime.WatermarkStrategy; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy; +import org.apache.flink.configuration.CheckpointingOptions; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.DeploymentOptions; +import org.apache.flink.configuration.GlobalConfiguration; +import org.apache.flink.configuration.HighAvailabilityOptions; +import org.apache.flink.configuration.JobManagerOptions; +import org.apache.flink.configuration.MemorySize; +import org.apache.flink.configuration.PipelineOptionsInternal; +import org.apache.flink.configuration.RestOptions; +import org.apache.flink.configuration.TaskManagerOptions; +import org.apache.flink.connector.datagen.source.DataGeneratorSource; +import org.apache.flink.core.testutils.AllCallbackWrapper; +import org.apache.flink.runtime.clusterframework.ApplicationStatus; +import org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils; +import org.apache.flink.runtime.highavailability.ApplicationResultStoreOptions; +import org.apache.flink.runtime.highavailability.JobResultStoreOptions; +import org.apache.flink.runtime.taskexecutor.TaskManagerRunner; +import org.apache.flink.runtime.util.TestingFatalErrorHandler; +import org.apache.flink.runtime.zookeeper.ZooKeeperExtension; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink; +import org.apache.flink.util.NetUtils; +import org.apache.flink.util.jackson.JacksonMapperFactory; + +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.IOException; +import java.io.InputStream; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.function.Predicate; +import java.util.jar.Attributes; +import java.util.jar.JarEntry; +import java.util.jar.JarOutputStream; +import java.util.jar.Manifest; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; + +/** + * Runs {@link MiniClusterJobEntryPoint}, configured the way that Flink Kubernetes Operator + * configures an application deployment: ZooKeeper HA, a fixed JobID in the config file, the cluster + * kept up after the application finishes, and result stores retained on commit. + */ +class MiniClusterJobEntryPointITCase { + + private static final Duration TIMEOUT = Duration.ofMinutes(2); + private static final Duration POLL_INTERVAL = Duration.ofMillis(200); + private static final Duration REQUEST_TIMEOUT = Duration.ofSeconds(5); + + /** How long a restarted process has to stay up to count as having survived the restart. */ + private static final Duration STAYS_UP_WINDOW = Duration.ofSeconds(5); + + @RegisterExtension + static final AllCallbackWrapper<ZooKeeperExtension> ZOOKEEPER = + new AllCallbackWrapper<>(new ZooKeeperExtension()); + + private static final ObjectMapper OBJECT_MAPPER = JacksonMapperFactory.createObjectMapper(); + private static final HttpClient HTTP_CLIENT = HttpClient.newHttpClient(); + + @TempDir private Path tempDir; + + private Path confDir; + private Path markerFile; + private JobID jobId; + private Path log4jConfig; + private int restPort; + private String haClusterId; + + @BeforeEach + void writeOperatorStyleConfiguration() throws Exception { + confDir = Files.createDirectories(tempDir.resolve("conf")); + markerFile = tempDir.resolve("main-invocations"); + jobId = JobID.generate(); + try (NetUtils.Port port = NetUtils.getAvailablePort()) { + restPort = port.getPort(); + } + + log4jConfig = tempDir.resolve("log4j2.properties"); + Files.write( + log4jConfig, + List.of( + "rootLogger.level = INFO", + "rootLogger.appenderRef.console.ref = ConsoleAppender", + "appender.console.name = ConsoleAppender", + "appender.console.type = CONSOLE", + "appender.console.layout.type = PatternLayout", + "appender.console.layout.pattern = %d{HH:mm:ss,SSS} %-5p %c{1} - %m%n")); + + haClusterId = "minicluster-itcase-" + UUID.randomUUID(); + final String haStorage = tempDir.resolve("ha").toUri().toString(); + Files.write( + confDir.resolve(GlobalConfiguration.FLINK_CONF_FILENAME), + List.of( + entry(HighAvailabilityOptions.HA_MODE, "zookeeper"), + entry( + HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM, + ZOOKEEPER.getCustomExtension().getConnectString()), + entry(HighAvailabilityOptions.HA_STORAGE_PATH, haStorage), + entry(HighAvailabilityOptions.HA_CLUSTER_ID, haClusterId), + entry(PipelineOptionsInternal.PIPELINE_FIXED_JOB_ID, jobId.toHexString()), + entry(DeploymentOptions.SHUTDOWN_ON_APPLICATION_FINISH, false), + entry(DeploymentOptions.SUBMIT_FAILED_JOB_ON_APPLICATION_ERROR, true), + entry(JobResultStoreOptions.DELETE_ON_COMMIT, false), + entry(ApplicationResultStoreOptions.DELETE_ON_COMMIT, false), + entry(RestOptions.ADDRESS, "localhost"), + entry(RestOptions.PORT, restPort), + entry(RestOptions.BIND_PORT, restPort))); + } + + @Test + void restartAfterApplicationFinishedDoesNotRerunMain() throws Exception { + final Process first = startMiniClusterProcess("first"); + try { + awaitJobState(first, "first", "FINISHED"); + assertThat(first.isAlive()) + .as("the process stays up after the application finishes") + .isTrue(); + } finally { + // SIGTERM is what Kubernetes sends when pods are deleted + stop(first); + } + assertThat(mainInvocations()).isEqualTo(1); + + final Process second = startMiniClusterProcess("second"); + try { + awaitClusterServing(second, "second"); + assertStaysUp(second, "second"); + assertThat(mainInvocations()) + .as("the restarted process must not run main() again") + .isEqualTo(1); + } finally { + stop(second); + } + } + + @Test + void applicationErrorIsReportedAsFailedJobWhileProcessStaysUp() throws Exception { + final List<String> arguments = applicationArguments(FailingJob.class, List.of()); + + final Process first = startProcess("first", MiniClusterJobEntryPoint.class, arguments); + try { + awaitJobState(first, "first", "FAILED"); + assertStaysUp(first, "first"); + } finally { + stop(first); + } + assertThat(mainInvocations()).isEqualTo(1); + + final Process second = startProcess("second", MiniClusterJobEntryPoint.class, arguments); + try { + awaitClusterServing(second, "second"); + assertStaysUp(second, "second"); + assertThat(mainInvocations()) + .as("the restarted process must not run main() again") + .isEqualTo(1); + } finally { + stop(second); + } + } + + @Test + void standaloneApplicationClusterRestartAfterApplicationFinishes() throws Exception { + final Configuration taskManagerMemory = new Configuration(); + taskManagerMemory.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1536m")); + + final String dynamicConfigs = + TaskExecutorProcessUtils.generateDynamicConfigsStr( + TaskExecutorProcessUtils.processSpecFromConfig(taskManagerMemory)); + + final List<String> taskManagerArguments = new ArrayList<>(); + Collections.addAll( + taskManagerArguments, + "--configDir", + confDir.toString(), + "-D" + TaskManagerOptions.HOST.key() + "=localhost"); + Collections.addAll(taskManagerArguments, dynamicConfigs.trim().split("\\s+")); + + final Process taskManager = + startProcess("taskmanager", TaskManagerRunner.class, taskManagerArguments); + try { + final Process first = startJobManagerProcess("first"); + try { + awaitJobState(first, "first", "FINISHED"); + assertThat(first.isAlive()) + .as("the process stays up after the application finishes") + .isTrue(); + } finally { + stop(first); + } + assertThat(mainInvocations()).isEqualTo(1); + + final Process second = startJobManagerProcess("second"); + try { + awaitClusterServing(second, "second"); + assertStaysUp(second, "second"); + assertThat(mainInvocations()) + .as("the restarted process must not run main() again") + .isEqualTo(1); + } finally { + stop(second); + } + } catch (AssertionError e) { + printLog("taskmanager"); + throw e; + } finally { + stop(taskManager); + } + } + + private Process startJobManagerProcess(String name) throws IOException { + final List<String> arguments = new ArrayList<>(applicationArguments()); + arguments.addAll(0, List.of("-D" + JobManagerOptions.ADDRESS.key() + "=localhost")); + return startProcess(name, StandaloneApplicationClusterEntryPoint.class, arguments); + } + + /** How the pod's previous process ended. */ + enum Termination { + /** The pod was deleted or rescheduled: the process got to shut down. */ + SIGTERM, + /** The process crashed, was OOM-killed, or its node was lost. */ + SIGKILL + } + + @ParameterizedTest + @EnumSource(Termination.class) + void restartResumesJobFromLatestCheckpoint(Termination termination) throws Exception { + final List<String> arguments = unboundedJobArguments(); + + final Process first = startProcess("first", MiniClusterJobEntryPoint.class, arguments); + final long lastCheckpointBeforeRestart; + try { + lastCheckpointBeforeRestart = + awaitRest( + first, + "first", + checkpointsPath(), + "three completed checkpoints", + json -> json.at("/latest/completed/id").asLong() >= 3) + .at("/latest/completed/id") + .asLong(); + } finally { + if (termination == Termination.SIGKILL) { + first.destroyForcibly().waitFor(); + } else { + stop(first); + } + } + + final Process second = startProcess("second", MiniClusterJobEntryPoint.class, arguments); + try { + final JsonNode restored = + awaitRest( + second, + "second", + checkpointsPath(), + "a restored checkpoint", + json -> json.at("/latest/restored/id").isNumber()) + .at("/latest/restored"); + assertThat(restored.get("is_savepoint").asBoolean()).isFalse(); + assertThat(restored.get("id").asLong()) + .as("job resumes from latest checkpoint") + .isGreaterThanOrEqualTo(lastCheckpointBeforeRestart); + + awaitJobState(second, "second", "RUNNING"); + assertThat(second.isAlive()).isTrue(); + System.out.println("main() invocations: " + mainInvocations()); Review Comment: turn into assertion? -- 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]
