<commit_before>/*
 * 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.test.recovery;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.testkit.TestActorRef;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.filefilter.TrueFileFilter;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.akka.ListeningBehaviour;
import org.apache.flink.runtime.instance.ActorGateway;
import org.apache.flink.runtime.instance.AkkaActorGateway;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobmanager.SubmittedJobGraph;
import org.apache.flink.runtime.jobmanager.Tasks;
import org.apache.flink.runtime.leaderelection.TestingListener;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.messages.JobManagerMessages;
import org.apache.flink.runtime.messages.JobManagerMessages.JobStatusResponse;
import org.apache.flink.runtime.messages.JobManagerMessages.LeaderSessionMessage;
import org.apache.flink.runtime.messages.JobManagerMessages.SubmitJob;
import org.apache.flink.runtime.taskmanager.TaskManager;
import org.apache.flink.runtime.testingUtils.TestingCluster;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.runtime.testutils.JobManagerActorTestUtils;
import org.apache.flink.runtime.testutils.JobManagerProcess;
import org.apache.flink.runtime.testutils.ZooKeeperTestUtils;
import org.apache.flink.runtime.util.ZooKeeperUtils;
import org.apache.flink.runtime.zookeeper.ZooKeeperTestEnvironment;
import org.apache.flink.util.TestLogger;
import org.apache.zookeeper.data.Stat;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.Test;
import scala.Option;
import scala.Some;
import scala.Tuple2;
import scala.concurrent.duration.Deadline;
import scala.concurrent.duration.FiniteDuration;

import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;

/**
 * Tests recovery of {@link SubmittedJobGraph} instances.
 */
public class JobManagerHAJobGraphRecoveryITCase extends TestLogger {

	private final static ZooKeeperTestEnvironment ZooKeeper = new ZooKeeperTestEnvironment(1);

	private final static FiniteDuration TestTimeOut = new FiniteDuration(5, TimeUnit.MINUTES);

	private static final File FileStateBackendBasePath;

	static {
		try {
			FileStateBackendBasePath = CommonTestUtils.createTempDirectory();
		}
		catch (IOException e) {
			throw new RuntimeException("Error in test setup. Could not create directory.", e);
		}
	}

	@AfterClass
	public static void tearDown() throws Exception {
		ZooKeeper.shutdown();

		if (FileStateBackendBasePath != null) {
			FileUtils.deleteDirectory(FileStateBackendBasePath);
		}
	}

	@Before
	public void cleanUp() throws Exception {
		if (FileStateBackendBasePath != null) {
			FileUtils.cleanDirectory(FileStateBackendBasePath);
		}

		ZooKeeper.deleteAll();
	}

	// ---------------------------------------------------------------------------------------------

	/**
	 * Tests that the recovery state is cleaned up after a JobManager stops.
	 */
	@Test
	public void testJobManagerCleanUp() throws Exception {
		Configuration config = ZooKeeperTestUtils.createZooKeeperRecoveryModeConfig(
				ZooKeeper.getConnectString(), FileStateBackendBasePath.getPath());

		// Configure the cluster
		config.setInteger(ConfigConstants.LOCAL_NUMBER_JOB_MANAGER, 1);
		config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1);

		TestingCluster flink = new TestingCluster(config, false, false);

		try {
			final Deadline deadline = TestTimeOut.fromNow();

			// Start the JobManager and TaskManager
			flink.start(true);

			JobGraph jobGraph = createBlockingJobGraph();

			ActorGateway jobManager = flink.getLeaderGateway(deadline.timeLeft());

			// Submit the job
			jobManager.tell(new SubmitJob(jobGraph, ListeningBehaviour.DETACHED));

			// Wait for the job to start
			JobManagerActorTestUtils.waitForJobStatus(jobGraph.getJobID(), JobStatus.RUNNING,
					jobManager, deadline.timeLeft());
		}
		finally {
			flink.shutdown();
		}

