This is an automated email from the ASF dual-hosted git repository. chesnay pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push: new 228a3da [FLINK-11593][tests] Port TaskManagerTest to new code base 228a3da is described below commit 228a3da1b974640c477d61b5390805b54e9610f9 Author: Clark Yang <yangshi...@youzan.com> AuthorDate: Wed Apr 10 18:09:50 2019 +0800 [FLINK-11593][tests] Port TaskManagerTest to new code base --- .../jobmaster/TestingAbstractInvokables.java | 36 + .../taskexecutor/TaskExecutorSubmissionTest.java | 951 +++++++++ .../runtime/taskexecutor/TaskExecutorTest.java | 44 + .../TaskManagerServicesConfigurationTest.java | 29 + .../TaskSubmissionTestEnvironment.java | 363 ++++ .../taskexecutor/TestTaskManagerActions.java | 113 + .../flink/runtime/taskmanager/TaskManagerTest.java | 2185 -------------------- .../flink/runtime/testingUtils/TestingUtils.scala | 186 -- 8 files changed, 1536 insertions(+), 2371 deletions(-) diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingAbstractInvokables.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingAbstractInvokables.java index 4e8d7eb..e97dec7 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingAbstractInvokables.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingAbstractInvokables.java @@ -25,6 +25,8 @@ import org.apache.flink.runtime.io.network.api.writer.RecordWriterBuilder; import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable; import org.apache.flink.types.IntValue; +import java.util.concurrent.CompletableFuture; + /** * {@link AbstractInvokable} for testing purposes. */ @@ -83,4 +85,38 @@ public class TestingAbstractInvokables { } } } + + public static final class TestInvokableRecordCancel extends AbstractInvokable { + + private static CompletableFuture<Boolean> gotCanceledFuture = new CompletableFuture<>(); + + public TestInvokableRecordCancel(Environment environment) { + super(environment); + } + + @Override + public void invoke() throws Exception { + final Object o = new Object(); + RecordWriter<IntValue> recordWriter = new RecordWriterBuilder().build(getEnvironment().getWriter(0)); + for (int i = 0; i < 1024; i++) { + recordWriter.emit(new IntValue(42)); + } + + gotCanceledFuture.get(); + + } + + @Override + public void cancel() { + gotCanceledFuture.complete(true); + } + + public static void resetGotCanceledFuture() { + gotCanceledFuture = new CompletableFuture<>(); + } + + public static CompletableFuture<Boolean> gotCanceled() { + return gotCanceledFuture; + } + } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java new file mode 100644 index 0000000..3bb8354 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java @@ -0,0 +1,951 @@ +/* + * 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.runtime.taskexecutor; + +import org.apache.flink.api.common.ExecutionConfig; +import org.apache.flink.api.common.JobID; +import org.apache.flink.api.common.time.Time; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.TaskManagerOptions; +import org.apache.flink.runtime.blob.PermanentBlobKey; +import org.apache.flink.runtime.clusterframework.types.AllocationID; +import org.apache.flink.runtime.concurrent.FutureUtils; +import org.apache.flink.runtime.deployment.InputChannelDeploymentDescriptor; +import org.apache.flink.runtime.deployment.InputGateDeploymentDescriptor; +import org.apache.flink.runtime.deployment.ResultPartitionDeploymentDescriptor; +import org.apache.flink.runtime.deployment.ResultPartitionLocation; +import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor; +import org.apache.flink.runtime.execution.ExecutionState; +import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; +import org.apache.flink.runtime.executiongraph.ExecutionGraphException; +import org.apache.flink.runtime.executiongraph.JobInformation; +import org.apache.flink.runtime.executiongraph.PartitionInfo; +import org.apache.flink.runtime.executiongraph.TaskInformation; +import org.apache.flink.runtime.io.network.ConnectionID; +import org.apache.flink.runtime.io.network.partition.PartitionNotFoundException; +import org.apache.flink.runtime.io.network.partition.ResultPartitionID; +import org.apache.flink.runtime.io.network.partition.ResultPartitionType; +import org.apache.flink.runtime.jobgraph.IntermediateDataSetID; +import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID; +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable; +import org.apache.flink.runtime.jobmanager.Tasks; +import org.apache.flink.runtime.jobmaster.JobMasterId; +import org.apache.flink.runtime.jobmaster.TestingAbstractInvokables; +import org.apache.flink.runtime.jobmaster.utils.TestingJobMasterGateway; +import org.apache.flink.runtime.jobmaster.utils.TestingJobMasterGatewayBuilder; +import org.apache.flink.runtime.messages.Acknowledge; +import org.apache.flink.runtime.messages.StackTraceSampleResponse; +import org.apache.flink.runtime.taskexecutor.exceptions.PartitionException; +import org.apache.flink.runtime.taskexecutor.exceptions.TaskException; +import org.apache.flink.runtime.taskexecutor.slot.TaskSlotTable; +import org.apache.flink.runtime.testtasks.BlockingNoOpInvokable; +import org.apache.flink.runtime.testutils.StoppableInvokable; +import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.NetUtils; +import org.apache.flink.util.Preconditions; +import org.apache.flink.util.SerializedValue; +import org.apache.flink.util.TestLogger; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TestName; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.URL; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CompletableFuture; + +import static org.hamcrest.CoreMatchers.startsWith; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +/** + * Tests for submission logic of the {@link TaskExecutor}. + */ +public class TaskExecutorSubmissionTest extends TestLogger { + + @Rule + public final TestName testName = new TestName(); + + private static final Time timeout = Time.milliseconds(10000L); + + private JobID jobId = new JobID(); + + /** + * Tests that we can submit a task to the TaskManager given that we've allocated a slot there. + */ + @Test(timeout = 10000L) + public void testTaskSubmission() throws Exception { + final ExecutionAttemptID eid = new ExecutionAttemptID(); + + final TaskDeploymentDescriptor tdd = createTestTaskDeploymentDescriptor("test task", eid, TaskExecutorTest.TestInvokable.class); + + final CompletableFuture<Void> taskRunningFuture = new CompletableFuture<>(); + + try (TaskSubmissionTestEnvironment env = + new TaskSubmissionTestEnvironment.Builder(jobId) + .setSlotSize(1) + .addTaskManagerActionListener(eid, ExecutionState.RUNNING, taskRunningFuture) + .build()) { + TaskExecutorGateway tmGateway = env.getTaskExecutorGateway(); + TaskSlotTable taskSlotTable = env.getTaskSlotTable(); + + taskSlotTable.allocateSlot(0, jobId, tdd.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd, env.getJobMasterId(), timeout).get(); + + taskRunningFuture.get(); + } + } + + /** + * Tests that the TaskManager sends a proper exception back to the sender if the submit task + * message fails. + */ + @Test(timeout = 10000L) + public void testSubmitTaskFailure() throws Exception { + final ExecutionAttemptID eid = new ExecutionAttemptID(); + + final TaskDeploymentDescriptor tdd = createTestTaskDeploymentDescriptor( + "test task", + eid, + BlockingNoOpInvokable.class, + 0); // this will make the submission fail because the number of key groups must be >= 1 + + try (TaskSubmissionTestEnvironment env = + new TaskSubmissionTestEnvironment.Builder(jobId) + .build()) { + TaskExecutorGateway tmGateway = env.getTaskExecutorGateway(); + TaskSlotTable taskSlotTable = env.getTaskSlotTable(); + + taskSlotTable.allocateSlot(0, jobId, tdd.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd, env.getJobMasterId(), timeout).get(); + } catch (Exception e) { + assertThat(e.getCause(), instanceOf(IllegalArgumentException.class)); + } + } + + /** + * Tests that we can cancel the task of the TaskManager given that we've submitted it. + */ + @Test(timeout = 10000L) + public void testTaskSubmissionAndCancelling() throws Exception { + final ExecutionAttemptID eid1 = new ExecutionAttemptID(); + final ExecutionAttemptID eid2 = new ExecutionAttemptID(); + + final TaskDeploymentDescriptor tdd1 = createTestTaskDeploymentDescriptor("test task", eid1, BlockingNoOpInvokable.class); + final TaskDeploymentDescriptor tdd2 = createTestTaskDeploymentDescriptor("test task", eid2, BlockingNoOpInvokable.class); + + final CompletableFuture<Void> task1RunningFuture = new CompletableFuture<>(); + final CompletableFuture<Void> task2RunningFuture = new CompletableFuture<>(); + final CompletableFuture<Void> task1CanceledFuture = new CompletableFuture<>(); + + try (TaskSubmissionTestEnvironment env = + new TaskSubmissionTestEnvironment.Builder(jobId) + .setSlotSize(2) + .addTaskManagerActionListener(eid1, ExecutionState.RUNNING, task1RunningFuture) + .addTaskManagerActionListener(eid2, ExecutionState.RUNNING, task2RunningFuture) + .addTaskManagerActionListener(eid1, ExecutionState.CANCELED, task1CanceledFuture) + .build()) { + TaskExecutorGateway tmGateway = env.getTaskExecutorGateway(); + TaskSlotTable taskSlotTable = env.getTaskSlotTable(); + + taskSlotTable.allocateSlot(0, jobId, tdd1.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd1, env.getJobMasterId(), timeout).get(); + task1RunningFuture.get(); + + taskSlotTable.allocateSlot(1, jobId, tdd2.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd2, env.getJobMasterId(), timeout).get(); + task2RunningFuture.get(); + + assertSame(taskSlotTable.getTask(eid1).getExecutionState(), ExecutionState.RUNNING); + assertSame(taskSlotTable.getTask(eid2).getExecutionState(), ExecutionState.RUNNING); + + tmGateway.cancelTask(eid1, timeout); + task1CanceledFuture.get(); + + assertSame(taskSlotTable.getTask(eid1).getExecutionState(), ExecutionState.CANCELED); + assertSame(taskSlotTable.getTask(eid2).getExecutionState(), ExecutionState.RUNNING); + } + } + + /** + * Tests that we can stop the task of the TaskManager given that we've submitted it. + */ + @Test(timeout = 10000L) + public void testTaskSubmissionAndStop() throws Exception { + final ExecutionAttemptID eid1 = new ExecutionAttemptID(); + final ExecutionAttemptID eid2 = new ExecutionAttemptID(); + + final TaskDeploymentDescriptor tdd1 = createTestTaskDeploymentDescriptor("test task", eid1, StoppableInvokable.class); + final TaskDeploymentDescriptor tdd2 = createTestTaskDeploymentDescriptor("test task", eid2, BlockingNoOpInvokable.class); + + final CompletableFuture<Void> task1RunningFuture = new CompletableFuture<>(); + final CompletableFuture<Void> task2RunningFuture = new CompletableFuture<>(); + final CompletableFuture<Void> task1FinishedFuture = new CompletableFuture<>(); + + try (TaskSubmissionTestEnvironment env = + new TaskSubmissionTestEnvironment.Builder(jobId) + .setSlotSize(2) + .addTaskManagerActionListener(eid1, ExecutionState.RUNNING, task1RunningFuture) + .addTaskManagerActionListener(eid2, ExecutionState.RUNNING, task2RunningFuture) + .addTaskManagerActionListener(eid1, ExecutionState.FINISHED, task1FinishedFuture) + .build()) { + TaskExecutorGateway tmGateway = env.getTaskExecutorGateway(); + TaskSlotTable taskSlotTable = env.getTaskSlotTable(); + + taskSlotTable.allocateSlot(0, jobId, tdd1.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd1, env.getJobMasterId(), timeout).get(); + task1RunningFuture.get(); + + taskSlotTable.allocateSlot(1, jobId, tdd2.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd2, env.getJobMasterId(), timeout).get(); + task2RunningFuture.get(); + + assertSame(taskSlotTable.getTask(eid1).getExecutionState(), ExecutionState.RUNNING); + assertSame(taskSlotTable.getTask(eid2).getExecutionState(), ExecutionState.RUNNING); + + tmGateway.stopTask(eid1, timeout); + task1FinishedFuture.get(); + + // task 2 does not implement StoppableTask which should cause the stop operation to fail + CompletableFuture<Acknowledge> acknowledgeOfTask2 = tmGateway.stopTask(eid2, timeout); + boolean hasTaskException = false; + try { + acknowledgeOfTask2.get(); + } catch (Throwable e) { + hasTaskException = ExceptionUtils.findThrowable(e, TaskException.class).isPresent(); + } + + assertTrue(hasTaskException); + assertSame(taskSlotTable.getTask(eid1).getExecutionState(), ExecutionState.FINISHED); + assertSame(taskSlotTable.getTask(eid2).getExecutionState(), ExecutionState.RUNNING); + } + } + + /** + * Tests that the TaskManager sends a proper exception back to the sender if the stop task + * message fails. + */ + @Test(timeout = 10000L) + public void testStopTaskFailure() throws Exception { + final ExecutionAttemptID eid = new ExecutionAttemptID(); + + final TaskDeploymentDescriptor tdd = createTestTaskDeploymentDescriptor("test task", eid, BlockingNoOpInvokable.class); + + final CompletableFuture<Void> taskRunningFuture = new CompletableFuture<>(); + + try (TaskSubmissionTestEnvironment env = + new TaskSubmissionTestEnvironment.Builder(jobId) + .setSlotSize(1) + .addTaskManagerActionListener(eid, ExecutionState.RUNNING, taskRunningFuture) + .build()) { + TaskExecutorGateway tmGateway = env.getTaskExecutorGateway(); + TaskSlotTable taskSlotTable = env.getTaskSlotTable(); + + taskSlotTable.allocateSlot(0, jobId, tdd.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd, env.getJobMasterId(), timeout).get(); + taskRunningFuture.get(); + + CompletableFuture<Acknowledge> stopFuture = tmGateway.stopTask(eid, timeout); + try { + stopFuture.get(); + } catch (Exception e) { + assertTrue(e.getCause() instanceof TaskException); + assertThat(e.getCause().getMessage(), startsWith("Cannot stop task for execution")); + } + } + } + + /** + * Tests that submitted tasks will fail when attempting to send/receive data if no + * ResultPartitions/InputGates are set up. + */ + @Test(timeout = 10000L) + public void testGateChannelEdgeMismatch() throws Exception { + final ExecutionAttemptID eid1 = new ExecutionAttemptID(); + final ExecutionAttemptID eid2 = new ExecutionAttemptID(); + + final TaskDeploymentDescriptor tdd1 = + createTestTaskDeploymentDescriptor("Sender", eid1, TestingAbstractInvokables.Sender.class); + final TaskDeploymentDescriptor tdd2 = + createTestTaskDeploymentDescriptor("Receiver", eid2, TestingAbstractInvokables.Receiver.class); + + final CompletableFuture<Void> task1RunningFuture = new CompletableFuture<>(); + final CompletableFuture<Void> task2RunningFuture = new CompletableFuture<>(); + final CompletableFuture<Void> task1FailedFuture = new CompletableFuture<>(); + final CompletableFuture<Void> task2FailedFuture = new CompletableFuture<>(); + + try (TaskSubmissionTestEnvironment env = + new TaskSubmissionTestEnvironment.Builder(jobId) + .addTaskManagerActionListener(eid1, ExecutionState.RUNNING, task1RunningFuture) + .addTaskManagerActionListener(eid2, ExecutionState.RUNNING, task2RunningFuture) + .addTaskManagerActionListener(eid1, ExecutionState.FAILED, task1FailedFuture) + .addTaskManagerActionListener(eid2, ExecutionState.FAILED, task2FailedFuture) + .setSlotSize(2) + .build()) { + TaskExecutorGateway tmGateway = env.getTaskExecutorGateway(); + TaskSlotTable taskSlotTable = env.getTaskSlotTable(); + + taskSlotTable.allocateSlot(0, jobId, tdd1.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd1, env.getJobMasterId(), timeout).get(); + task1RunningFuture.get(); + + taskSlotTable.allocateSlot(1, jobId, tdd2.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd2, env.getJobMasterId(), timeout).get(); + task2RunningFuture.get(); + + task1FailedFuture.get(); + task2FailedFuture.get(); + + assertSame(taskSlotTable.getTask(eid1).getExecutionState(), ExecutionState.FAILED); + assertSame(taskSlotTable.getTask(eid2).getExecutionState(), ExecutionState.FAILED); + } + } + + @Test(timeout = 10000L) + public void testRunJobWithForwardChannel() throws Exception { + final ExecutionAttemptID eid1 = new ExecutionAttemptID(); + final ExecutionAttemptID eid2 = new ExecutionAttemptID(); + + IntermediateResultPartitionID partitionId = new IntermediateResultPartitionID(); + + ResultPartitionDeploymentDescriptor task1ResultPartitionDescriptor = + new ResultPartitionDeploymentDescriptor(new IntermediateDataSetID(), partitionId, ResultPartitionType.PIPELINED, 1, + 1, true); + + InputGateDeploymentDescriptor task2InputGateDescriptor = + new InputGateDeploymentDescriptor(new IntermediateDataSetID(), ResultPartitionType.PIPELINED, 0, + new InputChannelDeploymentDescriptor[] { + new InputChannelDeploymentDescriptor(new ResultPartitionID(partitionId, eid1), + ResultPartitionLocation.createLocal()) }); + + final TaskDeploymentDescriptor tdd1 = + createTestTaskDeploymentDescriptor( + "Sender", + eid1, + TestingAbstractInvokables.Sender.class, + 1, + Collections.singletonList(task1ResultPartitionDescriptor), + Collections.emptyList()); + final TaskDeploymentDescriptor tdd2 = + createTestTaskDeploymentDescriptor( + "Receiver", + eid2, + TestingAbstractInvokables.Receiver.class, + 1, + Collections.emptyList(), + Collections.singletonList(task2InputGateDescriptor)); + + final CompletableFuture<Void> task1RunningFuture = new CompletableFuture<>(); + final CompletableFuture<Void> task2RunningFuture = new CompletableFuture<>(); + final CompletableFuture<Void> task1FinishedFuture = new CompletableFuture<>(); + final CompletableFuture<Void> task2FinishedFuture = new CompletableFuture<>(); + + final JobMasterId jobMasterId = JobMasterId.generate(); + TestingJobMasterGateway testingJobMasterGateway = + new TestingJobMasterGatewayBuilder() + .setFencingTokenSupplier(() -> jobMasterId) + .setScheduleOrUpdateConsumersFunction( + resultPartitionID -> CompletableFuture.completedFuture(Acknowledge.get())) + .build(); + + try (TaskSubmissionTestEnvironment env = + new TaskSubmissionTestEnvironment.Builder(jobId) + .setSlotSize(2) + .addTaskManagerActionListener(eid1, ExecutionState.RUNNING, task1RunningFuture) + .addTaskManagerActionListener(eid2, ExecutionState.RUNNING, task2RunningFuture) + .addTaskManagerActionListener(eid1, ExecutionState.FINISHED, task1FinishedFuture) + .addTaskManagerActionListener(eid2, ExecutionState.FINISHED, task2FinishedFuture) + .setJobMasterId(jobMasterId) + .setJobMasterGateway(testingJobMasterGateway) + .setMockNetworkEnvironment(false) + .build()) { + TaskExecutorGateway tmGateway = env.getTaskExecutorGateway(); + TaskSlotTable taskSlotTable = env.getTaskSlotTable(); + + taskSlotTable.allocateSlot(0, jobId, tdd1.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd1, jobMasterId, timeout).get(); + task1RunningFuture.get(); + + taskSlotTable.allocateSlot(1, jobId, tdd2.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd2, jobMasterId, timeout).get(); + task2RunningFuture.get(); + + task1FinishedFuture.get(); + task2FinishedFuture.get(); + + assertSame(taskSlotTable.getTask(eid1).getExecutionState(), ExecutionState.FINISHED); + assertSame(taskSlotTable.getTask(eid2).getExecutionState(), ExecutionState.FINISHED); + } + } + + /** + * This tests creates two tasks. The sender sends data but fails to send the + * state update back to the job manager. + * the second one blocks to be canceled + */ + @Test(timeout = 10000L) + public void testCancellingDependentAndStateUpdateFails() throws Exception { + final ExecutionAttemptID eid1 = new ExecutionAttemptID(); + final ExecutionAttemptID eid2 = new ExecutionAttemptID(); + + IntermediateResultPartitionID partitionId = new IntermediateResultPartitionID(); + + ResultPartitionDeploymentDescriptor task1ResultPartitionDescriptor = + new ResultPartitionDeploymentDescriptor(new IntermediateDataSetID(), partitionId, ResultPartitionType.PIPELINED, 1, + 1, true); + + InputGateDeploymentDescriptor task2InputGateDescriptor = + new InputGateDeploymentDescriptor(new IntermediateDataSetID(), ResultPartitionType.PIPELINED, 0, + new InputChannelDeploymentDescriptor[] { + new InputChannelDeploymentDescriptor(new ResultPartitionID(partitionId, eid1), + ResultPartitionLocation.createLocal()) }); + + final TaskDeploymentDescriptor tdd1 = + createTestTaskDeploymentDescriptor("Sender", + eid1, + TestingAbstractInvokables.Sender.class, 1, + Collections.singletonList(task1ResultPartitionDescriptor), + Collections.emptyList()); + final TaskDeploymentDescriptor tdd2 = + createTestTaskDeploymentDescriptor("Receiver", + eid2, + TestingAbstractInvokables.Receiver.class, + 1, + Collections.emptyList(), + Collections.singletonList(task2InputGateDescriptor)); + + final CompletableFuture<Void> task1RunningFuture = new CompletableFuture<>(); + final CompletableFuture<Void> task2RunningFuture = new CompletableFuture<>(); + final CompletableFuture<Void> task1FailedFuture = new CompletableFuture<>(); + final CompletableFuture<Void> task2CanceledFuture = new CompletableFuture<>(); + + final JobMasterId jobMasterId = JobMasterId.generate(); + TestingJobMasterGateway testingJobMasterGateway = + new TestingJobMasterGatewayBuilder() + .setFencingTokenSupplier(() -> jobMasterId) + .setUpdateTaskExecutionStateFunction(taskExecutionState -> { + if (taskExecutionState != null && taskExecutionState.getID().equals(eid1)) { + return FutureUtils.completedExceptionally( + new ExecutionGraphException("The execution attempt " + eid2 + " was not found.")); + } else { + return CompletableFuture.completedFuture(Acknowledge.get()); + } + }) + .build(); + + try (TaskSubmissionTestEnvironment env = + new TaskSubmissionTestEnvironment.Builder(jobId) + .setSlotSize(2) + .addTaskManagerActionListener(eid1, ExecutionState.RUNNING, task1RunningFuture) + .addTaskManagerActionListener(eid2, ExecutionState.RUNNING, task2RunningFuture) + .addTaskManagerActionListener(eid1, ExecutionState.FAILED, task1FailedFuture) + .addTaskManagerActionListener(eid2, ExecutionState.CANCELED, task2CanceledFuture) + .setJobMasterId(jobMasterId) + .setJobMasterGateway(testingJobMasterGateway) + .setMockNetworkEnvironment(false) + .build()) { + TaskExecutorGateway tmGateway = env.getTaskExecutorGateway(); + TaskSlotTable taskSlotTable = env.getTaskSlotTable(); + + taskSlotTable.allocateSlot(0, jobId, tdd1.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd1, jobMasterId, timeout).get(); + task1RunningFuture.get(); + + taskSlotTable.allocateSlot(1, jobId, tdd2.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd2, jobMasterId, timeout).get(); + task2RunningFuture.get(); + + task1FailedFuture.get(); + assertSame(taskSlotTable.getTask(eid1).getExecutionState(), ExecutionState.FAILED); + + tmGateway.cancelTask(eid2, timeout); + + task2CanceledFuture.get(); + assertSame(taskSlotTable.getTask(eid2).getExecutionState(), ExecutionState.CANCELED); + } + } + + /** + * Tests that repeated remote {@link PartitionNotFoundException}s ultimately fail the receiver. + */ + @Test(timeout = 10000L) + public void testRemotePartitionNotFound() throws Exception { + final ExecutionAttemptID eid = new ExecutionAttemptID(); + + final IntermediateDataSetID resultId = new IntermediateDataSetID(); + final ResultPartitionID partitionId = new ResultPartitionID(); + + final int dataPort = NetUtils.getAvailablePort(); + Configuration config = new Configuration(); + config.setInteger(TaskManagerOptions.DATA_PORT, dataPort); + config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100); + config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_MAX, 200); + + // Remote location (on the same TM though) for the partition + final ResultPartitionLocation loc = ResultPartitionLocation + .createRemote(new ConnectionID( + new InetSocketAddress("localhost", dataPort), 0)); + + final InputChannelDeploymentDescriptor[] inputChannelDeploymentDescriptors = + new InputChannelDeploymentDescriptor[] { + new InputChannelDeploymentDescriptor(partitionId, loc)}; + + final InputGateDeploymentDescriptor inputGateDeploymentDescriptor = + new InputGateDeploymentDescriptor(resultId, ResultPartitionType.PIPELINED, 0, inputChannelDeploymentDescriptors); + + final TaskDeploymentDescriptor tdd = + createTestTaskDeploymentDescriptor("Receiver", + eid, + Tasks.AgnosticReceiver.class, 1, + Collections.emptyList(), + Collections.singletonList(inputGateDeploymentDescriptor)); + + final CompletableFuture<Void> taskRunningFuture = new CompletableFuture<>(); + final CompletableFuture<Void> taskFailedFuture = new CompletableFuture<>(); + + try (TaskSubmissionTestEnvironment env = + new TaskSubmissionTestEnvironment.Builder(jobId) + .setSlotSize(2) + .addTaskManagerActionListener(eid, ExecutionState.RUNNING, taskRunningFuture) + .addTaskManagerActionListener(eid, ExecutionState.FAILED, taskFailedFuture) + .setConfiguration(config) + .setLocalCommunication(false) + .setMockNetworkEnvironment(false) + .build()) { + TaskExecutorGateway tmGateway = env.getTaskExecutorGateway(); + TaskSlotTable taskSlotTable = env.getTaskSlotTable(); + + taskSlotTable.allocateSlot(0, jobId, tdd.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd, env.getJobMasterId(), timeout).get(); + taskRunningFuture.get(); + + taskFailedFuture.get(); + assertThat(taskSlotTable.getTask(eid).getFailureCause(), instanceOf(PartitionNotFoundException.class)); + } + } + + /** + * Tests that the TaskManager sends proper exception back to the sender if the partition update fails. + */ + @Test + public void testUpdateTaskInputPartitionsFailure() throws Exception { + final ExecutionAttemptID eid = new ExecutionAttemptID(); + + final TaskDeploymentDescriptor tdd = createTestTaskDeploymentDescriptor("test task", eid, BlockingNoOpInvokable.class); + + final CompletableFuture<Void> taskRunningFuture = new CompletableFuture<>(); + + try (TaskSubmissionTestEnvironment env = + new TaskSubmissionTestEnvironment.Builder(jobId) + .setSlotSize(1) + .addTaskManagerActionListener(eid, ExecutionState.RUNNING, taskRunningFuture) + .build()) { + TaskExecutorGateway tmGateway = env.getTaskExecutorGateway(); + TaskSlotTable taskSlotTable = env.getTaskSlotTable(); + + taskSlotTable.allocateSlot(0, jobId, tdd.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd, env.getJobMasterId(), timeout).get(); + taskRunningFuture.get(); + + CompletableFuture<Acknowledge> updateFuture = tmGateway.updatePartitions( + eid, + Collections.singletonList( + new PartitionInfo( + new IntermediateDataSetID(), + new InputChannelDeploymentDescriptor(new ResultPartitionID(), ResultPartitionLocation.createLocal()))), + timeout); + try { + updateFuture.get(); + fail(); + } catch (Exception e) { + assertTrue(ExceptionUtils.findThrowable(e, PartitionException.class).isPresent()); + } + } + } + + /** + * Tests that repeated local {@link PartitionNotFoundException}s ultimately fail the receiver. + */ + @Test(timeout = 10000L) + public void testLocalPartitionNotFound() throws Exception { + final ExecutionAttemptID eid = new ExecutionAttemptID(); + + final IntermediateDataSetID resultId = new IntermediateDataSetID(); + final ResultPartitionID partitionId = new ResultPartitionID(); + + final ResultPartitionLocation loc = ResultPartitionLocation.createLocal(); + + final InputChannelDeploymentDescriptor[] inputChannelDeploymentDescriptors = + new InputChannelDeploymentDescriptor[] { + new InputChannelDeploymentDescriptor(partitionId, loc)}; + + final InputGateDeploymentDescriptor inputGateDeploymentDescriptor = + new InputGateDeploymentDescriptor(resultId, ResultPartitionType.PIPELINED, 0, inputChannelDeploymentDescriptors); + + final TaskDeploymentDescriptor tdd = + createTestTaskDeploymentDescriptor("Receiver", + eid, + Tasks.AgnosticReceiver.class, + 1, Collections.emptyList(), + Collections.singletonList(inputGateDeploymentDescriptor)); + + Configuration config = new Configuration(); + config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100); + config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_MAX, 200); + + final CompletableFuture<Void> taskRunningFuture = new CompletableFuture<>(); + final CompletableFuture<Void> taskFailedFuture = new CompletableFuture<>(); + + try (TaskSubmissionTestEnvironment env = + new TaskSubmissionTestEnvironment.Builder(jobId) + .setSlotSize(1) + .addTaskManagerActionListener(eid, ExecutionState.RUNNING, taskRunningFuture) + .addTaskManagerActionListener(eid, ExecutionState.FAILED, taskFailedFuture) + .setConfiguration(config) + .setMockNetworkEnvironment(false) + .build()) { + TaskExecutorGateway tmGateway = env.getTaskExecutorGateway(); + TaskSlotTable taskSlotTable = env.getTaskSlotTable(); + + taskSlotTable.allocateSlot(0, jobId, tdd.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd, env.getJobMasterId(), timeout).get(); + taskRunningFuture.get(); + + taskFailedFuture.get(); + + assertSame(taskSlotTable.getTask(eid).getExecutionState(), ExecutionState.FAILED); + assertThat(taskSlotTable.getTask(eid).getFailureCause(), instanceOf(PartitionNotFoundException.class)); + } + } + + /** + * Test that a failing schedule or update consumers call leads to the failing of the respective + * task. + * + * <p>IMPORTANT: We have to make sure that the invokable's cancel method is called, because only + * then the future is completed. We do this by not eagerly deploying consumer tasks and requiring + * the invokable to fill one memory segment. The completed memory segment will trigger the + * scheduling of the downstream operator since it is in pipeline mode. After we've filled the + * memory segment, we'll block the invokable and wait for the task failure due to the failed + * schedule or update consumers call. + */ + @Test(timeout = 10000L) + public void testFailingScheduleOrUpdateConsumers() throws Exception { + final Configuration configuration = new Configuration(); + + // set the memory segment to the smallest size possible, because we have to fill one + // memory buffer to trigger the schedule or update consumers message to the downstream + // operators + configuration.setString(TaskManagerOptions.MEMORY_SEGMENT_SIZE, "4096"); + + final ExecutionAttemptID eid = new ExecutionAttemptID(); + + final ResultPartitionDeploymentDescriptor resultPartitionDeploymentDescriptor = new ResultPartitionDeploymentDescriptor( + new IntermediateDataSetID(), + new IntermediateResultPartitionID(), + ResultPartitionType.PIPELINED, + 1, + 1, + true); + + final TaskDeploymentDescriptor tdd = createTestTaskDeploymentDescriptor( + "test task", + eid, + TestingAbstractInvokables.TestInvokableRecordCancel.class, + 1, + Collections.singletonList(resultPartitionDeploymentDescriptor), + Collections.emptyList()); + + final CompletableFuture<Void> taskRunningFuture = new CompletableFuture<>(); + + final Exception exception = new Exception("Failed schedule or update consumers"); + + final JobMasterId jobMasterId = JobMasterId.generate(); + TestingJobMasterGateway testingJobMasterGateway = + new TestingJobMasterGatewayBuilder() + .setFencingTokenSupplier(() -> jobMasterId) + .setUpdateTaskExecutionStateFunction(resultPartitionID -> FutureUtils.completedExceptionally(exception)) + .build(); + + try (TaskSubmissionTestEnvironment env = + new TaskSubmissionTestEnvironment.Builder(jobId) + .setSlotSize(1) + .setConfiguration(configuration) + .addTaskManagerActionListener(eid, ExecutionState.RUNNING, taskRunningFuture) + .setJobMasterId(jobMasterId) + .setJobMasterGateway(testingJobMasterGateway) + .setMockNetworkEnvironment(false) + .build()) { + TaskExecutorGateway tmGateway = env.getTaskExecutorGateway(); + TaskSlotTable taskSlotTable = env.getTaskSlotTable(); + + TestingAbstractInvokables.TestInvokableRecordCancel.resetGotCanceledFuture(); + + taskSlotTable.allocateSlot(0, jobId, tdd.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd, jobMasterId, timeout).get(); + taskRunningFuture.get(); + + CompletableFuture<Boolean> cancelFuture = TestingAbstractInvokables.TestInvokableRecordCancel.gotCanceled(); + + assertTrue(cancelFuture.get()); + assertTrue(ExceptionUtils.findThrowableWithMessage(taskSlotTable.getTask(eid).getFailureCause(), exception.getMessage()).isPresent()); + } + } + + // ------------------------------------------------------------------------ + // Stack trace sample + // ------------------------------------------------------------------------ + + /** + * Tests sampling of task stack traces. + */ + @Test(timeout = 10000L) + @SuppressWarnings("unchecked") + public void testRequestStackTraceSample() throws Exception { + final ExecutionAttemptID eid = new ExecutionAttemptID(); + final TaskDeploymentDescriptor tdd = createTestTaskDeploymentDescriptor("test task", eid, BlockingNoOpInvokable.class); + + final int sampleId1 = 112223; + final int sampleId2 = 19230; + final int sampleId3 = 1337; + final int sampleId4 = 44; + + final CompletableFuture<Void> taskRunningFuture = new CompletableFuture<>(); + final CompletableFuture<Void> taskCanceledFuture = new CompletableFuture<>(); + + try (TaskSubmissionTestEnvironment env = + new TaskSubmissionTestEnvironment.Builder(jobId) + .setSlotSize(1) + .addTaskManagerActionListener(eid, ExecutionState.RUNNING, taskRunningFuture) + .addTaskManagerActionListener(eid, ExecutionState.CANCELED, taskCanceledFuture) + .build()) { + TaskExecutorGateway tmGateway = env.getTaskExecutorGateway(); + TaskSlotTable taskSlotTable = env.getTaskSlotTable(); + + taskSlotTable.allocateSlot(0, jobId, tdd.getAllocationId(), Time.seconds(60)); + tmGateway.submitTask(tdd, env.getJobMasterId(), timeout).get(); + taskRunningFuture.get(); + + // + // 1) Trigger sample for non-existing task + // + ExecutionAttemptID nonExistTaskEid = new ExecutionAttemptID(); + + CompletableFuture<StackTraceSampleResponse> failedSampleFuture = + tmGateway.requestStackTraceSample(nonExistTaskEid, sampleId1, 100, Time.seconds(60L), 0, timeout); + try { + failedSampleFuture.get(); + } catch (Exception e) { + assertThat(e.getCause(), instanceOf(IllegalStateException.class)); + assertThat(e.getCause().getMessage(), startsWith("Cannot sample task")); + } + + // + // 2) Trigger sample for the blocking task + // + int numSamples = 5; + + CompletableFuture<StackTraceSampleResponse> successfulSampleFuture = + tmGateway.requestStackTraceSample(eid, sampleId2, numSamples, Time.milliseconds(100L), 0, timeout); + + StackTraceSampleResponse response = successfulSampleFuture.get(); + + assertEquals(response.getSampleId(), sampleId2); + assertEquals(response.getExecutionAttemptID(), eid); + + List<StackTraceElement[]> traces = response.getSamples(); + + assertEquals("Number of samples", numSamples, traces.size()); + + for (StackTraceElement[] trace : traces) { + boolean success = false; + for (StackTraceElement elem : trace) { + // Look for BlockingNoOpInvokable#invoke + if (elem.getClassName().equals( + BlockingNoOpInvokable.class.getName())) { + + assertEquals("invoke", elem.getMethodName()); + + success = true; + break; + } + // The BlockingNoOpInvokable might not be invoked here + if (elem.getClassName().equals(TestTaskManagerActions.class.getName())) { + + assertEquals("updateTaskExecutionState", elem.getMethodName()); + + success = true; + break; + } + if (elem.getClassName().equals(Thread.class) && elem.getMethodName().equals("setContextClassLoader")) { + success = true; + } + } + + assertTrue("Unexpected stack trace: " + + Arrays.toString(trace), success); + } + + // + // 3) Trigger sample for the blocking task with max depth + // + int maxDepth = 2; + + CompletableFuture<StackTraceSampleResponse> successfulSampleFutureWithMaxDepth = + tmGateway.requestStackTraceSample(eid, sampleId3, numSamples, Time.milliseconds(100L), maxDepth, timeout); + + StackTraceSampleResponse responseWithMaxDepth = successfulSampleFutureWithMaxDepth.get(); + + assertEquals(sampleId3, responseWithMaxDepth.getSampleId()); + assertEquals(eid, responseWithMaxDepth.getExecutionAttemptID()); + + List<StackTraceElement[]> tracesWithMaxDepth = responseWithMaxDepth.getSamples(); + + assertEquals("Number of samples", numSamples, tracesWithMaxDepth.size()); + + for (StackTraceElement[] trace : tracesWithMaxDepth) { + assertEquals("Max depth", maxDepth, trace.length); + } + + // + // 4) Trigger sample for the blocking task, but cancel it during sampling + // + int sleepTime = 100; + numSamples = 100; + + CompletableFuture<StackTraceSampleResponse> canceldSampleFuture = + tmGateway.requestStackTraceSample(eid, sampleId4, numSamples, Time.milliseconds(10L), maxDepth, timeout); + + Thread.sleep(sleepTime); + + tmGateway.cancelTask(eid, timeout); + taskCanceledFuture.get(); + + StackTraceSampleResponse responseAfterCancel = canceldSampleFuture.get(); + + assertEquals(eid, responseAfterCancel.getExecutionAttemptID()); + assertEquals(sampleId4, responseAfterCancel.getSampleId()); + } + } + + private TaskDeploymentDescriptor createTestTaskDeploymentDescriptor( + String taskName, + ExecutionAttemptID eid, + Class<? extends AbstractInvokable> abstractInvokable + ) throws IOException { + return createTestTaskDeploymentDescriptor(taskName, eid, abstractInvokable, 1); + } + + private TaskDeploymentDescriptor createTestTaskDeploymentDescriptor( + String taskName, + ExecutionAttemptID eid, + Class<? extends AbstractInvokable> abstractInvokable, + int maxNumberOfSubtasks + ) throws IOException { + return createTestTaskDeploymentDescriptor(taskName, + eid, + abstractInvokable, + maxNumberOfSubtasks, + Collections.emptyList(), + Collections.emptyList()); + } + + private TaskDeploymentDescriptor createTestTaskDeploymentDescriptor( + String taskName, + ExecutionAttemptID eid, + Class<? extends AbstractInvokable> abstractInvokable, + int maxNumberOfSubtasks, + Collection<ResultPartitionDeploymentDescriptor> producedPartitions, + Collection<InputGateDeploymentDescriptor> inputGates + ) throws IOException { + Preconditions.checkNotNull(producedPartitions); + Preconditions.checkNotNull(inputGates); + return createTaskDeploymentDescriptor( + jobId, testName.getMethodName(), eid, + new SerializedValue<>(new ExecutionConfig()), taskName, maxNumberOfSubtasks, 0, 1, 0, + new Configuration(), new Configuration(), abstractInvokable.getName(), + producedPartitions, + inputGates, + Collections.emptyList(), + Collections.emptyList(), + 0); + } + + private static TaskDeploymentDescriptor createTaskDeploymentDescriptor( + JobID jobId, + String jobName, + ExecutionAttemptID executionAttemptId, + SerializedValue<ExecutionConfig> serializedExecutionConfig, + String taskName, + int maxNumberOfSubtasks, + int subtaskIndex, + int numberOfSubtasks, + int attemptNumber, + Configuration jobConfiguration, + Configuration taskConfiguration, + String invokableClassName, + Collection<ResultPartitionDeploymentDescriptor> producedPartitions, + Collection<InputGateDeploymentDescriptor> inputGates, + Collection<PermanentBlobKey> requiredJarFiles, + Collection<URL> requiredClasspaths, + int targetSlotNumber) throws IOException { + + JobInformation jobInformation = new JobInformation( + jobId, + jobName, + serializedExecutionConfig, + jobConfiguration, + requiredJarFiles, + requiredClasspaths); + + TaskInformation taskInformation = new TaskInformation( + new JobVertexID(), + taskName, + numberOfSubtasks, + maxNumberOfSubtasks, + invokableClassName, + taskConfiguration); + + SerializedValue<JobInformation> serializedJobInformation = new SerializedValue<>(jobInformation); + SerializedValue<TaskInformation> serializedJobVertexInformation = new SerializedValue<>(taskInformation); + + return new TaskDeploymentDescriptor( + jobId, + new TaskDeploymentDescriptor.NonOffloaded<>(serializedJobInformation), + new TaskDeploymentDescriptor.NonOffloaded<>(serializedJobVertexInformation), + executionAttemptId, + new AllocationID(), + subtaskIndex, + attemptNumber, + targetSlotNumber, + null, + producedPartitions, + inputGates); + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java index 6087cf1..c7ae450 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java @@ -23,11 +23,13 @@ import org.apache.flink.api.common.JobID; import org.apache.flink.api.common.time.Time; import org.apache.flink.api.java.tuple.Tuple3; import org.apache.flink.api.java.tuple.Tuple4; +import org.apache.flink.configuration.ConfigConstants; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.TaskManagerOptions; import org.apache.flink.core.memory.MemoryType; import org.apache.flink.core.testutils.OneShotLatch; import org.apache.flink.runtime.blob.BlobCacheService; +import org.apache.flink.runtime.blob.TransientBlobKey; import org.apache.flink.runtime.blob.VoidBlobStore; import org.apache.flink.runtime.clusterframework.types.AllocationID; import org.apache.flink.runtime.clusterframework.types.ResourceID; @@ -97,6 +99,7 @@ import org.apache.flink.runtime.testtasks.NoOpInvokable; import org.apache.flink.runtime.util.TestingFatalErrorHandler; import org.apache.flink.util.ExceptionUtils; import org.apache.flink.util.FlinkException; +import org.apache.flink.util.NetUtils; import org.apache.flink.util.SerializedValue; import org.apache.flink.util.TestLogger; import org.apache.flink.util.function.FunctionUtils; @@ -134,11 +137,13 @@ import java.util.function.Function; import java.util.stream.Collectors; import static org.hamcrest.Matchers.containsInAnyOrder; +import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.empty; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.notNullValue; +import static org.hamcrest.Matchers.startsWith; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; @@ -1754,6 +1759,45 @@ public class TaskExecutorTest extends TestLogger { } } + @Test(timeout = 10000L) + public void testLogNotFoundHandling() throws Throwable { + final int dataPort = NetUtils.getAvailablePort(); + Configuration config = new Configuration(); + config.setInteger(TaskManagerOptions.DATA_PORT, dataPort); + config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100); + config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_MAX, 200); + config.setString(ConfigConstants.TASK_MANAGER_LOG_PATH_KEY, "/i/dont/exist"); + + try (TaskSubmissionTestEnvironment env = + new TaskSubmissionTestEnvironment.Builder(jobId) + .setConfiguration(config) + .setLocalCommunication(false) + .build()) { + TaskExecutorGateway tmGateway = env.getTaskExecutorGateway(); + try { + CompletableFuture<TransientBlobKey> logFuture = + tmGateway.requestFileUpload(FileType.LOG, timeout); + logFuture.get(); + } catch (Exception e) { + assertThat(e.getMessage(), containsString("The file LOG does not exist on the TaskExecutor.")); + } + } + } + + @Test(timeout = 10000L) + public void testTerminationOnFatalError() throws Throwable { + try (TaskSubmissionTestEnvironment env = new TaskSubmissionTestEnvironment.Builder(jobId).build()) { + String testExceptionMsg = "Test exception of fatal error."; + + env.getTaskExecutor().onFatalError(new Exception(testExceptionMsg)); + + Throwable exception = env.getTestingFatalErrorHandler().getErrorFuture().get(); + env.getTestingFatalErrorHandler().clearError(); + + assertThat(exception.getMessage(), startsWith(testExceptionMsg)); + } + } + private TaskExecutorLocalStateStoresManager createTaskExecutorLocalStateStoresManager() throws IOException { return new TaskExecutorLocalStateStoresManager( false, diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfigurationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfigurationTest.java index 01d4dd7..1deed3f 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfigurationTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerServicesConfigurationTest.java @@ -24,6 +24,8 @@ import org.apache.flink.util.TestLogger; import org.junit.Test; +import java.net.InetAddress; + import static org.junit.Assert.*; /** @@ -85,6 +87,33 @@ public class TaskManagerServicesConfigurationTest extends TestLogger { /** * Verifies that {@link TaskManagerServicesConfiguration#hasNewNetworkBufConf(Configuration)} + * returns the correct result for new configurations via + * {@link TaskManagerOptions#NETWORK_REQUEST_BACKOFF_INITIAL}, + * {@link TaskManagerOptions#NETWORK_REQUEST_BACKOFF_MAX}, + * {@link TaskManagerOptions#NETWORK_BUFFERS_PER_CHANNEL} and + * {@link TaskManagerOptions#NETWORK_EXTRA_BUFFERS_PER_GATE} + */ + @Test + public void testNetworkRequestBackoffAndBuffers() throws Exception { + + // set some non-default values + final Configuration config = new Configuration(); + config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100); + config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_MAX, 200); + config.setInteger(TaskManagerOptions.NETWORK_BUFFERS_PER_CHANNEL, 10); + config.setInteger(TaskManagerOptions.NETWORK_EXTRA_BUFFERS_PER_GATE, 100); + + TaskManagerServicesConfiguration tmConfig = + TaskManagerServicesConfiguration.fromConfiguration(config, InetAddress.getLoopbackAddress(), true); + + assertEquals(tmConfig.getNetworkConfig().partitionRequestInitialBackoff(), 100); + assertEquals(tmConfig.getNetworkConfig().partitionRequestMaxBackoff(), 200); + assertEquals(tmConfig.getNetworkConfig().networkBuffersPerChannel(), 10); + assertEquals(tmConfig.getNetworkConfig().floatingNetworkBuffersPerGate(), 100); + } + + /** + * Verifies that {@link TaskManagerServicesConfiguration#hasNewNetworkBufConf(Configuration)} * returns the correct result for mixed old/new configurations. */ @SuppressWarnings("deprecation") diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskSubmissionTestEnvironment.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskSubmissionTestEnvironment.java new file mode 100644 index 0000000..eb106c9 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskSubmissionTestEnvironment.java @@ -0,0 +1,363 @@ +/* + * 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.runtime.taskexecutor; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.api.common.time.Time; +import org.apache.flink.api.java.tuple.Tuple3; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.TaskManagerOptions; +import org.apache.flink.runtime.blob.BlobCacheService; +import org.apache.flink.runtime.blob.VoidBlobStore; +import org.apache.flink.runtime.clusterframework.types.AllocationID; +import org.apache.flink.runtime.clusterframework.types.ResourceID; +import org.apache.flink.runtime.clusterframework.types.ResourceProfile; +import org.apache.flink.runtime.concurrent.Executors; +import org.apache.flink.runtime.execution.ExecutionState; +import org.apache.flink.runtime.execution.librarycache.LibraryCacheManager; +import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; +import org.apache.flink.runtime.heartbeat.HeartbeatServices; +import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices; +import org.apache.flink.runtime.io.network.ConnectionManager; +import org.apache.flink.runtime.io.network.LocalConnectionManager; +import org.apache.flink.runtime.io.network.NetworkEnvironment; +import org.apache.flink.runtime.io.network.TaskEventDispatcher; +import org.apache.flink.runtime.io.network.buffer.NetworkBufferPool; +import org.apache.flink.runtime.io.network.netty.NettyConfig; +import org.apache.flink.runtime.io.network.netty.NettyConnectionManager; +import org.apache.flink.runtime.io.network.netty.PartitionProducerStateChecker; +import org.apache.flink.runtime.io.network.partition.ResultPartitionManager; +import org.apache.flink.runtime.jobmaster.JobMasterGateway; +import org.apache.flink.runtime.jobmaster.JobMasterId; +import org.apache.flink.runtime.jobmaster.utils.TestingJobMasterGateway; +import org.apache.flink.runtime.jobmaster.utils.TestingJobMasterGatewayBuilder; +import org.apache.flink.runtime.leaderretrieval.SettableLeaderRetrievalService; +import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups; +import org.apache.flink.runtime.rpc.RpcService; +import org.apache.flink.runtime.rpc.RpcUtils; +import org.apache.flink.runtime.rpc.TestingRpcService; +import org.apache.flink.runtime.state.TaskExecutorLocalStateStoresManager; +import org.apache.flink.runtime.taskexecutor.rpc.RpcResultPartitionConsumableNotifier; +import org.apache.flink.runtime.taskexecutor.slot.TaskSlotTable; +import org.apache.flink.runtime.taskexecutor.slot.TimerService; +import org.apache.flink.runtime.taskmanager.CheckpointResponder; +import org.apache.flink.runtime.taskmanager.NoOpTaskManagerActions; +import org.apache.flink.runtime.taskmanager.Task; +import org.apache.flink.runtime.taskmanager.TaskManagerActions; +import org.apache.flink.runtime.testingUtils.TestingUtils; +import org.apache.flink.runtime.util.TestingFatalErrorHandler; +import org.junit.rules.TemporaryFolder; +import org.mockito.Mockito; + +import javax.annotation.Nonnull; +import java.io.File; +import java.net.InetAddress; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.CompletableFuture; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Simple environment setup for task executor task. + */ +class TaskSubmissionTestEnvironment implements AutoCloseable { + + private final HeartbeatServices heartbeatServices = new HeartbeatServices(1000L, 1000L); + private final TestingRpcService testingRpcService = new TestingRpcService(); + private final BlobCacheService blobCacheService= new BlobCacheService(new Configuration(), new VoidBlobStore(), null); + private final Time timeout = Time.milliseconds(10000L); + private final TestingFatalErrorHandler testingFatalErrorHandler = new TestingFatalErrorHandler(); + private final TimerService<AllocationID> timerService = new TimerService<>(TestingUtils.defaultExecutor(), timeout.toMilliseconds()); + + private final TestingHighAvailabilityServices haServices; + private final TemporaryFolder temporaryFolder; + private final TaskSlotTable taskSlotTable; + private final JobMasterId jobMasterId; + + private TestingTaskExecutor taskExecutor; + + private TaskSubmissionTestEnvironment( + JobID jobId, + JobMasterId jobMasterId, + int slotSize, + boolean mockNetworkEnvironment, + TestingJobMasterGateway testingJobMasterGateway, + Configuration configuration, + boolean localCommunication, + List<Tuple3<ExecutionAttemptID, ExecutionState, CompletableFuture<Void>>> taskManagerActionListeners) throws Exception { + + this.haServices = new TestingHighAvailabilityServices(); + this.haServices.setResourceManagerLeaderRetriever(new SettableLeaderRetrievalService()); + this.haServices.setJobMasterLeaderRetriever(jobId, new SettableLeaderRetrievalService()); + + this.temporaryFolder = new TemporaryFolder(); + this.temporaryFolder.create(); + + this.jobMasterId = jobMasterId; + + if (slotSize > 0) { + this.taskSlotTable = generateTaskSlotTable(slotSize); + } else { + this.taskSlotTable = mock(TaskSlotTable.class); + when(taskSlotTable.tryMarkSlotActive(eq(jobId), any())).thenReturn(true); + when(taskSlotTable.addTask(any(Task.class))).thenReturn(true); + } + + JobMasterGateway jobMasterGateway; + if (testingJobMasterGateway == null) { + jobMasterGateway = new TestingJobMasterGatewayBuilder() + .setFencingTokenSupplier(() -> jobMasterId) + .build(); + } else { + jobMasterGateway = testingJobMasterGateway; + } + + TaskManagerActions taskManagerActions; + if (taskManagerActionListeners.size() == 0) { + taskManagerActions = new NoOpTaskManagerActions(); + } else { + TestTaskManagerActions testTaskManagerActions = new TestTaskManagerActions(taskSlotTable, jobMasterGateway); + for (Tuple3<ExecutionAttemptID, ExecutionState, CompletableFuture<Void>> listenerTuple : taskManagerActionListeners) { + testTaskManagerActions.addListener(listenerTuple.f0, listenerTuple.f1, listenerTuple.f2); + } + taskManagerActions = testTaskManagerActions; + } + + final NetworkEnvironment networkEnvironment = createNetworkEnvironment(localCommunication, configuration, testingRpcService, mockNetworkEnvironment); + + final JobManagerConnection jobManagerConnection = createJobManagerConnection(jobId, jobMasterGateway, testingRpcService, taskManagerActions, timeout); + final JobManagerTable jobManagerTable = new JobManagerTable(); + jobManagerTable.put(jobId, jobManagerConnection); + + TaskExecutorLocalStateStoresManager localStateStoresManager = new TaskExecutorLocalStateStoresManager( + false, + new File[]{temporaryFolder.newFolder()}, + Executors.directExecutor()); + + final TaskManagerServices taskManagerServices = new TaskManagerServicesBuilder() + .setNetworkEnvironment(networkEnvironment) + .setTaskSlotTable(taskSlotTable) + .setJobManagerTable(jobManagerTable) + .setTaskStateManager(localStateStoresManager) + .build(); + + taskExecutor = createTaskExecutor(taskManagerServices, configuration); + + taskExecutor.start(); + taskExecutor.waitUntilStarted(); + } + + public TestingTaskExecutor getTaskExecutor() { + return taskExecutor; + } + + public TaskExecutorGateway getTaskExecutorGateway() { + return taskExecutor.getSelfGateway(TaskExecutorGateway.class); + } + + public TaskSlotTable getTaskSlotTable() { + return taskSlotTable; + } + + public JobMasterId getJobMasterId() { + return jobMasterId; + } + + public TestingFatalErrorHandler getTestingFatalErrorHandler() { + return testingFatalErrorHandler; + } + + private TaskSlotTable generateTaskSlotTable(int numSlot) { + Collection<ResourceProfile> resourceProfiles = new ArrayList<>(); + for (int i = 0; i < numSlot; i++) { + resourceProfiles.add(ResourceProfile.UNKNOWN); + } + return new TaskSlotTable(resourceProfiles, timerService); + } + + @Nonnull + private TestingTaskExecutor createTaskExecutor(TaskManagerServices taskManagerServices, Configuration configuration) { + return new TestingTaskExecutor( + testingRpcService, + TaskManagerConfiguration.fromConfiguration(configuration), + haServices, + taskManagerServices, + heartbeatServices, + UnregisteredMetricGroups.createUnregisteredTaskManagerMetricGroup(), + null, + blobCacheService, + testingFatalErrorHandler + ); + } + + private static JobManagerConnection createJobManagerConnection(JobID jobId, JobMasterGateway jobMasterGateway, RpcService testingRpcService, TaskManagerActions taskManagerActions, Time timeout) { + final LibraryCacheManager libraryCacheManager = mock(LibraryCacheManager.class); + when(libraryCacheManager.getClassLoader(any(JobID.class))).thenReturn(ClassLoader.getSystemClassLoader()); + + final PartitionProducerStateChecker partitionProducerStateChecker = mock(PartitionProducerStateChecker.class); + when(partitionProducerStateChecker.requestPartitionProducerState(any(), any(), any())) + .thenReturn(CompletableFuture.completedFuture(ExecutionState.RUNNING)); + + return new JobManagerConnection( + jobId, + ResourceID.generate(), + jobMasterGateway, + taskManagerActions, + mock(CheckpointResponder.class), + new TestGlobalAggregateManager(), + libraryCacheManager, + new RpcResultPartitionConsumableNotifier(jobMasterGateway, testingRpcService.getExecutor(), timeout), + partitionProducerStateChecker); + } + + private static NetworkEnvironment createNetworkEnvironment(boolean localCommunication, Configuration configuration, RpcService testingRpcService, boolean mockNetworkEnvironment) throws Exception { + final ConnectionManager connectionManager; + if (!localCommunication) { + NettyConfig nettyConfig = TaskManagerServicesConfiguration + .fromConfiguration(configuration, InetAddress.getByName(testingRpcService.getAddress()), localCommunication).getNetworkConfig() + .nettyConfig(); + connectionManager = new NettyConnectionManager(nettyConfig); + } else { + connectionManager = new LocalConnectionManager(); + } + + final int numAllBuffers = 10; + final NetworkEnvironment networkEnvironment; + if (mockNetworkEnvironment) { + networkEnvironment = mock(NetworkEnvironment.class, Mockito.RETURNS_MOCKS); + } else { + networkEnvironment = createNetworkEnvironment( + numAllBuffers, + 128, + configuration.getInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_INITIAL), + configuration.getInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_MAX), + 2, + 8, + true, + connectionManager); + networkEnvironment.start(); + } + + return networkEnvironment; + } + + @Nonnull + private static NetworkEnvironment createNetworkEnvironment( + int numBuffers, + int memorySegmentSize, + int partitionRequestInitialBackoff, + int partitionRequestMaxBackoff, + int networkBuffersPerChannel, + int extraNetworkBuffersPerGate, + boolean enableCreditBased, + ConnectionManager connectionManager) { + return new NetworkEnvironment( + new NetworkBufferPool(numBuffers, memorySegmentSize), + connectionManager, + new ResultPartitionManager(), + new TaskEventDispatcher(), + partitionRequestInitialBackoff, + partitionRequestMaxBackoff, + networkBuffersPerChannel, + extraNetworkBuffersPerGate, + enableCreditBased); + } + + @Override + public void close() throws Exception { + RpcUtils.terminateRpcEndpoint(taskExecutor, timeout); + + timerService.stop(); + + blobCacheService.close(); + + temporaryFolder.delete(); + + testingFatalErrorHandler.rethrowError(); + } + + public static final class Builder { + + private JobID jobId; + private boolean mockNetworkEnvironment = true; + private int slotSize; + private JobMasterId jobMasterId = JobMasterId.generate(); + private TestingJobMasterGateway jobMasterGateway; + private boolean localCommunication = true; + private Configuration configuration = new Configuration(); + + private List<Tuple3<ExecutionAttemptID, ExecutionState, CompletableFuture<Void>>> taskManagerActionListeners = new ArrayList<>(); + + public Builder(JobID jobId) { + this.jobId = jobId; + } + + public Builder setMockNetworkEnvironment(boolean mockNetworkEnvironment) { + this.mockNetworkEnvironment = mockNetworkEnvironment; + return this; + } + + public Builder setSlotSize(int slotSize) { + this.slotSize = slotSize; + return this; + } + + public Builder setJobMasterId(JobMasterId jobMasterId) { + this.jobMasterId = jobMasterId; + return this; + } + + public Builder setJobMasterGateway(TestingJobMasterGateway jobMasterGateway) { + this.jobMasterGateway = jobMasterGateway; + return this; + } + + public Builder setLocalCommunication(boolean localCommunication) { + this.localCommunication = localCommunication; + return this; + } + + public Builder setConfiguration(Configuration configuration) { + this.configuration = configuration; + return this; + } + + public Builder addTaskManagerActionListener(ExecutionAttemptID eid, ExecutionState executionState, CompletableFuture<Void> future) { + taskManagerActionListeners.add(Tuple3.of(eid, executionState, future)); + return this; + } + + public TaskSubmissionTestEnvironment build() throws Exception { + return new TaskSubmissionTestEnvironment( + jobId, + jobMasterId, + slotSize, + mockNetworkEnvironment, + jobMasterGateway, + configuration, + localCommunication, + taskManagerActionListeners); + } + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestTaskManagerActions.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestTaskManagerActions.java new file mode 100644 index 0000000..de36e23 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestTaskManagerActions.java @@ -0,0 +1,113 @@ +/* + * 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.runtime.taskexecutor; + +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.runtime.execution.ExecutionState; +import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; +import org.apache.flink.runtime.jobmaster.JobMasterGateway; +import org.apache.flink.runtime.messages.Acknowledge; +import org.apache.flink.runtime.taskexecutor.slot.TaskSlotTable; +import org.apache.flink.runtime.taskmanager.TaskExecutionState; +import org.apache.flink.runtime.taskmanager.TaskManagerActions; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; + +/** + * Customized {@link TaskManagerActions} that wait for ExecutionState changes. + */ +public class TestTaskManagerActions implements TaskManagerActions { + + private final JobMasterGateway jobMasterGateway; + private final TaskSlotTable taskSlotTable; + private final TaskManagerActionListeners taskManagerActionListeners = new TaskManagerActionListeners(); + + public TestTaskManagerActions(TaskSlotTable taskSlotTable, JobMasterGateway jobMasterGateway) { + this.taskSlotTable = taskSlotTable; + this.jobMasterGateway = jobMasterGateway; + } + + public void addListener(ExecutionAttemptID eid, ExecutionState executionState, CompletableFuture<Void> future) { + taskManagerActionListeners.addListener(eid, executionState, future); + } + + @Override + public void notifyFatalError(String message, Throwable cause) { + } + + @Override + public void failTask(ExecutionAttemptID executionAttemptID, Throwable cause) { + if (taskSlotTable != null) { + taskSlotTable.getTask(executionAttemptID).failExternally(cause); + } + } + + @Override + public void updateTaskExecutionState(TaskExecutionState taskExecutionState) { + Optional<CompletableFuture<Void>> listenerFuture = + taskManagerActionListeners.getListenerFuture(taskExecutionState.getID(), taskExecutionState.getExecutionState()); + if (listenerFuture.isPresent()) { + listenerFuture.get().complete(null); + } + if (jobMasterGateway != null) { + CompletableFuture<Acknowledge> futureAcknowledge = jobMasterGateway.updateTaskExecutionState(taskExecutionState); + + futureAcknowledge.whenComplete( + (ack, throwable) -> { + if (throwable != null) { + failTask(taskExecutionState.getID(), throwable); + } + } + ); + } + } + + private static class TaskManagerActionListeners { + private final Map<ExecutionAttemptID, List<Tuple2<ExecutionState, CompletableFuture<Void>>>> expectExecutionStateAndFutures; + + private TaskManagerActionListeners() { + this.expectExecutionStateAndFutures = new HashMap<>(); + } + + private void addListener(ExecutionAttemptID eid, ExecutionState executionState, CompletableFuture<Void> future) { + final List<Tuple2<ExecutionState, CompletableFuture<Void>>> expectExecutionStateAndFutureList = expectExecutionStateAndFutures + .getOrDefault(eid, new ArrayList<>()); + expectExecutionStateAndFutureList.add(Tuple2.of(executionState, future)); + expectExecutionStateAndFutures.put(eid, expectExecutionStateAndFutureList); + } + + private Optional<CompletableFuture<Void>> getListenerFuture(ExecutionAttemptID eid, ExecutionState executionState) { + List<Tuple2<ExecutionState, CompletableFuture<Void>>> expectStateAndFutureList = expectExecutionStateAndFutures.get(eid); + if (expectExecutionStateAndFutures == null) { + return Optional.empty(); + } + for (Tuple2<ExecutionState, CompletableFuture<Void>> expectStateAndFuture : expectStateAndFutureList) { + if (expectStateAndFuture.f0 == executionState) { + return Optional.of(expectStateAndFuture.f1); + } + } + return Optional.empty(); + } + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java deleted file mode 100644 index 937fb1e..0000000 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerTest.java +++ /dev/null @@ -1,2185 +0,0 @@ -/* - * 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.runtime.taskmanager; - -import org.apache.flink.api.common.ExecutionConfig; -import org.apache.flink.api.common.JobID; -import org.apache.flink.api.common.time.Time; -import org.apache.flink.configuration.ConfigConstants; -import org.apache.flink.configuration.Configuration; -import org.apache.flink.configuration.TaskManagerOptions; -import org.apache.flink.runtime.akka.AkkaUtils; -import org.apache.flink.runtime.akka.FlinkUntypedActor; -import org.apache.flink.runtime.blob.PermanentBlobKey; -import org.apache.flink.runtime.clusterframework.types.AllocationID; -import org.apache.flink.runtime.deployment.InputChannelDeploymentDescriptor; -import org.apache.flink.runtime.deployment.InputGateDeploymentDescriptor; -import org.apache.flink.runtime.deployment.ResultPartitionDeploymentDescriptor; -import org.apache.flink.runtime.deployment.ResultPartitionLocation; -import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor; -import org.apache.flink.runtime.execution.Environment; -import org.apache.flink.runtime.execution.ExecutionState; -import org.apache.flink.runtime.executiongraph.ExecutionAttemptID; -import org.apache.flink.runtime.executiongraph.JobInformation; -import org.apache.flink.runtime.executiongraph.TaskInformation; -import org.apache.flink.runtime.highavailability.HighAvailabilityServices; -import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices; -import org.apache.flink.runtime.instance.ActorGateway; -import org.apache.flink.runtime.instance.AkkaActorGateway; -import org.apache.flink.runtime.instance.InstanceID; -import org.apache.flink.runtime.io.network.ConnectionID; -import org.apache.flink.runtime.io.network.api.writer.RecordWriter; -import org.apache.flink.runtime.io.network.api.writer.RecordWriterBuilder; -import org.apache.flink.runtime.io.network.partition.PartitionNotFoundException; -import org.apache.flink.runtime.io.network.partition.ResultPartitionID; -import org.apache.flink.runtime.io.network.partition.ResultPartitionType; -import org.apache.flink.runtime.jobgraph.IntermediateDataSetID; -import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID; -import org.apache.flink.runtime.jobgraph.JobVertexID; -import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable; -import org.apache.flink.runtime.jobmanager.Tasks; -import org.apache.flink.runtime.jobmaster.TestingAbstractInvokables; -import org.apache.flink.runtime.leaderretrieval.SettableLeaderRetrievalService; -import org.apache.flink.runtime.leaderretrieval.StandaloneLeaderRetrievalService; -import org.apache.flink.runtime.messages.Acknowledge; -import org.apache.flink.runtime.messages.RegistrationMessages; -import org.apache.flink.runtime.messages.StackTraceSampleMessages.TriggerStackTraceSample; -import org.apache.flink.runtime.messages.StackTraceSampleResponse; -import org.apache.flink.runtime.messages.TaskManagerMessages; -import org.apache.flink.runtime.messages.TaskManagerMessages.FatalError; -import org.apache.flink.runtime.messages.TaskMessages; -import org.apache.flink.runtime.messages.TaskMessages.CancelTask; -import org.apache.flink.runtime.messages.TaskMessages.StopTask; -import org.apache.flink.runtime.messages.TaskMessages.SubmitTask; -import org.apache.flink.runtime.taskexecutor.TaskManagerServicesConfiguration; -import org.apache.flink.runtime.testingUtils.TestingTaskManagerMessages; -import org.apache.flink.runtime.testingUtils.TestingUtils; -import org.apache.flink.runtime.testtasks.BlockingNoOpInvokable; -import org.apache.flink.runtime.testutils.StoppableInvokable; -import org.apache.flink.types.IntValue; -import org.apache.flink.util.NetUtils; -import org.apache.flink.util.SerializedValue; -import org.apache.flink.util.TestLogger; - -import akka.actor.ActorRef; -import akka.actor.ActorSystem; -import akka.actor.Kill; -import akka.actor.Props; -import akka.actor.Status; -import akka.japi.Creator; -import akka.testkit.JavaTestKit; -import org.junit.After; -import org.junit.AfterClass; -import org.junit.Assert; -import org.junit.Before; -import org.junit.BeforeClass; -import org.junit.Test; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.net.InetAddress; -import java.net.InetSocketAddress; -import java.net.URL; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collection; -import java.util.Collections; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.UUID; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.TimeUnit; - -import scala.Option; -import scala.concurrent.Await; -import scala.concurrent.Future; -import scala.concurrent.duration.FiniteDuration; -import scala.util.Failure; - -import static org.apache.flink.runtime.messages.JobManagerMessages.RequestPartitionProducerState; -import static org.apache.flink.runtime.messages.JobManagerMessages.ScheduleOrUpdateConsumers; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; - -/** - * Tests for the legacy {@link TaskManager}. - */ -@SuppressWarnings("serial") -public class TaskManagerTest extends TestLogger { - - private static final Logger LOG = LoggerFactory.getLogger(TaskManagerTest.class); - - private static final FiniteDuration timeout = new FiniteDuration(1, TimeUnit.MINUTES); - - private static final FiniteDuration d = new FiniteDuration(60, TimeUnit.SECONDS); - private static final Time timeD = Time.seconds(60L); - - private static ActorSystem system; - - static final UUID LEADER_SESSION_ID = UUID.randomUUID(); - - private TestingHighAvailabilityServices highAvailabilityServices; - - @BeforeClass - public static void setup() { - system = AkkaUtils.createLocalActorSystem(new Configuration()); - } - - @AfterClass - public static void teardown() { - JavaTestKit.shutdownActorSystem(system); - } - - @Before - public void setupTest() { - highAvailabilityServices = new TestingHighAvailabilityServices(); - } - - @After - public void tearDownTest() throws Exception { - if (highAvailabilityServices != null) { - highAvailabilityServices.closeAndCleanupAllData(); - - highAvailabilityServices = null; - } - } - - @Test - public void testSubmitAndExecuteTask() throws IOException { - new JavaTestKit(system){{ - - ActorGateway taskManager = null; - final ActorGateway jobManager = TestingUtils.createForwardingActor( - system, - getTestActor(), - HighAvailabilityServices.DEFAULT_LEADER_ID, - Option.<String>empty()); - - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new StandaloneLeaderRetrievalService(jobManager.path(), jobManager.leaderSessionID())); - - try { - taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, - new Configuration(), - true, - false); - - final ActorGateway tm = taskManager; - - // handle the registration - new Within(d) { - @Override - protected void run() { - expectMsgClass(RegistrationMessages.RegisterTaskManager.class); - - final InstanceID iid = new InstanceID(); - assertEquals(tm.actor(), getLastSender()); - tm.tell( - new RegistrationMessages.AcknowledgeRegistration( - iid, - 12345), - jobManager); - } - }; - - final JobID jid = new JobID(); - final JobVertexID vid = new JobVertexID(); - final ExecutionAttemptID eid = new ExecutionAttemptID(); - final SerializedValue<ExecutionConfig> executionConfig = new SerializedValue<>(new ExecutionConfig()); - - final TaskDeploymentDescriptor tdd = createTaskDeploymentDescriptor( - jid, "TestJob", vid, eid, executionConfig, - "TestTask", 7, 2, 7, 0, new Configuration(), new Configuration(), - TestInvokableCorrect.class.getName(), - Collections.<ResultPartitionDeploymentDescriptor>emptyList(), - Collections.<InputGateDeploymentDescriptor>emptyList(), - new ArrayList<PermanentBlobKey>(), Collections.emptyList(), 0); - - new Within(d) { - - @Override - protected void run() { - tm.tell(new SubmitTask(tdd), jobManager); - - // TaskManager should acknowledge the submission - // heartbeats may be interleaved - long deadline = System.currentTimeMillis() + 10000; - do { - Object message = receiveOne(d); - if (message.equals(Acknowledge.get())) { - break; - } - } while (System.currentTimeMillis() < deadline); - - // task should have switched to running - Object toRunning = new TaskMessages.UpdateTaskExecutionState( - new TaskExecutionState(jid, eid, ExecutionState.RUNNING)); - - // task should have switched to finished - Object toFinished = new TaskMessages.UpdateTaskExecutionState( - new TaskExecutionState(jid, eid, ExecutionState.FINISHED)); - - deadline = System.currentTimeMillis() + 10000; - do { - Object message = receiveOne(d); - if (message.equals(toRunning)) { - break; - } - else if (!(message instanceof TaskManagerMessages.Heartbeat)) { - fail("Unexpected message: " + message); - } - } while (System.currentTimeMillis() < deadline); - - deadline = System.currentTimeMillis() + 10000; - do { - Object message = receiveOne(d); - if (message.equals(toFinished)) { - break; - } - else if (!(message instanceof TaskManagerMessages.Heartbeat)) { - fail("Unexpected message: " + message); - } - } while (System.currentTimeMillis() < deadline); - } - }; - } - finally { - // shut down the actors - TestingUtils.stopActor(taskManager); - TestingUtils.stopActor(jobManager); - } - }}; - } - - @Test - public void testJobSubmissionAndCanceling() { - new JavaTestKit(system){{ - - ActorGateway jobManager = null; - ActorGateway taskManager = null; - - final ActorGateway testActorGateway = new AkkaActorGateway( - getTestActor(), - LEADER_SESSION_ID); - - try { - ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, LEADER_SESSION_ID)); - jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); - - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new StandaloneLeaderRetrievalService(jobManager.path(), jobManager.leaderSessionID())); - - taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, - new Configuration(), - true, - true); - - final JobID jid1 = new JobID(); - final JobID jid2 = new JobID(); - - JobVertexID vid1 = new JobVertexID(); - JobVertexID vid2 = new JobVertexID(); - - final ExecutionAttemptID eid1 = new ExecutionAttemptID(); - final ExecutionAttemptID eid2 = new ExecutionAttemptID(); - - final TaskDeploymentDescriptor tdd1 = createTaskDeploymentDescriptor( - jid1, "TestJob1", vid1, eid1, - new SerializedValue<>(new ExecutionConfig()), - "TestTask1", 5, 1, 5, 0, - new Configuration(), new Configuration(), TestInvokableBlockingCancelable.class.getName(), - Collections.<ResultPartitionDeploymentDescriptor>emptyList(), - Collections.<InputGateDeploymentDescriptor>emptyList(), - new ArrayList<>(), Collections.<URL>emptyList(), 0); - - final TaskDeploymentDescriptor tdd2 = createTaskDeploymentDescriptor( - jid2, "TestJob2", vid2, eid2, - new SerializedValue<>(new ExecutionConfig()), - "TestTask2", 7, 2, 7, 0, - new Configuration(), new Configuration(), TestInvokableBlockingCancelable.class.getName(), - Collections.<ResultPartitionDeploymentDescriptor>emptyList(), - Collections.<InputGateDeploymentDescriptor>emptyList(), - new ArrayList<>(), Collections.emptyList(), 0); - - final ActorGateway tm = taskManager; - - new Within(d) { - - @Override - protected void run() { - try { - Future<Object> t1Running = tm.ask( - new TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid1), - timeout); - Future<Object> t2Running = tm.ask( - new TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid2), - timeout); - - tm.tell(new SubmitTask(tdd1), testActorGateway); - tm.tell(new SubmitTask(tdd2), testActorGateway); - - expectMsgEquals(Acknowledge.get()); - expectMsgEquals(Acknowledge.get()); - - Await.ready(t1Running, d); - Await.ready(t2Running, d); - - tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), testActorGateway); - - Map<ExecutionAttemptID, Task> runningTasks = expectMsgClass(TestingTaskManagerMessages - .ResponseRunningTasks.class).asJava(); - - assertEquals(2, runningTasks.size()); - Task t1 = runningTasks.get(eid1); - Task t2 = runningTasks.get(eid2); - assertNotNull(t1); - assertNotNull(t2); - - assertEquals(ExecutionState.RUNNING, t1.getExecutionState()); - assertEquals(ExecutionState.RUNNING, t2.getExecutionState()); - - tm.tell(new CancelTask(eid1), testActorGateway); - - expectMsgEquals(Acknowledge.get()); - - Future<Object> response = tm.ask(new TestingTaskManagerMessages.NotifyWhenTaskRemoved(eid1), - timeout); - Await.ready(response, d); - - assertEquals(ExecutionState.CANCELED, t1.getExecutionState()); - - tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), testActorGateway); - runningTasks = expectMsgClass(TestingTaskManagerMessages - .ResponseRunningTasks.class).asJava(); - - assertEquals(1, runningTasks.size()); - - tm.tell(new CancelTask(eid1), testActorGateway); - expectMsgEquals(Acknowledge.get()); - - tm.tell(new CancelTask(eid2), testActorGateway); - expectMsgEquals(Acknowledge.get()); - - response = tm.ask(new TestingTaskManagerMessages.NotifyWhenTaskRemoved(eid2), - timeout); - Await.ready(response, d); - - assertEquals(ExecutionState.CANCELED, t2.getExecutionState()); - - tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), testActorGateway); - runningTasks = expectMsgClass(TestingTaskManagerMessages - .ResponseRunningTasks.class).asJava(); - - assertEquals(0, runningTasks.size()); - } catch (Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - } - }; - } - catch (Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - finally { - TestingUtils.stopActor(taskManager); - TestingUtils.stopActor(jobManager); - } - }}; - } - - @Test - public void testJobSubmissionAndStop() throws Exception { - new JavaTestKit(system){{ - - ActorGateway jobManager = null; - ActorGateway taskManager = null; - - final ActorGateway testActorGateway = new AkkaActorGateway( - getTestActor(), - LEADER_SESSION_ID); - - try { - ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, LEADER_SESSION_ID)); - jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); - - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new StandaloneLeaderRetrievalService(jobManager.path(), jobManager.leaderSessionID())); - - taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, - new Configuration(), - true, - true); - - final JobID jid1 = new JobID(); - final JobID jid2 = new JobID(); - - JobVertexID vid1 = new JobVertexID(); - JobVertexID vid2 = new JobVertexID(); - - final ExecutionAttemptID eid1 = new ExecutionAttemptID(); - final ExecutionAttemptID eid2 = new ExecutionAttemptID(); - - final SerializedValue<ExecutionConfig> executionConfig = new SerializedValue<>(new ExecutionConfig()); - - final TaskDeploymentDescriptor tdd1 = createTaskDeploymentDescriptor(jid1, "TestJob", vid1, eid1, executionConfig, - "TestTask1", 5, 1, 5, 0, new Configuration(), new Configuration(), StoppableInvokable.class.getName(), - Collections.<ResultPartitionDeploymentDescriptor>emptyList(), - Collections.<InputGateDeploymentDescriptor>emptyList(), - new ArrayList<>(), Collections.emptyList(), 0); - - final TaskDeploymentDescriptor tdd2 = createTaskDeploymentDescriptor(jid2, "TestJob", vid2, eid2, executionConfig, - "TestTask2", 7, 2, 7, 0, new Configuration(), new Configuration(), TestInvokableBlockingCancelable.class.getName(), - Collections.<ResultPartitionDeploymentDescriptor>emptyList(), - Collections.<InputGateDeploymentDescriptor>emptyList(), - new ArrayList<>(), Collections.emptyList(), 0); - - final ActorGateway tm = taskManager; - - new Within(d) { - - @Override - protected void run() { - try { - Future<Object> t1Running = tm.ask( - new TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid1), - timeout); - Future<Object> t2Running = tm.ask( - new TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid2), - timeout); - - tm.tell(new SubmitTask(tdd1), testActorGateway); - tm.tell(new SubmitTask(tdd2), testActorGateway); - - expectMsgEquals(Acknowledge.get()); - expectMsgEquals(Acknowledge.get()); - - Await.ready(t1Running, d); - Await.ready(t2Running, d); - - tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), testActorGateway); - - Map<ExecutionAttemptID, Task> runningTasks = expectMsgClass(TestingTaskManagerMessages - .ResponseRunningTasks.class).asJava(); - - assertEquals(2, runningTasks.size()); - Task t1 = runningTasks.get(eid1); - Task t2 = runningTasks.get(eid2); - assertNotNull(t1); - assertNotNull(t2); - - assertEquals(ExecutionState.RUNNING, t1.getExecutionState()); - assertEquals(ExecutionState.RUNNING, t2.getExecutionState()); - - tm.tell(new StopTask(eid1), testActorGateway); - - expectMsgEquals(Acknowledge.get()); - - Future<Object> response = tm.ask( - new TestingTaskManagerMessages.NotifyWhenTaskRemoved(eid1), - timeout); - Await.ready(response, d); - - assertEquals(ExecutionState.FINISHED, t1.getExecutionState()); - - tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), testActorGateway); - runningTasks = expectMsgClass(TestingTaskManagerMessages - .ResponseRunningTasks.class).asJava(); - - assertEquals(1, runningTasks.size()); - - tm.tell(new StopTask(eid1), testActorGateway); - expectMsgEquals(Acknowledge.get()); - - tm.tell(new StopTask(eid2), testActorGateway); - expectMsgClass(Status.Failure.class); - - assertEquals(ExecutionState.RUNNING, t2.getExecutionState()); - - tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), testActorGateway); - runningTasks = expectMsgClass(TestingTaskManagerMessages - .ResponseRunningTasks.class).asJava(); - - assertEquals(1, runningTasks.size()); - } catch (Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - } - }; - } - finally { - TestingUtils.stopActor(taskManager); - TestingUtils.stopActor(jobManager); - } - }}; - } - - @Test - public void testGateChannelEdgeMismatch() { - new JavaTestKit(system){{ - - ActorGateway jobManager = null; - ActorGateway taskManager = null; - - final ActorGateway testActorGateway = new AkkaActorGateway( - getTestActor(), - LEADER_SESSION_ID); - - try { - ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, LEADER_SESSION_ID)); - jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); - - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new StandaloneLeaderRetrievalService(jobManager.path(), jobManager.leaderSessionID())); - - taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, - new Configuration(), - true, - true); - - final ActorGateway tm = taskManager; - - final JobID jid = new JobID(); - - JobVertexID vid1 = new JobVertexID(); - JobVertexID vid2 = new JobVertexID(); - - final ExecutionAttemptID eid1 = new ExecutionAttemptID(); - final ExecutionAttemptID eid2 = new ExecutionAttemptID(); - - final TaskDeploymentDescriptor tdd1 = createTaskDeploymentDescriptor( - jid, "TestJob", vid1, eid1, - new SerializedValue<>(new ExecutionConfig()), - "Sender", 1, 0, 1, 0, - new Configuration(), new Configuration(), TestingAbstractInvokables.Sender.class.getName(), - Collections.<ResultPartitionDeploymentDescriptor>emptyList(), - Collections.<InputGateDeploymentDescriptor>emptyList(), - new ArrayList<>(), Collections.emptyList(), 0); - - final TaskDeploymentDescriptor tdd2 = createTaskDeploymentDescriptor( - jid, "TestJob", vid2, eid2, - new SerializedValue<>(new ExecutionConfig()), - "Receiver", 7, 2, 7, 0, - new Configuration(), new Configuration(), TestingAbstractInvokables.Receiver.class.getName(), - Collections.<ResultPartitionDeploymentDescriptor>emptyList(), - Collections.<InputGateDeploymentDescriptor>emptyList(), - new ArrayList<>(), Collections.emptyList(), 0); - - new Within(d){ - - @Override - protected void run() { - try { - tm.tell(new SubmitTask(tdd1), testActorGateway); - tm.tell(new SubmitTask(tdd2), testActorGateway); - - expectMsgEquals(Acknowledge.get()); - expectMsgEquals(Acknowledge.get()); - - tm.tell(new TestingTaskManagerMessages.NotifyWhenTaskRemoved(eid1), - testActorGateway); - tm.tell(new TestingTaskManagerMessages.NotifyWhenTaskRemoved(eid2), - testActorGateway); - - expectMsgEquals(true); - expectMsgEquals(true); - - tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), testActorGateway); - Map<ExecutionAttemptID, Task> tasks = expectMsgClass(TestingTaskManagerMessages - .ResponseRunningTasks.class).asJava(); - - assertEquals(0, tasks.size()); - } catch (Exception e){ - e.printStackTrace(); - fail(e.getMessage()); - } - } - }; - } - catch (Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - finally { - // shut down the actors - TestingUtils.stopActor(taskManager); - TestingUtils.stopActor(jobManager); - } - }}; - } - - @Test - public void testRunJobWithForwardChannel() { - new JavaTestKit(system){{ - - ActorGateway jobManager = null; - ActorGateway taskManager = null; - - final ActorGateway testActorGateway = new AkkaActorGateway( - getTestActor(), - LEADER_SESSION_ID); - try { - final JobID jid = new JobID(); - - JobVertexID vid1 = new JobVertexID(); - JobVertexID vid2 = new JobVertexID(); - - final ExecutionAttemptID eid1 = new ExecutionAttemptID(); - final ExecutionAttemptID eid2 = new ExecutionAttemptID(); - - ActorRef jm = system.actorOf(Props.create(new SimpleLookupJobManagerCreator(LEADER_SESSION_ID))); - jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); - - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new StandaloneLeaderRetrievalService(jobManager.path(), jobManager.leaderSessionID())); - - taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, - new Configuration(), - true, - true); - - final ActorGateway tm = taskManager; - - IntermediateResultPartitionID partitionId = new IntermediateResultPartitionID(); - - List<ResultPartitionDeploymentDescriptor> irpdd = new ArrayList<ResultPartitionDeploymentDescriptor>(); - irpdd.add(new ResultPartitionDeploymentDescriptor(new IntermediateDataSetID(), partitionId, ResultPartitionType.PIPELINED, 1, 1, true)); - - InputGateDeploymentDescriptor ircdd = - new InputGateDeploymentDescriptor( - new IntermediateDataSetID(), ResultPartitionType.PIPELINED, - 0, new InputChannelDeploymentDescriptor[]{ - new InputChannelDeploymentDescriptor(new ResultPartitionID(partitionId, eid1), ResultPartitionLocation.createLocal()) - } - ); - - final TaskDeploymentDescriptor tdd1 = createTaskDeploymentDescriptor( - jid, "TestJob", vid1, eid1, - new SerializedValue<>(new ExecutionConfig()), - "Sender", 1, 0, 1, 0, - new Configuration(), new Configuration(), TestingAbstractInvokables.Sender.class.getName(), - irpdd, Collections.<InputGateDeploymentDescriptor>emptyList(), new ArrayList<>(), - Collections.emptyList(), 0); - - final TaskDeploymentDescriptor tdd2 = createTaskDeploymentDescriptor( - jid, "TestJob", vid2, eid2, - new SerializedValue<>(new ExecutionConfig()), - "Receiver", 7, 2, 7, 0, - new Configuration(), new Configuration(), TestingAbstractInvokables.Receiver.class.getName(), - Collections.<ResultPartitionDeploymentDescriptor>emptyList(), - Collections.singletonList(ircdd), - new ArrayList<>(), Collections.emptyList(), 0); - - new Within(d) { - - @Override - protected void run() { - try { - Future<Object> t1Running = tm.ask( - new TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid1), - timeout); - - Future<Object> t2Running = tm.ask( - new TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid2), - timeout); - - // submit the sender task - tm.tell(new SubmitTask(tdd1), testActorGateway); - expectMsgEquals(Acknowledge.get()); - - // wait until the sender task is running - Await.ready(t1Running, d); - - // only now (after the sender is running), submit the receiver task - tm.tell(new SubmitTask(tdd2), testActorGateway); - expectMsgEquals(Acknowledge.get()); - - // wait until the receiver task is running - Await.ready(t2Running, d); - - tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), testActorGateway); - Map<ExecutionAttemptID, Task> tasks = expectMsgClass(TestingTaskManagerMessages.ResponseRunningTasks - .class).asJava(); - - Task t1 = tasks.get(eid1); - Task t2 = tasks.get(eid2); - - // wait until the tasks are done. thread races may cause the tasks to be done before - // we get to the check, so we need to guard the check - if (t1 != null) { - Future<Object> response = tm.ask(new TestingTaskManagerMessages.NotifyWhenTaskRemoved(eid1), - timeout); - Await.ready(response, d); - } - - if (t2 != null) { - Future<Object> response = tm.ask(new TestingTaskManagerMessages.NotifyWhenTaskRemoved(eid2), - timeout); - Await.ready(response, d); - assertEquals(ExecutionState.FINISHED, t2.getExecutionState()); - } - - tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), testActorGateway); - tasks = expectMsgClass(TestingTaskManagerMessages.ResponseRunningTasks - .class).asJava(); - - assertEquals(0, tasks.size()); - } - catch (Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - } - }; - } - catch (Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - finally { - // shut down the actors - TestingUtils.stopActor(taskManager); - TestingUtils.stopActor(jobManager); - } - }}; - } - - @Test - public void testCancellingDependentAndStateUpdateFails() { - // this tests creates two tasks. the sender sends data, and fails to send the - // state update back to the job manager - // the second one blocks to be canceled - new JavaTestKit(system){{ - - ActorGateway jobManager = null; - ActorGateway taskManager = null; - - final ActorGateway testActorGateway = new AkkaActorGateway( - getTestActor(), - LEADER_SESSION_ID); - try { - final JobID jid = new JobID(); - - JobVertexID vid1 = new JobVertexID(); - JobVertexID vid2 = new JobVertexID(); - - final ExecutionAttemptID eid1 = new ExecutionAttemptID(); - final ExecutionAttemptID eid2 = new ExecutionAttemptID(); - - ActorRef jm = system.actorOf( - Props.create( - new SimpleLookupFailingUpdateJobManagerCreator( - LEADER_SESSION_ID, - eid2) - ) - ); - - jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); - - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new StandaloneLeaderRetrievalService(jobManager.path(), jobManager.leaderSessionID())); - - taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, - new Configuration(), - true, - true); - - final ActorGateway tm = taskManager; - - IntermediateResultPartitionID partitionId = new IntermediateResultPartitionID(); - - List<ResultPartitionDeploymentDescriptor> irpdd = new ArrayList<ResultPartitionDeploymentDescriptor>(); - irpdd.add(new ResultPartitionDeploymentDescriptor(new IntermediateDataSetID(), partitionId, ResultPartitionType.PIPELINED, 1, 1, true)); - - InputGateDeploymentDescriptor ircdd = - new InputGateDeploymentDescriptor( - new IntermediateDataSetID(), ResultPartitionType.PIPELINED, - 0, new InputChannelDeploymentDescriptor[]{ - new InputChannelDeploymentDescriptor(new ResultPartitionID(partitionId, eid1), ResultPartitionLocation.createLocal()) - } - ); - - final TaskDeploymentDescriptor tdd1 = createTaskDeploymentDescriptor( - jid, "TestJob", vid1, eid1, - new SerializedValue<>(new ExecutionConfig()), - "Sender", 1, 0, 1, 0, - new Configuration(), new Configuration(), TestingAbstractInvokables.Sender.class.getName(), - irpdd, Collections.<InputGateDeploymentDescriptor>emptyList(), - new ArrayList<>(), Collections.emptyList(), 0); - - final TaskDeploymentDescriptor tdd2 = createTaskDeploymentDescriptor( - jid, "TestJob", vid2, eid2, - new SerializedValue<>(new ExecutionConfig()), - "Receiver", 7, 2, 7, 0, - new Configuration(), new Configuration(), Tasks.BlockingReceiver.class.getName(), - Collections.<ResultPartitionDeploymentDescriptor>emptyList(), - Collections.singletonList(ircdd), - new ArrayList<>(), Collections.emptyList(), 0); - - new Within(d){ - - @Override - protected void run() { - try { - Future<Object> t1Running = tm.ask( - new TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid1), - timeout); - - Future<Object> t2Running = tm.ask( - new TestingTaskManagerMessages.NotifyWhenTaskIsRunning(eid2), - timeout); - - tm.tell(new SubmitTask(tdd2), testActorGateway); - tm.tell(new SubmitTask(tdd1), testActorGateway); - - expectMsgEquals(Acknowledge.get()); - expectMsgEquals(Acknowledge.get()); - - Await.ready(t1Running, d); - Await.ready(t2Running, d); - - tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), testActorGateway); - Map<ExecutionAttemptID, Task> tasks = expectMsgClass(TestingTaskManagerMessages - .ResponseRunningTasks.class).asJava(); - - Task t1 = tasks.get(eid1); - Task t2 = tasks.get(eid2); - - tm.tell(new CancelTask(eid2), testActorGateway); - expectMsgEquals(Acknowledge.get()); - - if (t2 != null) { - Future<Object> response = tm.ask(new TestingTaskManagerMessages.NotifyWhenTaskRemoved(eid2), - timeout); - Await.ready(response, d); - } - - if (t1 != null) { - if (t1.getExecutionState() == ExecutionState.RUNNING) { - tm.tell(new CancelTask(eid1), testActorGateway); - expectMsgEquals(Acknowledge.get()); - } - Future<Object> response = tm.ask( - new TestingTaskManagerMessages.NotifyWhenTaskRemoved(eid1), - timeout); - Await.ready(response, d); - } - - tm.tell(TestingTaskManagerMessages.getRequestRunningTasksMessage(), testActorGateway); - tasks = expectMsgClass(TestingTaskManagerMessages - .ResponseRunningTasks.class).asJava(); - - assertEquals(0, tasks.size()); - } - catch (Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - } - }; - } - catch (Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - finally { - // shut down the actors - TestingUtils.stopActor(taskManager); - TestingUtils.stopActor(jobManager); - } - }}; - } - - /** - * Tests that repeated remote {@link PartitionNotFoundException}s ultimately fail the receiver. - */ - @Test - public void testRemotePartitionNotFound() throws Exception { - - new JavaTestKit(system){{ - - ActorGateway jobManager = null; - ActorGateway taskManager = null; - - final ActorGateway testActorGateway = new AkkaActorGateway( - getTestActor(), - LEADER_SESSION_ID); - - try { - final IntermediateDataSetID resultId = new IntermediateDataSetID(); - - // Create the JM - ActorRef jm = system.actorOf(Props.create( - new SimplePartitionStateLookupJobManagerCreator(LEADER_SESSION_ID, getTestActor()))); - - jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); - - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new StandaloneLeaderRetrievalService(jobManager.path(), jobManager.leaderSessionID())); - - final int dataPort = NetUtils.getAvailablePort(); - Configuration config = new Configuration(); - config.setInteger(TaskManagerOptions.DATA_PORT, dataPort); - config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100); - config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_MAX, 200); - - taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, - config, - false, - true); - - // --------------------------------------------------------------------------------- - - final ActorGateway tm = taskManager; - - final JobID jid = new JobID(); - final JobVertexID vid = new JobVertexID(); - final ExecutionAttemptID eid = new ExecutionAttemptID(); - - final ResultPartitionID partitionId = new ResultPartitionID(); - - // Remote location (on the same TM though) for the partition - final ResultPartitionLocation loc = ResultPartitionLocation - .createRemote(new ConnectionID( - new InetSocketAddress("localhost", dataPort), 0)); - - final InputChannelDeploymentDescriptor[] icdd = - new InputChannelDeploymentDescriptor[] { - new InputChannelDeploymentDescriptor(partitionId, loc)}; - - final InputGateDeploymentDescriptor igdd = - new InputGateDeploymentDescriptor(resultId, ResultPartitionType.PIPELINED, 0, icdd); - - final TaskDeploymentDescriptor tdd = createTaskDeploymentDescriptor( - jid, "TestJob", vid, eid, - new SerializedValue<>(new ExecutionConfig()), - "Receiver", 1, 0, 1, 0, - new Configuration(), new Configuration(), - Tasks.AgnosticReceiver.class.getName(), - Collections.<ResultPartitionDeploymentDescriptor>emptyList(), - Collections.singletonList(igdd), - Collections.emptyList(), - Collections.emptyList(), 0); - - new Within(d) { - @Override - protected void run() { - // Submit the task - tm.tell(new SubmitTask(tdd), testActorGateway); - expectMsgClass(Acknowledge.get().getClass()); - - // Wait to be notified about the final execution state by the mock JM - TaskExecutionState msg = expectMsgClass(TaskExecutionState.class); - - // The task should fail after repeated requests - assertEquals(ExecutionState.FAILED, msg.getExecutionState()); - Throwable t = msg.getError(ClassLoader.getSystemClassLoader()); - assertEquals("Thrown exception was not a PartitionNotFoundException: " + t.getMessage(), - PartitionNotFoundException.class, t.getClass()); - } - }; - } - catch (Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - finally { - TestingUtils.stopActor(taskManager); - TestingUtils.stopActor(jobManager); - } - }}; - } - - @Test - public void testTaskManagerServicesConfiguration() throws Exception { - - // set some non-default values - final Configuration config = new Configuration(); - config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100); - config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_MAX, 200); - config.setInteger(TaskManagerOptions.NETWORK_BUFFERS_PER_CHANNEL, 10); - config.setInteger(TaskManagerOptions.NETWORK_EXTRA_BUFFERS_PER_GATE, 100); - - TaskManagerServicesConfiguration tmConfig = - TaskManagerServicesConfiguration.fromConfiguration(config, InetAddress.getLoopbackAddress(), true); - - assertEquals(tmConfig.getNetworkConfig().partitionRequestInitialBackoff(), 100); - assertEquals(tmConfig.getNetworkConfig().partitionRequestMaxBackoff(), 200); - assertEquals(tmConfig.getNetworkConfig().networkBuffersPerChannel(), 10); - assertEquals(tmConfig.getNetworkConfig().floatingNetworkBuffersPerGate(), 100); - } - - /** - * Tests that repeated local {@link PartitionNotFoundException}s ultimately fail the receiver. - */ - @Test - public void testLocalPartitionNotFound() throws Exception { - - new JavaTestKit(system){{ - - ActorGateway jobManager = null; - ActorGateway taskManager = null; - - final ActorGateway testActorGateway = new AkkaActorGateway( - getTestActor(), - LEADER_SESSION_ID); - - try { - final IntermediateDataSetID resultId = new IntermediateDataSetID(); - - // Create the JM - ActorRef jm = system.actorOf(Props.create( - new SimplePartitionStateLookupJobManagerCreator(LEADER_SESSION_ID, getTestActor()))); - - jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); - - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new StandaloneLeaderRetrievalService(jobManager.path(), jobManager.leaderSessionID())); - - final Configuration config = new Configuration(); - config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100); - config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_MAX, 200); - - taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, - config, - true, - true); - - // --------------------------------------------------------------------------------- - - final ActorGateway tm = taskManager; - - final JobID jid = new JobID(); - final JobVertexID vid = new JobVertexID(); - final ExecutionAttemptID eid = new ExecutionAttemptID(); - - final ResultPartitionID partitionId = new ResultPartitionID(); - - // Local location (on the same TM though) for the partition - final ResultPartitionLocation loc = ResultPartitionLocation.createLocal(); - - final InputChannelDeploymentDescriptor[] icdd = - new InputChannelDeploymentDescriptor[] { - new InputChannelDeploymentDescriptor(partitionId, loc)}; - - final InputGateDeploymentDescriptor igdd = - new InputGateDeploymentDescriptor(resultId, ResultPartitionType.PIPELINED, 0, icdd); - - final TaskDeploymentDescriptor tdd = createTaskDeploymentDescriptor( - jid, "TestJob", vid, eid, - new SerializedValue<>(new ExecutionConfig()), - "Receiver", 1, 0, 1, 0, - new Configuration(), new Configuration(), - Tasks.AgnosticReceiver.class.getName(), - Collections.<ResultPartitionDeploymentDescriptor>emptyList(), - Collections.singletonList(igdd), - Collections.emptyList(), - Collections.emptyList(), 0); - - new Within(new FiniteDuration(120, TimeUnit.SECONDS)) { - @Override - protected void run() { - // Submit the task - tm.tell(new SubmitTask(tdd), testActorGateway); - expectMsgClass(Acknowledge.get().getClass()); - - // Wait to be notified about the final execution state by the mock JM - TaskExecutionState msg = expectMsgClass(TaskExecutionState.class); - - // The task should fail after repeated requests - assertEquals(msg.getExecutionState(), ExecutionState.FAILED); - - Throwable error = msg.getError(getClass().getClassLoader()); - if (error.getClass() != PartitionNotFoundException.class) { - error.printStackTrace(); - fail("Wrong exception: " + error.getMessage()); - } - } - }; - } - catch (Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - finally { - TestingUtils.stopActor(taskManager); - TestingUtils.stopActor(jobManager); - } - }}; - } - - @Test - public void testLogNotFoundHandling() throws Exception { - - new JavaTestKit(system){{ - - // we require a JobManager so that the BlobService is also started - ActorGateway jobManager = null; - ActorGateway taskManager = null; - - try { - - // Create the JM - ActorRef jm = system.actorOf(Props.create( - new SimplePartitionStateLookupJobManagerCreator(LEADER_SESSION_ID, getTestActor()))); - - jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); - - final int dataPort = NetUtils.getAvailablePort(); - Configuration config = new Configuration(); - config.setInteger(TaskManagerOptions.DATA_PORT, dataPort); - config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_INITIAL, 100); - config.setInteger(TaskManagerOptions.NETWORK_REQUEST_BACKOFF_MAX, 200); - config.setString(ConfigConstants.TASK_MANAGER_LOG_PATH_KEY, "/i/dont/exist"); - - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new StandaloneLeaderRetrievalService(jobManager.path(), jobManager.leaderSessionID())); - - taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, - config, - false, - true); - - // --------------------------------------------------------------------------------- - - final ActorGateway tm = taskManager; - - new Within(d) { - @Override - protected void run() { - Future<Object> logFuture = tm.ask(TaskManagerMessages.getRequestTaskManagerLog(), timeout); - try { - Await.result(logFuture, timeout); - Assert.fail(); - } catch (Exception e) { - Assert.assertTrue(e.getMessage().startsWith("TaskManager log files are unavailable. Log file could not be found at")); - } - } - }; - } finally { - TestingUtils.stopActor(taskManager); - TestingUtils.stopActor(jobManager); - } - }}; - } - - // ------------------------------------------------------------------------ - // Stack trace sample - // ------------------------------------------------------------------------ - - /** - * Tests sampling of task stack traces. - */ - @Test - @SuppressWarnings("unchecked") - public void testTriggerStackTraceSampleMessage() throws Exception { - new JavaTestKit(system) {{ - ActorGateway taskManagerActorGateway = null; - - // We need this to be a JM that answers to update messages for - // robustness on Travis (if jobs need to be resubmitted in (4)). - ActorRef jm = system.actorOf(Props.create(new SimpleLookupJobManagerCreator( - HighAvailabilityServices.DEFAULT_LEADER_ID))); - ActorGateway jobManagerActorGateway = new AkkaActorGateway( - jm, - HighAvailabilityServices.DEFAULT_LEADER_ID); - - final ActorGateway testActorGateway = new AkkaActorGateway( - getTestActor(), - HighAvailabilityServices.DEFAULT_LEADER_ID); - - try { - final ActorGateway jobManager = jobManagerActorGateway; - - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new StandaloneLeaderRetrievalService(jobManager.path(), jobManager.leaderSessionID())); - - final ActorGateway taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, - new Configuration(), - true, - false); - - final JobID jobId = new JobID(); - - // Single blocking task - final TaskDeploymentDescriptor tdd = createTaskDeploymentDescriptor( - jobId, - "Job", - new JobVertexID(), - new ExecutionAttemptID(), - new SerializedValue<>(new ExecutionConfig()), - "Task", - 1, - 0, - 1, - 0, - new Configuration(), - new Configuration(), - BlockingNoOpInvokable.class.getName(), - Collections.<ResultPartitionDeploymentDescriptor>emptyList(), - Collections.<InputGateDeploymentDescriptor>emptyList(), - Collections.emptyList(), - Collections.emptyList(), - 0); - - // Submit the task - new Within(d) { - - @Override - protected void run() { - try { - // Make sure to register - Future<?> connectFuture = taskManager.ask(new TestingTaskManagerMessages - .NotifyWhenRegisteredAtJobManager(jobManager.actor()), remaining()); - Await.ready(connectFuture, remaining()); - - Future<Object> taskRunningFuture = taskManager.ask( - new TestingTaskManagerMessages.NotifyWhenTaskIsRunning( - tdd.getExecutionAttemptId()), timeout); - - taskManager.tell(new SubmitTask(tdd)); - - Await.ready(taskRunningFuture, d); - } catch (Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - } - }; - - // - // 1) Trigger sample for non-existing task - // - new Within(d) { - @Override - protected void run() { - try { - ExecutionAttemptID taskId = new ExecutionAttemptID(); - - taskManager.tell(new TriggerStackTraceSample( - 112223, - taskId, - 100, - timeD, - 0), - testActorGateway); - - // Receive the expected message (heartbeat races possible) - Object[] msg = receiveN(1); - while (!(msg[0] instanceof Status.Failure)) { - msg = receiveN(1); - } - - Status.Failure response = (Status.Failure) msg[0]; - - assertEquals(IllegalStateException.class, response.cause().getClass()); - } catch (Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - } - }; - - // - // 2) Trigger sample for the blocking task - // - new Within(d) { - @Override - protected void run() { - boolean success = false; - Throwable lastError = null; - - for (int i = 0; i < 100 && !success; i++) { - try { - int numSamples = 5; - - taskManager.tell(new TriggerStackTraceSample( - 19230, - tdd.getExecutionAttemptId(), - numSamples, - Time.milliseconds(100L), - 0), - testActorGateway); - - // Receive the expected message (heartbeat races possible) - Object[] msg = receiveN(1); - while (!(msg[0] instanceof StackTraceSampleResponse)) { - msg = receiveN(1); - } - - StackTraceSampleResponse response = (StackTraceSampleResponse) msg[0]; - - // ---- Verify response ---- - assertEquals(19230, response.getSampleId()); - assertEquals(tdd.getExecutionAttemptId(), response.getExecutionAttemptID()); - - List<StackTraceElement[]> traces = response.getSamples(); - - assertEquals("Number of samples", numSamples, traces.size()); - - for (StackTraceElement[] trace : traces) { - // Look for BlockingNoOpInvokable#invoke - for (StackTraceElement elem : trace) { - if (elem.getClassName().equals( - BlockingNoOpInvokable.class.getName())) { - - assertEquals("invoke", elem.getMethodName()); - success = true; - break; - } - } - - assertTrue("Unexpected stack trace: " + - Arrays.toString(trace), success); - } - } catch (Throwable t) { - lastError = t; - LOG.warn("Failed to find invokable.", t); - } - - try { - Thread.sleep(100); - } catch (InterruptedException e) { - LOG.error("Interrupted while sleeping before retry.", e); - break; - } - } - - if (!success) { - if (lastError == null) { - fail("Failed to find invokable"); - } else { - fail(lastError.getMessage()); - } - } - } - }; - - // - // 3) Trigger sample for the blocking task with max depth - // - new Within(d) { - @Override - protected void run() { - try { - int numSamples = 5; - int maxDepth = 2; - - taskManager.tell(new TriggerStackTraceSample( - 1337, - tdd.getExecutionAttemptId(), - numSamples, - Time.milliseconds(100L), - maxDepth), - testActorGateway); - - // Receive the expected message (heartbeat races possible) - Object[] msg = receiveN(1); - while (!(msg[0] instanceof StackTraceSampleResponse)) { - msg = receiveN(1); - } - - StackTraceSampleResponse response = (StackTraceSampleResponse) msg[0]; - - // ---- Verify response ---- - assertEquals(1337, response.getSampleId()); - assertEquals(tdd.getExecutionAttemptId(), response.getExecutionAttemptID()); - - List<StackTraceElement[]> traces = response.getSamples(); - - assertEquals("Number of samples", numSamples, traces.size()); - - for (StackTraceElement[] trace : traces) { - assertEquals("Max depth", maxDepth, trace.length); - } - } catch (Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - } - }; - - // - // 4) Trigger sample for the blocking task, but cancel it during sampling - // - new Within(d) { - @Override - protected void run() { - try { - int maxAttempts = 10; - int sleepTime = 100; - for (int i = 0; i < maxAttempts; i++, sleepTime *= 2) { - // Trigger many samples in order to cancel the task - // during a sample - taskManager.tell( - new TriggerStackTraceSample( - 44, - tdd.getExecutionAttemptId(), - Integer.MAX_VALUE, - Time.milliseconds(10L), - 0), - testActorGateway); - - Thread.sleep(sleepTime); - - Future<?> removeFuture = taskManager.ask( - new TestingTaskManagerMessages.NotifyWhenJobRemoved(jobId), - remaining()); - - // Cancel the task - taskManager.tell(new CancelTask(tdd.getExecutionAttemptId())); - - // Receive the expected message (heartbeat races possible) - while (true) { - Object[] msg = receiveN(1); - if (msg[0] instanceof StackTraceSampleResponse) { - StackTraceSampleResponse response = (StackTraceSampleResponse) msg[0]; - - assertEquals(tdd.getExecutionAttemptId(), response.getExecutionAttemptID()); - assertEquals(44, response.getSampleId()); - - // Done - return; - } else if (msg[0] instanceof Failure) { - // Wait for removal before resubmitting - Await.ready(removeFuture, remaining()); - - Future<?> taskRunningFuture = taskManager.ask( - new TestingTaskManagerMessages.NotifyWhenTaskIsRunning( - tdd.getExecutionAttemptId()), timeout); - - // Resubmit - taskManager.tell(new SubmitTask(tdd)); - - Await.ready(taskRunningFuture, remaining()); - - // Retry the sample message - break; - } else { - // Different message - continue; - } - } - } - } catch (Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - } - }; - } finally { - TestingUtils.stopActor(taskManagerActorGateway); - TestingUtils.stopActor(jobManagerActorGateway); - } - }}; - } - - @Test - public void testTerminationOnFatalError() { - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new SettableLeaderRetrievalService()); - - new JavaTestKit(system){{ - - final ActorGateway taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, // no jobmanager - new Configuration(), - true, - false); - - try { - watch(taskManager.actor()); - taskManager.tell(new FatalError("test fatal error", new Exception("something super bad"))); - expectTerminated(d, taskManager.actor()); - } - finally { - taskManager.tell(Kill.getInstance()); - } - }}; - } - - /** - * Test that a failing schedule or update consumers call leads to the failing of the respective - * task. - * - * <p>IMPORTANT: We have to make sure that the invokable's cancel method is called, because only - * then the future is completed. We do this by not eagerly deploy consumer tasks and requiring - * the invokable to fill one memory segment. The completed memory segment will trigger the - * scheduling of the downstream operator since it is in pipeline mode. After we've filled the - * memory segment, we'll block the invokable and wait for the task failure due to the failed - * schedule or update consumers call. - */ - @Test(timeout = 10000L) - public void testFailingScheduleOrUpdateConsumersMessage() throws Exception { - new JavaTestKit(system) {{ - final Configuration configuration = new Configuration(); - - // set the memory segment to the smallest size possible, because we have to fill one - // memory buffer to trigger the schedule or update consumers message to the downstream - // operators - configuration.setString(TaskManagerOptions.MEMORY_SEGMENT_SIZE, "4096"); - - final JobID jid = new JobID(); - final JobVertexID vid = new JobVertexID(); - final ExecutionAttemptID eid = new ExecutionAttemptID(); - final SerializedValue<ExecutionConfig> executionConfig = new SerializedValue<>(new ExecutionConfig()); - - final ResultPartitionDeploymentDescriptor resultPartitionDeploymentDescriptor = new ResultPartitionDeploymentDescriptor( - new IntermediateDataSetID(), - new IntermediateResultPartitionID(), - ResultPartitionType.PIPELINED, - 1, - 1, - true); - - final TaskDeploymentDescriptor tdd = createTaskDeploymentDescriptor(jid, "TestJob", vid, eid, executionConfig, - "TestTask", 1, 0, 1, 0, new Configuration(), new Configuration(), - TestInvokableRecordCancel.class.getName(), - Collections.singletonList(resultPartitionDeploymentDescriptor), - Collections.<InputGateDeploymentDescriptor>emptyList(), - new ArrayList<>(), Collections.emptyList(), 0); - - ActorRef jmActorRef = system.actorOf(Props.create(FailingScheduleOrUpdateConsumersJobManager.class, LEADER_SESSION_ID), "jobmanager"); - ActorGateway jobManager = new AkkaActorGateway(jmActorRef, LEADER_SESSION_ID); - - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new StandaloneLeaderRetrievalService(jobManager.path(), jobManager.leaderSessionID())); - - final ActorGateway taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, - configuration, - true, - true); - - try { - TestInvokableRecordCancel.resetGotCanceledFuture(); - - Future<Object> result = taskManager.ask(new SubmitTask(tdd), timeout); - - Await.result(result, timeout); - - CompletableFuture<Boolean> cancelFuture = TestInvokableRecordCancel.gotCanceled(); - - assertEquals(true, cancelFuture.get()); - } finally { - TestingUtils.stopActor(taskManager); - TestingUtils.stopActor(jobManager); - } - }}; - } - - /** - * Tests that the TaskManager sends a proper exception back to the sender if the submit task - * message fails. - */ - @Test - public void testSubmitTaskFailure() throws Exception { - ActorGateway jobManager = null; - ActorGateway taskManager = null; - - try { - - ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, LEADER_SESSION_ID)); - jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); - - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new StandaloneLeaderRetrievalService(jobManager.path(), jobManager.leaderSessionID())); - - taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, - new Configuration(), - true, - true); - - TaskDeploymentDescriptor tdd = createTaskDeploymentDescriptor( - new JobID(), - "test job", - new JobVertexID(), - new ExecutionAttemptID(), - new SerializedValue<>(new ExecutionConfig()), - "test task", - 0, // this will make the submission fail because the number of key groups must be >= 1 - 0, - 1, - 0, - new Configuration(), - new Configuration(), - "Foobar", - Collections.<ResultPartitionDeploymentDescriptor>emptyList(), - Collections.<InputGateDeploymentDescriptor>emptyList(), - Collections.emptyList(), - Collections.emptyList(), - 0); - - Future<Object> submitResponse = taskManager.ask(new SubmitTask(tdd), timeout); - - try { - Await.result(submitResponse, timeout); - - fail("The submit task message should have failed."); - } catch (IllegalArgumentException e) { - // expected - } - } finally { - TestingUtils.stopActor(jobManager); - TestingUtils.stopActor(taskManager); - } - } - - /** - * Tests that the TaskManager sends a proper exception back to the sender if the stop task - * message fails. - */ - @Test - public void testStopTaskFailure() throws Exception { - ActorGateway jobManager = null; - ActorGateway taskManager = null; - - try { - final ExecutionAttemptID executionAttemptId = new ExecutionAttemptID(); - - ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, LEADER_SESSION_ID)); - jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); - - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new StandaloneLeaderRetrievalService(jobManager.path(), jobManager.leaderSessionID())); - - taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, - new Configuration(), - true, - true); - - TaskDeploymentDescriptor tdd = createTaskDeploymentDescriptor( - new JobID(), - "test job", - new JobVertexID(), - executionAttemptId, - new SerializedValue<>(new ExecutionConfig()), - "test task", - 1, - 0, - 1, - 0, - new Configuration(), - new Configuration(), - BlockingNoOpInvokable.class.getName(), - Collections.<ResultPartitionDeploymentDescriptor>emptyList(), - Collections.<InputGateDeploymentDescriptor>emptyList(), - Collections.emptyList(), - Collections.emptyList(), - 0); - - Future<Object> submitResponse = taskManager.ask(new SubmitTask(tdd), timeout); - - Await.result(submitResponse, timeout); - - final Future<Object> taskRunning = taskManager.ask(new TestingTaskManagerMessages.NotifyWhenTaskIsRunning(executionAttemptId), timeout); - - Await.result(taskRunning, timeout); - - Future<Object> stopResponse = taskManager.ask(new StopTask(executionAttemptId), timeout); - - try { - Await.result(stopResponse, timeout); - - fail("The stop task message should have failed."); - } catch (UnsupportedOperationException e) { - // expected - } - } finally { - TestingUtils.stopActor(jobManager); - TestingUtils.stopActor(taskManager); - } - } - - /** - * Tests that the TaskManager sends a proper exception back to the sender if the trigger stack - * trace message fails. - */ - @Test - public void testStackTraceSampleFailure() throws Exception { - ActorGateway jobManager = null; - ActorGateway taskManager = null; - - try { - - ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, LEADER_SESSION_ID)); - jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); - - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new StandaloneLeaderRetrievalService(jobManager.path(), jobManager.leaderSessionID())); - - taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, - new Configuration(), - true, - true); - - Future<Object> stackTraceResponse = taskManager.ask( - new TriggerStackTraceSample( - 0, - new ExecutionAttemptID(), - 0, - Time.milliseconds(1L), - 0), - timeout); - - try { - Await.result(stackTraceResponse, timeout); - - fail("The trigger stack trace message should have failed."); - } catch (IllegalStateException e) { - // expected - } - } finally { - TestingUtils.stopActor(jobManager); - TestingUtils.stopActor(taskManager); - } - } - - /** - * Tests that the TaskManager sends a proper exception back to the sender if the trigger stack - * trace message fails. - */ - @Test - public void testUpdateTaskInputPartitionsFailure() throws Exception { - ActorGateway jobManager = null; - ActorGateway taskManager = null; - - try { - - final ExecutionAttemptID executionAttemptId = new ExecutionAttemptID(); - - ActorRef jm = system.actorOf(Props.create(SimpleJobManager.class, LEADER_SESSION_ID)); - jobManager = new AkkaActorGateway(jm, LEADER_SESSION_ID); - - highAvailabilityServices.setJobMasterLeaderRetriever( - HighAvailabilityServices.DEFAULT_JOB_ID, - new StandaloneLeaderRetrievalService(jobManager.path(), jobManager.leaderSessionID())); - - taskManager = TestingUtils.createTaskManager( - system, - highAvailabilityServices, - new Configuration(), - true, - true); - - TaskDeploymentDescriptor tdd = createTaskDeploymentDescriptor( - new JobID(), - "test job", - new JobVertexID(), - executionAttemptId, - new SerializedValue<>(new ExecutionConfig()), - "test task", - 1, - 0, - 1, - 0, - new Configuration(), - new Configuration(), - BlockingNoOpInvokable.class.getName(), - Collections.<ResultPartitionDeploymentDescriptor>emptyList(), - Collections.<InputGateDeploymentDescriptor>emptyList(), - Collections.emptyList(), - Collections.emptyList(), - 0); - - Future<Object> submitResponse = taskManager.ask(new SubmitTask(tdd), timeout); - - Await.result(submitResponse, timeout); - - Future<Object> partitionUpdateResponse = taskManager.ask( - new TaskMessages.UpdateTaskSinglePartitionInfo( - executionAttemptId, - new IntermediateDataSetID(), - new InputChannelDeploymentDescriptor(new ResultPartitionID(), ResultPartitionLocation.createLocal())), - timeout); - - try { - Await.result(partitionUpdateResponse, timeout); - - fail("The update task input partitions message should have failed."); - } catch (Exception e) { - // expected - } - } finally { - TestingUtils.stopActor(jobManager); - TestingUtils.stopActor(taskManager); - } - } - - // -------------------------------------------------------------------------------------------- - - public static class SimpleJobManager extends FlinkUntypedActor { - - private final UUID leaderSessionID; - - public SimpleJobManager(UUID leaderSessionID) { - this.leaderSessionID = leaderSessionID; - } - - @Override - public void handleMessage(Object message) throws Exception { - if (message instanceof RegistrationMessages.RegisterTaskManager) { - final InstanceID iid = new InstanceID(); - final ActorRef self = getSelf(); - getSender().tell( - decorateMessage( - new RegistrationMessages.AcknowledgeRegistration( - iid, - 12345) - ), - self); - } - else if (message instanceof TaskMessages.UpdateTaskExecutionState){ - getSender().tell(true, getSelf()); - } - } - - @Override - protected UUID getLeaderSessionID() { - return leaderSessionID; - } - } - - public static class FailingScheduleOrUpdateConsumersJobManager extends SimpleJobManager { - - public FailingScheduleOrUpdateConsumersJobManager(UUID leaderSessionId) { - super(leaderSessionId); - } - - @Override - public void handleMessage(Object message) throws Exception { - if (message instanceof ScheduleOrUpdateConsumers) { - getSender().tell( - decorateMessage( - new Status.Failure(new Exception("Could not schedule or update consumers."))), - getSelf()); - } else { - super.handleMessage(message); - } - } - } - - public static class SimpleLookupJobManager extends SimpleJobManager { - - public SimpleLookupJobManager(UUID leaderSessionID) { - super(leaderSessionID); - } - - @Override - public void handleMessage(Object message) throws Exception { - if (message instanceof ScheduleOrUpdateConsumers) { - getSender().tell( - decorateMessage(Acknowledge.get()), - getSelf() - ); - } else { - super.handleMessage(message); - } - } - } - - public static class SimpleLookupFailingUpdateJobManager extends SimpleLookupJobManager{ - - private final Set<ExecutionAttemptID> validIDs; - - public SimpleLookupFailingUpdateJobManager(UUID leaderSessionID, Set<ExecutionAttemptID> ids) { - super(leaderSessionID); - this.validIDs = new HashSet<>(ids); - } - - @Override - public void handleMessage(Object message) throws Exception{ - if (message instanceof TaskMessages.UpdateTaskExecutionState) { - TaskMessages.UpdateTaskExecutionState updateMsg = - (TaskMessages.UpdateTaskExecutionState) message; - - if (validIDs.contains(updateMsg.taskExecutionState().getID())) { - getSender().tell(true, getSelf()); - } else { - getSender().tell(false, getSelf()); - } - } else { - super.handleMessage(message); - } - } - } - - public static class SimplePartitionStateLookupJobManager extends SimpleJobManager { - - private final ActorRef testActor; - - public SimplePartitionStateLookupJobManager(UUID leaderSessionID, ActorRef testActor) { - super(leaderSessionID); - this.testActor = testActor; - } - - @Override - public void handleMessage(Object message) throws Exception { - if (message instanceof RequestPartitionProducerState) { - getSender().tell(decorateMessage(ExecutionState.RUNNING), getSelf()); - } - else if (message instanceof TaskMessages.UpdateTaskExecutionState) { - final TaskExecutionState msg = ((TaskMessages.UpdateTaskExecutionState) message) - .taskExecutionState(); - - if (msg.getExecutionState().isTerminal()) { - testActor.tell(msg, self()); - } - } else { - super.handleMessage(message); - } - } - } - - public static class SimpleLookupJobManagerCreator implements Creator<SimpleLookupJobManager>{ - - private final UUID leaderSessionID; - - public SimpleLookupJobManagerCreator(UUID leaderSessionID) { - this.leaderSessionID = leaderSessionID; - } - - @Override - public SimpleLookupJobManager create() throws Exception { - return new SimpleLookupJobManager(leaderSessionID); - } - } - - public static class SimpleLookupFailingUpdateJobManagerCreator implements Creator<SimpleLookupFailingUpdateJobManager>{ - - private final UUID leaderSessionID; - - private final Set<ExecutionAttemptID> validIDs; - - public SimpleLookupFailingUpdateJobManagerCreator(UUID leaderSessionID, ExecutionAttemptID ... ids) { - this.leaderSessionID = leaderSessionID; - - validIDs = new HashSet<ExecutionAttemptID>(); - - for (ExecutionAttemptID id : ids) { - this.validIDs.add(id); - } - } - - @Override - public SimpleLookupFailingUpdateJobManager create() throws Exception { - return new SimpleLookupFailingUpdateJobManager(leaderSessionID, validIDs); - } - } - - public static class SimplePartitionStateLookupJobManagerCreator implements Creator<SimplePartitionStateLookupJobManager>{ - - private final UUID leaderSessionID; - - private final ActorRef testActor; - - public SimplePartitionStateLookupJobManagerCreator(UUID leaderSessionID, ActorRef testActor) { - this.leaderSessionID = leaderSessionID; - - this.testActor = testActor; - } - - @Override - public SimplePartitionStateLookupJobManager create() throws Exception { - return new SimplePartitionStateLookupJobManager(leaderSessionID, testActor); - } - } - - // -------------------------------------------------------------------------------------------- - - public static final class TestInvokableCorrect extends AbstractInvokable { - - public TestInvokableCorrect(Environment environment) { - super(environment); - } - - @Override - public void invoke() {} - } - - public static class TestInvokableBlockingCancelable extends AbstractInvokable { - - public TestInvokableBlockingCancelable(Environment environment) { - super(environment); - } - - @Override - public void invoke() throws Exception { - final Object o = new Object(); - //noinspection SynchronizationOnLocalVariableOrMethodParameter - synchronized (o) { - //noinspection InfiniteLoopStatement - while (true) { - o.wait(); - } - } - } - } - - public static final class TestInvokableRecordCancel extends AbstractInvokable { - - private static final Object lock = new Object(); - private static CompletableFuture<Boolean> gotCanceledFuture = new CompletableFuture<>(); - - public TestInvokableRecordCancel(Environment environment) { - super(environment); - } - - @Override - public void invoke() throws Exception { - final Object o = new Object(); - RecordWriter<IntValue> recordWriter = new RecordWriterBuilder().build(getEnvironment().getWriter(0)); - - for (int i = 0; i < 1024; i++) { - recordWriter.emit(new IntValue(42)); - } - - synchronized (o) { - //noinspection InfiniteLoopStatement - while (true) { - o.wait(); - } - } - - } - - @Override - public void cancel() { - synchronized (lock) { - gotCanceledFuture.complete(true); - } - } - - public static void resetGotCanceledFuture() { - synchronized (lock) { - gotCanceledFuture = new CompletableFuture<>(); - } - } - - public static CompletableFuture<Boolean> gotCanceled() { - synchronized (lock) { - return gotCanceledFuture; - } - } - } - - private static TaskDeploymentDescriptor createTaskDeploymentDescriptor( - JobID jobId, - String jobName, - JobVertexID jobVertexId, - ExecutionAttemptID executionAttemptId, - SerializedValue<ExecutionConfig> serializedExecutionConfig, - String taskName, - int numberOfKeyGroups, - int subtaskIndex, - int parallelism, - int attemptNumber, - Configuration jobConfiguration, - Configuration taskConfiguration, - String invokableClassName, - Collection<ResultPartitionDeploymentDescriptor> producedPartitions, - Collection<InputGateDeploymentDescriptor> inputGates, - Collection<PermanentBlobKey> requiredJarFiles, - Collection<URL> requiredClasspaths, - int targetSlotNumber) throws IOException { - - JobInformation jobInformation = new JobInformation( - jobId, - jobName, - serializedExecutionConfig, - jobConfiguration, - requiredJarFiles, - requiredClasspaths); - - TaskInformation taskInformation = new TaskInformation( - jobVertexId, - taskName, - parallelism, - numberOfKeyGroups, - invokableClassName, - taskConfiguration); - - SerializedValue<JobInformation> serializedJobInformation = new SerializedValue<>(jobInformation); - SerializedValue<TaskInformation> serializedJobVertexInformation = new SerializedValue<>(taskInformation); - - return new TaskDeploymentDescriptor( - jobId, - new TaskDeploymentDescriptor.NonOffloaded<>(serializedJobInformation), - new TaskDeploymentDescriptor.NonOffloaded<>(serializedJobVertexInformation), - executionAttemptId, - new AllocationID(), - subtaskIndex, - attemptNumber, - targetSlotNumber, - null, - producedPartitions, - inputGates); - - } -} diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala index 1bce221..c68b45e 100644 --- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala +++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala @@ -200,81 +200,6 @@ object TestingUtils { } } - /** Creates a local TaskManager in the given ActorSystem. It is given a - * [[StandaloneLeaderRetrievalService]] which returns the given jobManagerURL. After creating - * the TaskManager, waitForRegistration specifies whether one waits until the TaskManager has - * registered at the JobManager. An ActorGateway to the TaskManager is returned. - * - * @param actorSystem ActorSystem in which the TaskManager shall be started - * @param highAvailabilityServices Service factory for high availability - * @param configuration Configuration - * @param useLocalCommunication true if the network stack shall use exclusively local - * communication - * @param waitForRegistration true if the method will wait until the TaskManager has connected to - * the JobManager - * @return ActorGateway of the created TaskManager - */ - def createTaskManager( - actorSystem: ActorSystem, - highAvailabilityServices: HighAvailabilityServices, - configuration: Configuration, - useLocalCommunication: Boolean, - waitForRegistration: Boolean) - : ActorGateway = { - createTaskManager( - actorSystem, - highAvailabilityServices, - configuration, - useLocalCommunication, - waitForRegistration, - classOf[TestingTaskManager] - ) - } - - def createTaskManager( - actorSystem: ActorSystem, - highAvailabilityServices: HighAvailabilityServices, - configuration: Configuration, - useLocalCommunication: Boolean, - waitForRegistration: Boolean, - taskManagerClass: Class[_ <: TaskManager]) - : ActorGateway = { - - val resultingConfiguration = new Configuration() - - resultingConfiguration.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "10m") - - resultingConfiguration.addAll(configuration) - - val metricRegistry = new MetricRegistryImpl( - MetricRegistryConfiguration.fromConfiguration(configuration)) - - val taskManagerResourceId = ResourceID.generate() - - val taskManager = TaskManager.startTaskManagerComponentsAndActor( - resultingConfiguration, - taskManagerResourceId, - actorSystem, - highAvailabilityServices, - metricRegistry, - "localhost", - None, - useLocalCommunication, - taskManagerClass - ) - - val leaderId = if (waitForRegistration) { - val notificationResult = (taskManager ? NotifyWhenRegisteredAtJobManager)(TESTING_DURATION) - .mapTo[RegisteredAtJobManager] - - Await.result(notificationResult, TESTING_DURATION).leaderId - } else { - HighAvailabilityServices.DEFAULT_LEADER_ID - } - - new AkkaActorGateway(taskManager, leaderId) - } - /** Stops the given actor by sending it a Kill message * * @param actor @@ -284,115 +209,4 @@ object TestingUtils { actor ! Kill } } - - /** Stops the given actor by sending it a Kill message - * - * @param actorGateway - */ - def stopActor(actorGateway: ActorGateway): Unit = { - if (actorGateway != null) { - stopActor(actorGateway.actor()) - } - } - - def stopActorGracefully(actor: ActorRef): Unit = { - val gracefulStopFuture = Patterns.gracefulStop(actor, TestingUtils.TESTING_TIMEOUT) - - Await.result(gracefulStopFuture, TestingUtils.TESTING_TIMEOUT) - } - - def stopActorGracefully(actorGateway: ActorGateway): Unit = { - stopActorGracefully(actorGateway.actor()) - } - - def stopActorsGracefully(actors: ActorRef*): Unit = { - val gracefulStopFutures = actors.flatMap{ - actor => - Option(actor) match { - case Some(actorRef) => Some(Patterns.gracefulStop(actorRef, TestingUtils.TESTING_TIMEOUT)) - case None => None - } - } - - implicit val executionContext = defaultExecutionContext - - val globalStopFuture = scala.concurrent.Future.sequence(gracefulStopFutures) - - Await.result(globalStopFuture, TestingUtils.TESTING_TIMEOUT) - } - - def stopActorsGracefully(actors: java.util.List[ActorRef]): Unit = { - import scala.collection.JavaConverters._ - - stopActorsGracefully(actors.asScala: _*) - } - - def stopActorGatewaysGracefully(actorGateways: ActorGateway*): Unit = { - val actors = actorGateways.flatMap { - actorGateway => - Option(actorGateway) match { - case Some(actorGateway) => Some(actorGateway.actor()) - case None => None - } - } - - stopActorsGracefully(actors: _*) - } - - def stopActorGatewaysGracefully(actorGateways: java.util.List[ActorGateway]): Unit = { - import scala.collection.JavaConverters._ - - stopActorGatewaysGracefully(actorGateways.asScala: _*) - } - - /** Creates a forwarding JobManager which sends all received message to the forwarding target. - * - * @param actorSystem The actor system to start the actor in. - * @param forwardingTarget Target to forward to. - * @param leaderId leader id for the actor gateway - * @param actorName Name for forwarding Actor - * @return - */ - def createForwardingActor( - actorSystem: ActorSystem, - forwardingTarget: ActorRef, - leaderId: UUID, - actorName: Option[String] = None) - : ActorGateway = { - - val actor = actorName match { - case Some(name) => - actorSystem.actorOf( - Props( - classOf[ForwardingActor], - forwardingTarget, - Option(leaderId)), - name - ) - case None => - actorSystem.actorOf( - Props( - classOf[ForwardingActor], - forwardingTarget, - Option(leaderId)) - ) - } - - new AkkaActorGateway(actor, leaderId) - } - - class ForwardingActor(val target: ActorRef, val leaderSessionID: Option[UUID]) - extends FlinkActor with LeaderSessionMessageFilter with LogMessages { - - /** Handle incoming messages - * - * @return - */ - override def handleMessage: Receive = { - case msg => target.forward(msg) - } - - override val log: Logger = Logger(getClass) - } - }