You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by az...@apache.org on 2020/04/14 09:14:44 UTC
[flink] branch master updated (9ef41a4 -> b91c059)
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 9ef41a4 [FLINK-16961][hive] Bump Netty to 4.1.44
new 280f344 [FLINK-15936] Harden TaskExecutorTest#testSlotAcceptance
new 2735fa7 [hotfix][coordination] Log if TaskExecutor#establishResourceManagerConnection fails
new b91c059 [hotifx][tests] Remove dead code in TaskExecutorTest
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../flink/runtime/taskexecutor/TaskExecutor.java | 14 +-
.../runtime/taskexecutor/TaskExecutorTest.java | 274 ++++++++++-----------
2 files changed, 139 insertions(+), 149 deletions(-)
[flink] 03/03: [hotifx][tests] Remove dead code in TaskExecutorTest
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b91c05953565a7b511ece8c8c0187a44ec9f22a1
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Thu Apr 9 11:44:31 2020 +0300
[hotifx][tests] Remove dead code in TaskExecutorTest
---
.../runtime/taskexecutor/TaskExecutorTest.java | 39 +---------------------
1 file changed, 1 insertion(+), 38 deletions(-)
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 f140667..4d8beb8 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
@@ -57,7 +57,6 @@ import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.TaskExecutorPartitionTracker;
import org.apache.flink.runtime.io.network.partition.TaskExecutorPartitionTrackerImpl;
import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
-import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
import org.apache.flink.runtime.jobmaster.AllocatedSlotInfo;
import org.apache.flink.runtime.jobmaster.AllocatedSlotReport;
import org.apache.flink.runtime.jobmaster.JMTMRegistrationSuccess;
@@ -1957,25 +1956,19 @@ public class TaskExecutorTest extends TestLogger {
.build());
jobManagerLeaderRetriever.notifyListener(jobMasterGateway.getAddress(), jobMasterGateway.getFencingToken().toUUID());
- return new TaskExecutorTestingContext(offerSlotsLatch, jobMasterGateway, jobLeaderService, taskSlotTable, taskExecutor);
+ return new TaskExecutorTestingContext(jobMasterGateway, taskSlotTable, taskExecutor);
}
private class TaskExecutorTestingContext implements AutoCloseable {
- private final OneShotLatch offerSlotsLatch;
private final TestingJobMasterGateway jobMasterGateway;
- private final JobLeaderService jobLeaderService;
private final TaskSlotTable taskSlotTable;
private final TestingTaskExecutor taskExecutor;
private TaskExecutorTestingContext(
- OneShotLatch offerSlotsLatch,
TestingJobMasterGateway jobMasterGateway,
- JobLeaderService jobLeaderService,
TaskSlotTable taskSlotTable,
TestingTaskExecutor taskExecutor) {
- this.offerSlotsLatch = offerSlotsLatch;
this.jobMasterGateway = jobMasterGateway;
- this.jobLeaderService = jobLeaderService;
this.taskSlotTable = taskSlotTable;
this.taskExecutor = taskExecutor;
}
@@ -1985,36 +1978,6 @@ public class TaskExecutorTest extends TestLogger {
taskExecutor.waitUntilStarted();
}
- private void startAllocateSlotAndSubmit(
- final Class<? extends AbstractInvokable> task) throws Exception {
- final AllocationID allocationId = new AllocationID();
-
- start();
-
- taskSlotTable.allocateSlot(0, jobId, allocationId, Time.milliseconds(10000L));
-
- // we have to add the job after the TaskExecutor, because otherwise the service has not
- // been properly started.
- jobLeaderService.addJob(jobId, jobMasterGateway.getAddress());
- offerSlotsLatch.await();
-
- taskExecutor
- .getSelfGateway(TaskExecutorGateway.class)
- .submitTask(
- createTaskDeploymentDescriptor(allocationId, task),
- jobMasterGateway.getFencingToken(),
- timeout);
- }
-
- private TaskDeploymentDescriptor createTaskDeploymentDescriptor(
- final AllocationID allocationId,
- final Class<? extends AbstractInvokable> task) throws IOException {
- return TaskDeploymentDescriptorBuilder
- .newBuilder(jobId, task)
- .setAllocationId(allocationId)
- .build();
- }
-
@Override
public void close() throws ExecutionException, InterruptedException, TimeoutException {
RpcUtils.terminateRpcEndpoint(taskExecutor, timeout);
[flink] 02/03: [hotfix][coordination] Log if
TaskExecutor#establishResourceManagerConnection fails
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2735fa7715a8bfd20702f624ad881201c40f0d06
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Wed Apr 8 10:31:23 2020 +0300
[hotfix][coordination] Log if TaskExecutor#establishResourceManagerConnection fails
---
.../apache/flink/runtime/taskexecutor/TaskExecutor.java | 14 +++++++++-----
1 file changed, 9 insertions(+), 5 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
index 2b60ab0..5d8f40a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
@@ -1802,11 +1802,15 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
// filter out outdated connections
//noinspection ObjectEquality
if (resourceManagerConnection == connection) {
- establishResourceManagerConnection(
- resourceManagerGateway,
- resourceManagerId,
- taskExecutorRegistrationId,
- clusterInformation);
+ try {
+ establishResourceManagerConnection(
+ resourceManagerGateway,
+ resourceManagerId,
+ taskExecutorRegistrationId,
+ clusterInformation);
+ } catch (Throwable t) {
+ log.error("Establishing Resource Manager connection in Task Executor failed", t);
+ }
}
});
}
[flink] 01/03: [FLINK-15936] Harden
TaskExecutorTest#testSlotAcceptance
Posted by az...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
azagrebin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 280f344db45d84ad72b59f7f91a75a8875a368e7
Author: Andrey Zagrebin <az...@apache.org>
AuthorDate: Wed Apr 8 10:17:58 2020 +0300
[FLINK-15936] Harden TaskExecutorTest#testSlotAcceptance
The test called taskSlotTable.allocateSlot from the test main thread,
concurrently with taskSlotTable.createSlotReport while trying to register
RM in the main TM thread. This silently failed the RM registration in TM.runAsync.
As a result, RM.notifySlotAvailable was not called in TM.
The commmit refactors the test to wait properly for RM registration
and allocate slots through gateway in TM thread.
The commit also uses proper testing RM/JM instead of mocks.
To verify, the test has been looped locally 30k times.
This closes #11667.
---
.../runtime/taskexecutor/TaskExecutorTest.java | 235 +++++++++++----------
1 file changed, 129 insertions(+), 106 deletions(-)
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 df42ca6..f140667 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
@@ -88,6 +88,7 @@ import org.apache.flink.runtime.state.TaskExecutorLocalStateStoresManager;
import org.apache.flink.runtime.taskexecutor.TaskSubmissionTestEnvironment.Builder;
import org.apache.flink.runtime.taskexecutor.exceptions.RegistrationTimeoutException;
import org.apache.flink.runtime.taskexecutor.exceptions.TaskManagerException;
+import org.apache.flink.runtime.taskexecutor.exceptions.TaskSubmissionException;
import org.apache.flink.runtime.taskexecutor.partition.ClusterPartitionReport;
import org.apache.flink.runtime.taskexecutor.slot.SlotNotFoundException;
import org.apache.flink.runtime.taskexecutor.slot.SlotOffer;
@@ -135,6 +136,7 @@ import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledExecutorService;
@@ -156,7 +158,6 @@ import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;
import static org.hamcrest.Matchers.startsWith;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
@@ -798,89 +799,80 @@ public class TaskExecutorTest extends TestLogger {
*/
@Test
public void testSlotAcceptance() throws Exception {
- final TaskSlotTable<Task> taskSlotTable = TaskSlotUtils.createTaskSlotTable(2);
- final JobManagerTable jobManagerTable = new JobManagerTable();
- final JobLeaderService jobLeaderService = new JobLeaderService(unresolvedTaskManagerLocation, RetryingRegistrationConfiguration.defaultConfiguration());
-
- final String resourceManagerAddress = "rm";
- final UUID resourceManagerLeaderId = UUID.randomUUID();
-
- final String jobManagerAddress = "jm";
- final UUID jobManagerLeaderId = UUID.randomUUID();
-
- resourceManagerLeaderRetriever.notifyListener(resourceManagerAddress, resourceManagerLeaderId);
- jobManagerLeaderRetriever.notifyListener(jobManagerAddress, jobManagerLeaderId);
-
- final TestingResourceManagerGateway resourceManagerGateway = new TestingResourceManagerGateway();
- final ResourceID resourceManagerResourceId = resourceManagerGateway.getOwnResourceId();
final InstanceID registrationId = new InstanceID();
-
- final CompletableFuture<ResourceID> registrationFuture = new CompletableFuture<>();
- resourceManagerGateway.setRegisterTaskExecutorFunction(
- taskExecutorRegistration -> {
- registrationFuture.complete(taskExecutorRegistration.getResourceId());
- return CompletableFuture.completedFuture(new TaskExecutorRegistrationSuccess(registrationId, resourceManagerResourceId, new ClusterInformation("localhost", 1234)));
- });
-
+ final OneShotLatch taskExecutorIsRegistered = new OneShotLatch();
final CompletableFuture<Tuple3<InstanceID, SlotID, AllocationID>> availableSlotFuture = new CompletableFuture<>();
- resourceManagerGateway.setNotifySlotAvailableConsumer(availableSlotFuture::complete);
-
- final ResourceID jmResourceId = new ResourceID(jobManagerAddress);
+ final TestingResourceManagerGateway resourceManagerGateway =
+ createRmWithTmRegisterAndNotifySlotHooks(registrationId, taskExecutorIsRegistered, availableSlotFuture);
final AllocationID allocationId1 = new AllocationID();
final AllocationID allocationId2 = new AllocationID();
final SlotOffer offer1 = new SlotOffer(allocationId1, 0, ResourceProfile.ANY);
- final JobMasterGateway jobMasterGateway = mock(JobMasterGateway.class);
-
- when(jobMasterGateway.registerTaskManager(
- any(String.class),
- eq(unresolvedTaskManagerLocation),
- any(Time.class)
- )).thenReturn(CompletableFuture.completedFuture(new JMTMRegistrationSuccess(jmResourceId)));
- when(jobMasterGateway.getHostname()).thenReturn(jobManagerAddress);
-
- when(jobMasterGateway.offerSlots(
- any(ResourceID.class), any(Collection.class), any(Time.class)))
- .thenReturn(CompletableFuture.completedFuture((Collection<SlotOffer>) Collections.singleton(offer1)));
-
- rpc.registerGateway(resourceManagerAddress, resourceManagerGateway);
- rpc.registerGateway(jobManagerAddress, jobMasterGateway);
-
- final TaskExecutorLocalStateStoresManager localStateStoresManager = createTaskExecutorLocalStateStoresManager();
+ final OneShotLatch offerSlotsLatch = new OneShotLatch();
+ final OneShotLatch taskInTerminalState = new OneShotLatch();
+ final CompletableFuture<Collection<SlotOffer>> offerResultFuture = new CompletableFuture<>();
+ final TestingJobMasterGateway jobMasterGateway =
+ createJobMasterWithSlotOfferAndTaskTerminationHooks(offerSlotsLatch, taskInTerminalState, offerResultFuture);
- final TaskManagerServices taskManagerServices = new TaskManagerServicesBuilder()
- .setUnresolvedTaskManagerLocation(unresolvedTaskManagerLocation)
- .setTaskSlotTable(taskSlotTable)
- .setJobManagerTable(jobManagerTable)
- .setJobLeaderService(jobLeaderService)
- .setTaskStateManager(localStateStoresManager)
- .build();
+ rpc.registerGateway(resourceManagerGateway.getAddress(), resourceManagerGateway);
+ rpc.registerGateway(jobMasterGateway.getAddress(), jobMasterGateway);
- TaskExecutor taskManager = createTaskExecutor(taskManagerServices);
+ final TaskSlotTable<Task> taskSlotTable = TaskSlotUtils.createTaskSlotTable(2);
+ final TaskManagerServices taskManagerServices = createTaskManagerServicesWithTaskSlotTable(taskSlotTable);
+ final TestingTaskExecutor taskManager = createTestingTaskExecutor(taskManagerServices);
try {
taskManager.start();
+ taskManager.waitUntilStarted();
- assertThat(registrationFuture.get(), equalTo(unresolvedTaskManagerLocation.getResourceID()));
+ final TaskExecutorGateway tmGateway = taskManager.getSelfGateway(TaskExecutorGateway.class);
- taskSlotTable.allocateSlot(0, jobId, allocationId1, Time.milliseconds(10000L));
- taskSlotTable.allocateSlot(1, jobId, allocationId2, Time.milliseconds(10000L));
+ // wait until registered at the RM
+ taskExecutorIsRegistered.await();
- // we have to add the job after the TaskExecutor, because otherwise the service has not
- // been properly started.
- jobLeaderService.addJob(jobId, jobManagerAddress);
+ // request 2 slots for the given allocation ids
+ requestSlots(
+ tmGateway,
+ Arrays.asList(allocationId1, allocationId2),
+ resourceManagerGateway.getFencingToken(),
+ jobMasterGateway.getAddress());
+
+ // notify job leader to start slot offering
+ jobManagerLeaderRetriever.notifyListener(jobMasterGateway.getAddress(), jobMasterGateway.getFencingToken().toUUID());
+
+ // wait until slots have been offered
+ offerSlotsLatch.await();
+ offerResultFuture.complete(Collections.singletonList(offer1));
final Tuple3<InstanceID, SlotID, AllocationID> instanceIDSlotIDAllocationIDTuple3 = availableSlotFuture.get();
final Tuple3<InstanceID, SlotID, AllocationID> expectedResult = Tuple3.of(registrationId, new SlotID(unresolvedTaskManagerLocation.getResourceID(), 1), allocationId2);
assertThat(instanceIDSlotIDAllocationIDTuple3, equalTo(expectedResult));
-
- assertTrue(taskSlotTable.tryMarkSlotActive(jobId, allocationId1));
- assertFalse(taskSlotTable.tryMarkSlotActive(jobId, allocationId2));
- assertTrue(taskSlotTable.isSlotFree(1));
+ // the slot 1 can be activate for task submission
+ submitNoOpInvokableTask(allocationId1, jobMasterGateway, tmGateway);
+ // wait for the task completion
+ taskInTerminalState.await();
+ // the slot 2 can NOT be activate for task submission
+ try {
+ submitNoOpInvokableTask(allocationId2, jobMasterGateway, tmGateway);
+ fail("It should not be possible to submit task to acquired by JM slot with index 1 (allocationId2)");
+ } catch (CompletionException e) {
+ assertThat(e.getCause(), instanceOf(TaskSubmissionException.class));
+ }
+ // the slot 2 is free to request
+ tmGateway
+ .requestSlot(
+ new SlotID(unresolvedTaskManagerLocation.getResourceID(), 1),
+ jobId,
+ allocationId2,
+ ResourceProfile.UNKNOWN,
+ jobMasterGateway.getAddress(),
+ resourceManagerGateway.getFencingToken(),
+ timeout)
+ .join();
} finally {
RpcUtils.terminateRpcEndpoint(taskManager, timeout);
}
@@ -891,21 +883,11 @@ public class TaskExecutorTest extends TestLogger {
*/
@Test
public void testSubmitTaskBeforeAcceptSlot() throws Exception {
- final TaskSlotTable<Task> taskSlotTable = TaskSlotUtils.createTaskSlotTable(2);
- final JobManagerTable jobManagerTable = new JobManagerTable();
- final JobLeaderService jobLeaderService = new JobLeaderService(unresolvedTaskManagerLocation, RetryingRegistrationConfiguration.defaultConfiguration());
-
- final TestingResourceManagerGateway resourceManagerGateway = new TestingResourceManagerGateway();
- resourceManagerLeaderRetriever.notifyListener(resourceManagerGateway.getAddress(), resourceManagerGateway.getFencingToken().toUUID());
-
- final CompletableFuture<Tuple3<InstanceID, SlotID, AllocationID>> availableSlotFuture = new CompletableFuture<>();
- resourceManagerGateway.setNotifySlotAvailableConsumer(availableSlotFuture::complete);
-
+ final InstanceID registrationId = new InstanceID();
final OneShotLatch taskExecutorIsRegistered = new OneShotLatch();
- resourceManagerGateway.setSendSlotReportFunction(ignored -> {
- taskExecutorIsRegistered.trigger();
- return CompletableFuture.completedFuture(Acknowledge.get());
- });
+ final CompletableFuture<Tuple3<InstanceID, SlotID, AllocationID>> availableSlotFuture = new CompletableFuture<>();
+ final TestingResourceManagerGateway resourceManagerGateway =
+ createRmWithTmRegisterAndNotifySlotHooks(registrationId, taskExecutorIsRegistered, availableSlotFuture);
final AllocationID allocationId1 = new AllocationID();
final AllocationID allocationId2 = new AllocationID();
@@ -915,33 +897,14 @@ public class TaskExecutorTest extends TestLogger {
final OneShotLatch offerSlotsLatch = new OneShotLatch();
final OneShotLatch taskInTerminalState = new OneShotLatch();
final CompletableFuture<Collection<SlotOffer>> offerResultFuture = new CompletableFuture<>();
- final TestingJobMasterGateway jobMasterGateway = new TestingJobMasterGatewayBuilder()
- .setOfferSlotsFunction((resourceID, slotOffers) -> {
- offerSlotsLatch.trigger();
- return offerResultFuture;
- })
- .setUpdateTaskExecutionStateFunction(taskExecutionState -> {
- if (taskExecutionState.getExecutionState().isTerminal()) {
- taskInTerminalState.trigger();
- }
- return CompletableFuture.completedFuture(Acknowledge.get());
- })
- .build();
+ final TestingJobMasterGateway jobMasterGateway =
+ createJobMasterWithSlotOfferAndTaskTerminationHooks(offerSlotsLatch, taskInTerminalState, offerResultFuture);
rpc.registerGateway(resourceManagerGateway.getAddress(), resourceManagerGateway);
rpc.registerGateway(jobMasterGateway.getAddress(), jobMasterGateway);
- final TaskExecutorLocalStateStoresManager localStateStoresManager = createTaskExecutorLocalStateStoresManager();
-
- final TaskManagerServices taskManagerServices = new TaskManagerServicesBuilder()
- .setUnresolvedTaskManagerLocation(unresolvedTaskManagerLocation)
- .setShuffleEnvironment(nettyShuffleEnvironment)
- .setTaskSlotTable(taskSlotTable)
- .setJobLeaderService(jobLeaderService)
- .setJobManagerTable(jobManagerTable)
- .setTaskStateManager(localStateStoresManager)
- .build();
-
+ final TaskSlotTable<Task> taskSlotTable = TaskSlotUtils.createTaskSlotTable(2);
+ final TaskManagerServices taskManagerServices = createTaskManagerServicesWithTaskSlotTable(taskSlotTable);
final TestingTaskExecutor taskManager = createTestingTaskExecutor(taskManagerServices);
try {
@@ -966,13 +929,7 @@ public class TaskExecutorTest extends TestLogger {
// wait until slots have been offered
offerSlotsLatch.await();
- final TaskDeploymentDescriptor tdd = TaskDeploymentDescriptorBuilder
- .newBuilder(jobId, NoOpInvokable.class)
- .setAllocationId(allocationId1)
- .build();
-
- // submit the task without having acknowledge the offered slots
- tmGateway.submitTask(tdd, jobMasterGateway.getFencingToken(), timeout).get();
+ submitNoOpInvokableTask(allocationId1, jobMasterGateway, tmGateway);
// acknowledge the offered slots
offerResultFuture.complete(Collections.singleton(offer1));
@@ -988,6 +945,61 @@ public class TaskExecutorTest extends TestLogger {
}
}
+ private TestingResourceManagerGateway createRmWithTmRegisterAndNotifySlotHooks(
+ InstanceID registrationId,
+ OneShotLatch taskExecutorIsRegistered,
+ CompletableFuture<Tuple3<InstanceID, SlotID, AllocationID>> availableSlotFuture) {
+ final TestingResourceManagerGateway resourceManagerGateway = new TestingResourceManagerGateway();
+ resourceManagerLeaderRetriever.notifyListener(
+ resourceManagerGateway.getAddress(),
+ resourceManagerGateway.getFencingToken().toUUID());
+
+ resourceManagerGateway.setRegisterTaskExecutorFunction(
+ taskExecutorRegistration -> CompletableFuture.completedFuture(
+ new TaskExecutorRegistrationSuccess(registrationId, resourceManagerGateway.getOwnResourceId(),
+ new ClusterInformation("localhost", 1234))));
+
+ resourceManagerGateway.setNotifySlotAvailableConsumer(availableSlotFuture::complete);
+
+ resourceManagerGateway.setSendSlotReportFunction(ignored -> {
+ taskExecutorIsRegistered.trigger();
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ });
+ return resourceManagerGateway;
+ }
+
+ private TaskManagerServices createTaskManagerServicesWithTaskSlotTable(
+ TaskSlotTable<Task> taskSlotTable) throws IOException {
+ return new TaskManagerServicesBuilder()
+ .setUnresolvedTaskManagerLocation(unresolvedTaskManagerLocation)
+ .setShuffleEnvironment(nettyShuffleEnvironment)
+ .setTaskSlotTable(taskSlotTable)
+ .setJobLeaderService(new JobLeaderService(
+ unresolvedTaskManagerLocation,
+ RetryingRegistrationConfiguration.defaultConfiguration()))
+ .setJobManagerTable(new JobManagerTable())
+ .setTaskStateManager(createTaskExecutorLocalStateStoresManager())
+ .build();
+ }
+
+ private static TestingJobMasterGateway createJobMasterWithSlotOfferAndTaskTerminationHooks(
+ OneShotLatch offerSlotsLatch,
+ OneShotLatch taskInTerminalState,
+ CompletableFuture<Collection<SlotOffer>> offerResultFuture) {
+ return new TestingJobMasterGatewayBuilder()
+ .setOfferSlotsFunction((resourceID, slotOffers) -> {
+ offerSlotsLatch.trigger();
+ return offerResultFuture;
+ })
+ .setUpdateTaskExecutionStateFunction(taskExecutionState -> {
+ if (taskExecutionState.getExecutionState().isTerminal()) {
+ taskInTerminalState.trigger();
+ }
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .build();
+ }
+
private void requestSlots(TaskExecutorGateway tmGateway, Iterable<? extends AllocationID> allocationIds, ResourceManagerId resourceManagerId, String jobMasterGatewayAddress) {
int slotIndex = 0;
for (AllocationID allocationId : allocationIds) {
@@ -1005,6 +1017,17 @@ public class TaskExecutorTest extends TestLogger {
}
}
+ private void submitNoOpInvokableTask(
+ AllocationID allocationId,
+ TestingJobMasterGateway jobMasterGateway,
+ TaskExecutorGateway tmGateway) throws IOException {
+ final TaskDeploymentDescriptor tdd = TaskDeploymentDescriptorBuilder
+ .newBuilder(jobId, NoOpInvokable.class)
+ .setAllocationId(allocationId)
+ .build();
+ tmGateway.submitTask(tdd, jobMasterGateway.getFencingToken(), timeout).join();
+ }
+
/**
* This tests makes sure that duplicate JobMaster gained leadership messages are filtered out
* by the TaskExecutor. See FLINK-7526.