		// Verify that everything is clean
		verifyCleanRecoveryState(config);
	}

	/**
	 * Tests that submissions to non-leaders are handled.
	 */
	@Test
	public void testSubmitJobToNonLeader() throws Exception {
		Configuration config = ZooKeeperTestUtils.createZooKeeperRecoveryModeConfig(
				ZooKeeper.getConnectString(), FileStateBackendBasePath.getPath());

		// Configure the cluster
		config.setInteger(ConfigConstants.LOCAL_NUMBER_JOB_MANAGER, 2);
		config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1);

		TestingCluster flink = new TestingCluster(config, false, false);

		try {
			final Deadline deadline = TestTimeOut.fromNow();

			// Start the JobManager and TaskManager
			flink.start(true);

			JobGraph jobGraph = createBlockingJobGraph();

			List<ActorRef> bothJobManagers = flink.getJobManagersAsJava();

			ActorGateway leadingJobManager = flink.getLeaderGateway(deadline.timeLeft());

			ActorGateway nonLeadingJobManager;
			if (bothJobManagers.get(0).equals(leadingJobManager.actor())) {
				nonLeadingJobManager = new AkkaActorGateway(bothJobManagers.get(1), null);
			}
			else {
				nonLeadingJobManager = new AkkaActorGateway(bothJobManagers.get(0), null);
			}

			// Submit the job
			nonLeadingJobManager.tell(new SubmitJob(jobGraph, ListeningBehaviour.DETACHED));

			// Wait for the job to start. We are asking the *leading** JM here although we've
			// submitted the job to the non-leading JM. This is the behaviour under test.
			JobManagerActorTestUtils.waitForJobStatus(jobGraph.getJobID(), JobStatus.RUNNING,
					leadingJobManager, deadline.timeLeft());

			log.info("Wait that the non-leader removes the submitted job.");

			// Make sure that the **non-leading** JM has actually removed the job graph from its
			// local state.
			boolean success = false;
			while (!success && deadline.hasTimeLeft()) {
				JobStatusResponse jobStatusResponse = JobManagerActorTestUtils.requestJobStatus(
						jobGraph.getJobID(), nonLeadingJobManager, deadline.timeLeft());

				if (jobStatusResponse instanceof JobManagerMessages.JobNotFound) {
					success = true;
				}
				else {
					log.info(((JobManagerMessages.CurrentJobStatus)jobStatusResponse).status().toString());
					Thread.sleep(100);
				}
			}<commit_msg>[tests] Add log statements to JobManagerHAJobGraphRecoveryITCase<commit_after>/*
 * 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.test.recovery;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.testkit.TestActorRef;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.filefilter.TrueFileFilter;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.akka.ListeningBehaviour;
import org.apache.flink.runtime.instance.ActorGateway;
import org.apache.flink.runtime.instance.AkkaActorGateway;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobmanager.SubmittedJobGraph;
import org.apache.flink.runtime.jobmanager.Tasks;
import org.apache.flink.runtime.leaderelection.TestingListener;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.messages.JobManagerMessages;
import org.apache.flink.runtime.messages.JobManagerMessages.JobStatusResponse;
import org.apache.flink.runtime.messages.JobManagerMessages.LeaderSessionMessage;
import org.apache.flink.runtime.messages.JobManagerMessages.SubmitJob;
import org.apache.flink.runtime.taskmanager.TaskManager;
import org.apache.flink.runtime.testingUtils.TestingCluster;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.runtime.testutils.JobManagerActorTestUtils;
import org.apache.flink.runtime.testutils.JobManagerProcess;
import org.apache.flink.runtime.testutils.ZooKeeperTestUtils;
import org.apache.flink.runtime.util.ZooKeeperUtils;
import org.apache.flink.runtime.zookeeper.ZooKeeperTestEnvironment;
import org.apache.flink.util.TestLogger;
import org.apache.zookeeper.data.Stat;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.Test;
import scala.Option;
import scala.Some;
import scala.Tuple2;
import scala.concurrent.duration.Deadline;
import scala.concurrent.duration.FiniteDuration;

import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;

/**
 * Tests recovery of {@link SubmittedJobGraph} instances.
 */
public class JobManagerHAJobGraphRecoveryITCase extends TestLogger {

	private final static ZooKeeperTestEnvironment ZooKeeper = new ZooKeeperTestEnvironment(1);

	private final static FiniteDuration TestTimeOut = new FiniteDuration(5, TimeUnit.MINUTES);

	private static final File FileStateBackendBasePath;

	static {
		try {
			FileStateBackendBasePath = CommonTestUtils.createTempDirectory();
		}
		catch (IOException e) {
			throw new RuntimeException("Error in test setup. Could not create directory.", e);
		}
	}

	@AfterClass
	public static void tearDown() throws Exception {
		ZooKeeper.shutdown();

		if (FileStateBackendBasePath != null) {
			FileUtils.deleteDirectory(FileStateBackendBasePath);
		}
	}

	@Before
	public void cleanUp() throws Exception {
		if (FileStateBackendBasePath != null) {
			FileUtils.cleanDirectory(FileStateBackendBasePath);
		}

		ZooKeeper.deleteAll();
	}

	// ---------------------------------------------------------------------------------------------

	/**
	 * Tests that the recovery state is cleaned up after a JobManager stops.
	 */
	@Test
	public void testJobManagerCleanUp() throws Exception {
		Configuration config = ZooKeeperTestUtils.createZooKeeperRecoveryModeConfig(
				ZooKeeper.getConnectString(), FileStateBackendBasePath.getPath());

		// Configure the cluster
		config.setInteger(ConfigConstants.LOCAL_NUMBER_JOB_MANAGER, 1);
		config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1);

		TestingCluster flink = new TestingCluster(config, false, false);

		try {
			final Deadline deadline = TestTimeOut.fromNow();

			// Start the JobManager and TaskManager
			flink.start(true);

			JobGraph jobGraph = createBlockingJobGraph();

			ActorGateway jobManager = flink.getLeaderGateway(deadline.timeLeft());

			// Submit the job
			jobManager.tell(new SubmitJob(jobGraph, ListeningBehaviour.DETACHED));

			// Wait for the job to start
			JobManagerActorTestUtils.waitForJobStatus(jobGraph.getJobID(), JobStatus.RUNNING,
					jobManager, deadline.timeLeft());
		}
		finally {
			flink.shutdown();
		}

		// Verify that everything is clean
		verifyCleanRecoveryState(config);
	}

	/**
	 * Tests that submissions to non-leaders are handled.
	 */
	@Test
	public void testSubmitJobToNonLeader() throws Exception {
		Configuration config = ZooKeeperTestUtils.createZooKeeperRecoveryModeConfig(
				ZooKeeper.getConnectString(), FileStateBackendBasePath.getPath());

		// Configure the cluster
		config.setInteger(ConfigConstants.LOCAL_NUMBER_JOB_MANAGER, 2);
		config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1);

		TestingCluster flink = new TestingCluster(config, false, false);

		try {
			final Deadline deadline = TestTimeOut.fromNow();

			// Start the JobManager and TaskManager
			flink.start(true);

			JobGraph jobGraph = createBlockingJobGraph();

			List<ActorRef> bothJobManagers = flink.getJobManagersAsJava();

			ActorGateway leadingJobManager = flink.getLeaderGateway(deadline.timeLeft());

			ActorGateway nonLeadingJobManager;
			if (bothJobManagers.get(0).equals(leadingJobManager.actor())) {
				nonLeadingJobManager = new AkkaActorGateway(bothJobManagers.get(1), null);
			}
			else {
				nonLeadingJobManager = new AkkaActorGateway(bothJobManagers.get(0), null);
			}

			log.info("Leading job manager: " + leadingJobManager);
			log.info("Non-leading job manager: " + nonLeadingJobManager);

			// Submit the job
			nonLeadingJobManager.tell(new SubmitJob(jobGraph, ListeningBehaviour.DETACHED));

			log.info("Submitted job graph to " + nonLeadingJobManager);

			// Wait for the job to start. We are asking the *leading** JM here although we've
			// submitted the job to the non-leading JM. This is the behaviour under test.
			JobManagerActorTestUtils.waitForJobStatus(jobGraph.getJobID(), JobStatus.RUNNING,
					leadingJobManager, deadline.timeLeft());

			log.info("Wait that the non-leader removes the submitted job.");

			// Make sure that the **non-leading** JM has actually removed the job graph from its
			// local state.
			boolean success = false;
			while (!success && deadline.hasTimeLeft()) {
				JobStatusResponse jobStatusResponse = JobManagerActorTestUtils.requestJobStatus(
						jobGraph.getJobID(), nonLeadingJobManager, deadline.timeLeft());

				if (jobStatusResponse instanceof JobManagerMessages.JobNotFound) {
					success = true;
				}
				else {
					log.info(((JobManagerMessages.CurrentJobStatus)jobStatusResponse).status().toString());
					Thread.sleep(100);
				}
			}
{'fraction_non_alphanumeric': 0.1039626144292939, 'fraction_numerical': 0.0021765571986428526, 'mean_word_length': 4.730741012472487, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3321543', 'n_tokens_mistral': 4916, 'n_tokens_neox': 4480, 'n_words': 1056}