You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/10/15 20:08:46 UTC
[flink] branch release-1.6 updated (5b5d953 -> b64cbd6)
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a change to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 5b5d953 [FLINK-10135][metrics] Expose missing cluster-level metris again
new 2b44b33 [hotfix] Fix checkstyle violations in TaskExecutorTest
new a24b1b8 [hotfix] Introduce TestingJobMasterGatewayBuilder
new c60fdc4 [FLINK-9932] Harden slot allocation protocol
new b64cbd6 [FLINK-9932] Add TaskExecutorTest#testOfferSlotToJobMasterAfterTimeout
The 4 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:
.../function/TriConsumer.java} | 23 +-
.../function/TriFunction.java} | 25 +-
.../flink/runtime/taskexecutor/TaskExecutor.java | 32 +-
.../runtime/taskexecutor/slot/TaskSlotTable.java | 10 +-
.../jobmaster/utils/TestingJobMasterGateway.java | 219 ++++++++++++--
.../utils/TestingJobMasterGatewayBuilder.java | 253 ++++++++++++++++
.../runtime/taskexecutor/TaskExecutorTest.java | 323 +++++++++++----------
.../taskexecutor/slot/TaskSlotTableTest.java | 95 ++++++
.../taskexecutor/slot/TestingSlotActions.java} | 45 +--
.../slot/TestingSlotActionsBuilder.java} | 34 ++-
10 files changed, 799 insertions(+), 260 deletions(-)
copy flink-core/src/main/java/org/apache/flink/{api/common/distributions/RangeBoundaries.java => util/function/TriConsumer.java} (66%)
copy flink-core/src/main/java/org/apache/flink/{api/common/distributions/RangeBoundaries.java => util/function/TriFunction.java} (62%)
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGatewayBuilder.java
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/slot/TaskSlotTableTest.java
copy flink-runtime/src/{main/java/org/apache/flink/runtime/taskexecutor/slot/SlotActions.java => test/java/org/apache/flink/runtime/taskexecutor/slot/TestingSlotActions.java} (53%)
copy flink-runtime/src/{main/java/org/apache/flink/runtime/taskexecutor/slot/SlotActions.java => test/java/org/apache/flink/runtime/taskexecutor/slot/TestingSlotActionsBuilder.java} (55%)
[flink] 02/04: [hotfix] Introduce TestingJobMasterGatewayBuilder
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a24b1b8505be350e10782cc76cc4ddae9698f475
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Mon Oct 15 13:20:43 2018 +0200
[hotfix] Introduce TestingJobMasterGatewayBuilder
---
.../apache/flink/util/function/TriConsumer.java | 42 ++++
.../apache/flink/util/function/TriFunction.java | 44 ++++
.../jobmaster/utils/TestingJobMasterGateway.java | 219 +++++++++++++++---
.../utils/TestingJobMasterGatewayBuilder.java | 253 +++++++++++++++++++++
.../runtime/taskexecutor/TaskExecutorTest.java | 56 ++---
5 files changed, 545 insertions(+), 69 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/util/function/TriConsumer.java b/flink-core/src/main/java/org/apache/flink/util/function/TriConsumer.java
new file mode 100644
index 0000000..181a9bd
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/util/function/TriConsumer.java
@@ -0,0 +1,42 @@
+/*
+ * 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.util.function;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/**
+ * Operation which is performed on three given arguments.
+ *
+ * @param <S> type of the first argument
+ * @param <T> type of the second argument
+ * @param <U> type of the third argument
+ */
+@PublicEvolving
+@FunctionalInterface
+public interface TriConsumer<S, T, U> {
+
+ /**
+ * Performs this operation on the given arguments.
+ *
+ * @param s first argument
+ * @param t second argument
+ * @param u third argument
+ */
+ void accept(S s, T t, U u);
+}
diff --git a/flink-core/src/main/java/org/apache/flink/util/function/TriFunction.java b/flink-core/src/main/java/org/apache/flink/util/function/TriFunction.java
new file mode 100644
index 0000000..7e1e967
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/util/function/TriFunction.java
@@ -0,0 +1,44 @@
+/*
+ * 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.util.function;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/**
+ * Function which takes three arguments.
+ *
+ * @param <S> type of the first argument
+ * @param <T> type of the second argument
+ * @param <U> type of the third argument
+ * @param <R> type of the return value
+ */
+@PublicEvolving
+@FunctionalInterface
+public interface TriFunction<S, T, U, R> {
+
+ /**
+ * Applies this function to the given arguments.
+ *
+ * @param s the first function argument
+ * @param t the second function argument
+ * @param u the third function argument
+ * @return the function result
+ */
+ R apply(S s, T t, U u);
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
index e887fc1..c9c55a1 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGateway.java
@@ -20,6 +20,9 @@ package org.apache.flink.runtime.jobmaster.utils;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.tuple.Tuple5;
+import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.queryablestate.KvStateID;
import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
@@ -48,165 +51,321 @@ import org.apache.flink.runtime.taskexecutor.AccumulatorReport;
import org.apache.flink.runtime.taskexecutor.slot.SlotOffer;
import org.apache.flink.runtime.taskmanager.TaskExecutionState;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+import org.apache.flink.util.function.TriConsumer;
+import org.apache.flink.util.function.TriFunction;
+import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
/**
* {@link JobMasterGateway} implementation for testing purposes.
*/
public class TestingJobMasterGateway implements JobMasterGateway {
+ @Nonnull
+ private final String address;
+
+ @Nonnull
+ private final String hostname;
+
+ @Nonnull
+ private final Supplier<CompletableFuture<Acknowledge>> cancelFunction;
+
+ @Nonnull
+ private final Supplier<CompletableFuture<Acknowledge>> stopFunction;
+
+ @Nonnull
+ private final BiFunction<Integer, RescalingBehaviour, CompletableFuture<Acknowledge>> rescalingJobFunction;
+
+ @Nonnull
+ private final TriFunction<Collection<JobVertexID>, Integer, RescalingBehaviour, CompletableFuture<Acknowledge>> rescalingOperatorsFunction;
+
+ @Nonnull
+ private final Function<TaskExecutionState, CompletableFuture<Acknowledge>> updateTaskExecutionStateFunction;
+
+ @Nonnull
+ private final BiFunction<JobVertexID, ExecutionAttemptID, CompletableFuture<SerializedInputSplit>> requestNextInputSplitFunction;
+
+ @Nonnull
+ private final BiFunction<IntermediateDataSetID, ResultPartitionID, CompletableFuture<ExecutionState>> requestPartitionStateFunction;
+
+ @Nonnull
+ private final Function<ResultPartitionID, CompletableFuture<Acknowledge>> scheduleOrUpdateConsumersFunction;
+
+ @Nonnull
+ private final Function<ResourceID, CompletableFuture<Acknowledge>> disconnectTaskManagerFunction;
+
+ @Nonnull
+ private final Consumer<ResourceManagerId> disconnectResourceManagerConsumer;
+
+ @Nonnull
+ private final Supplier<CompletableFuture<ClassloadingProps>> classloadingPropsSupplier;
+
+ @Nonnull
+ private final BiFunction<ResourceID, Collection<SlotOffer>, CompletableFuture<Collection<SlotOffer>>> offerSlotsFunction;
+
+ @Nonnull
+ private final TriConsumer<ResourceID, AllocationID, Throwable> failSlotConsumer;
+
+ @Nonnull
+ private final BiFunction<String, TaskManagerLocation, CompletableFuture<RegistrationResponse>> registerTaskManagerFunction;
+
+ @Nonnull
+ private final BiConsumer<ResourceID, AccumulatorReport> taskManagerHeartbeatConsumer;
+
+ @Nonnull
+ private final Consumer<ResourceID> resourceManagerHeartbeatConsumer;
+
+ @Nonnull
+ private final Supplier<CompletableFuture<JobDetails>> requestJobDetailsSupplier;
+
+ @Nonnull
+ private final Supplier<CompletableFuture<ArchivedExecutionGraph>> requestJobSupplier;
+
+ @Nonnull
+ private final BiFunction<String, Boolean, CompletableFuture<String>> triggerSavepointFunction;
+
+ @Nonnull
+ private final Function<JobVertexID, CompletableFuture<OperatorBackPressureStatsResponse>> requestOperatorBackPressureStatsFunction;
+
+ @Nonnull
+ private final BiConsumer<AllocationID, Throwable> notifyAllocationFailureConsumer;
+
+ @Nonnull
+ private final Consumer<Tuple5<JobID, ExecutionAttemptID, Long, CheckpointMetrics, TaskStateSnapshot>> acknowledgeCheckpointConsumer;
+
+ @Nonnull
+ private final Consumer<Tuple4<JobID, ExecutionAttemptID, Long, Throwable>> declineCheckpointConsumer;
+
+ @Nonnull
+ private final Supplier<JobMasterId> fencingTokenSupplier;
+
+ @Nonnull
+ private final BiFunction<JobID, String, CompletableFuture<KvStateLocation>> requestKvStateLocationFunction;
+
+ @Nonnull
+ private final Function<Tuple6<JobID, JobVertexID, KeyGroupRange, String, KvStateID, InetSocketAddress>, CompletableFuture<Acknowledge>> notifyKvStateRegisteredFunction;
+
+ @Nonnull
+ private final Function<Tuple4<JobID, JobVertexID, KeyGroupRange, String>, CompletableFuture<Acknowledge>> notifyKvStateUnregisteredFunction;
+
+ public TestingJobMasterGateway(
+ @Nonnull String address,
+ @Nonnull String hostname,
+ @Nonnull Supplier<CompletableFuture<Acknowledge>> cancelFunction,
+ @Nonnull Supplier<CompletableFuture<Acknowledge>> stopFunction,
+ @Nonnull BiFunction<Integer, RescalingBehaviour, CompletableFuture<Acknowledge>> rescalingJobFunction,
+ @Nonnull TriFunction<Collection<JobVertexID>, Integer, RescalingBehaviour, CompletableFuture<Acknowledge>> rescalingOperatorsFunction,
+ @Nonnull Function<TaskExecutionState, CompletableFuture<Acknowledge>> updateTaskExecutionStateFunction,
+ @Nonnull BiFunction<JobVertexID, ExecutionAttemptID, CompletableFuture<SerializedInputSplit>> requestNextInputSplitFunction,
+ @Nonnull BiFunction<IntermediateDataSetID, ResultPartitionID, CompletableFuture<ExecutionState>> requestPartitionStateFunction,
+ @Nonnull Function<ResultPartitionID, CompletableFuture<Acknowledge>> scheduleOrUpdateConsumersFunction,
+ @Nonnull Function<ResourceID, CompletableFuture<Acknowledge>> disconnectTaskManagerFunction,
+ @Nonnull Consumer<ResourceManagerId> disconnectResourceManagerConsumer,
+ @Nonnull Supplier<CompletableFuture<ClassloadingProps>> classloadingPropsSupplier,
+ @Nonnull BiFunction<ResourceID, Collection<SlotOffer>, CompletableFuture<Collection<SlotOffer>>> offerSlotsFunction,
+ @Nonnull TriConsumer<ResourceID, AllocationID, Throwable> failSlotConsumer,
+ @Nonnull BiFunction<String, TaskManagerLocation, CompletableFuture<RegistrationResponse>> registerTaskManagerFunction,
+ @Nonnull BiConsumer<ResourceID, AccumulatorReport> taskManagerHeartbeatConsumer,
+ @Nonnull Consumer<ResourceID> resourceManagerHeartbeatConsumer,
+ @Nonnull Supplier<CompletableFuture<JobDetails>> requestJobDetailsSupplier,
+ @Nonnull Supplier<CompletableFuture<ArchivedExecutionGraph>> requestJobSupplier,
+ @Nonnull BiFunction<String, Boolean, CompletableFuture<String>> triggerSavepointFunction,
+ @Nonnull Function<JobVertexID, CompletableFuture<OperatorBackPressureStatsResponse>> requestOperatorBackPressureStatsFunction,
+ @Nonnull BiConsumer<AllocationID, Throwable> notifyAllocationFailureConsumer,
+ @Nonnull Consumer<Tuple5<JobID, ExecutionAttemptID, Long, CheckpointMetrics, TaskStateSnapshot>> acknowledgeCheckpointConsumer,
+ @Nonnull Consumer<Tuple4<JobID, ExecutionAttemptID, Long, Throwable>> declineCheckpointConsumer,
+ @Nonnull Supplier<JobMasterId> fencingTokenSupplier,
+ @Nonnull BiFunction<JobID, String, CompletableFuture<KvStateLocation>> requestKvStateLocationFunction,
+ @Nonnull Function<Tuple6<JobID, JobVertexID, KeyGroupRange, String, KvStateID, InetSocketAddress>, CompletableFuture<Acknowledge>> notifyKvStateRegisteredFunction,
+ @Nonnull Function<Tuple4<JobID, JobVertexID, KeyGroupRange, String>, CompletableFuture<Acknowledge>> notifyKvStateUnregisteredFunction) {
+ this.address = address;
+ this.hostname = hostname;
+ this.cancelFunction = cancelFunction;
+ this.stopFunction = stopFunction;
+ this.rescalingJobFunction = rescalingJobFunction;
+ this.rescalingOperatorsFunction = rescalingOperatorsFunction;
+ this.updateTaskExecutionStateFunction = updateTaskExecutionStateFunction;
+ this.requestNextInputSplitFunction = requestNextInputSplitFunction;
+ this.requestPartitionStateFunction = requestPartitionStateFunction;
+ this.scheduleOrUpdateConsumersFunction = scheduleOrUpdateConsumersFunction;
+ this.disconnectTaskManagerFunction = disconnectTaskManagerFunction;
+ this.disconnectResourceManagerConsumer = disconnectResourceManagerConsumer;
+ this.classloadingPropsSupplier = classloadingPropsSupplier;
+ this.offerSlotsFunction = offerSlotsFunction;
+ this.failSlotConsumer = failSlotConsumer;
+ this.registerTaskManagerFunction = registerTaskManagerFunction;
+ this.taskManagerHeartbeatConsumer = taskManagerHeartbeatConsumer;
+ this.resourceManagerHeartbeatConsumer = resourceManagerHeartbeatConsumer;
+ this.requestJobDetailsSupplier = requestJobDetailsSupplier;
+ this.requestJobSupplier = requestJobSupplier;
+ this.triggerSavepointFunction = triggerSavepointFunction;
+ this.requestOperatorBackPressureStatsFunction = requestOperatorBackPressureStatsFunction;
+ this.notifyAllocationFailureConsumer = notifyAllocationFailureConsumer;
+ this.acknowledgeCheckpointConsumer = acknowledgeCheckpointConsumer;
+ this.declineCheckpointConsumer = declineCheckpointConsumer;
+ this.fencingTokenSupplier = fencingTokenSupplier;
+ this.requestKvStateLocationFunction = requestKvStateLocationFunction;
+ this.notifyKvStateRegisteredFunction = notifyKvStateRegisteredFunction;
+ this.notifyKvStateUnregisteredFunction = notifyKvStateUnregisteredFunction;
+ }
+
@Override
public CompletableFuture<Acknowledge> cancel(Time timeout) {
- throw new UnsupportedOperationException();
+ return cancelFunction.get();
}
@Override
public CompletableFuture<Acknowledge> stop(Time timeout) {
- throw new UnsupportedOperationException();
+ return stopFunction.get();
}
@Override
public CompletableFuture<Acknowledge> rescaleJob(int newParallelism, RescalingBehaviour rescalingBehaviour, Time timeout) {
- throw new UnsupportedOperationException();
+ return rescalingJobFunction.apply(newParallelism, rescalingBehaviour);
}
@Override
public CompletableFuture<Acknowledge> rescaleOperators(Collection<JobVertexID> operators, int newParallelism, RescalingBehaviour rescalingBehaviour, Time timeout) {
- throw new UnsupportedOperationException();
+ return rescalingOperatorsFunction.apply(operators, newParallelism, rescalingBehaviour);
}
@Override
public CompletableFuture<Acknowledge> updateTaskExecutionState(TaskExecutionState taskExecutionState) {
- throw new UnsupportedOperationException();
+ return updateTaskExecutionStateFunction.apply(taskExecutionState);
}
@Override
public CompletableFuture<SerializedInputSplit> requestNextInputSplit(JobVertexID vertexID, ExecutionAttemptID executionAttempt) {
- throw new UnsupportedOperationException();
+ return requestNextInputSplitFunction.apply(vertexID, executionAttempt);
}
@Override
public CompletableFuture<ExecutionState> requestPartitionState(IntermediateDataSetID intermediateResultId, ResultPartitionID partitionId) {
- throw new UnsupportedOperationException();
+ return requestPartitionStateFunction.apply(intermediateResultId, partitionId);
}
@Override
public CompletableFuture<Acknowledge> scheduleOrUpdateConsumers(ResultPartitionID partitionID, Time timeout) {
- throw new UnsupportedOperationException();
+ return scheduleOrUpdateConsumersFunction.apply(partitionID);
}
@Override
public CompletableFuture<Acknowledge> disconnectTaskManager(ResourceID resourceID, Exception cause) {
- throw new UnsupportedOperationException();
+ return disconnectTaskManagerFunction.apply(resourceID);
}
@Override
public void disconnectResourceManager(ResourceManagerId resourceManagerId, Exception cause) {
- throw new UnsupportedOperationException();
+ disconnectResourceManagerConsumer.accept(resourceManagerId);
}
@Override
public CompletableFuture<ClassloadingProps> requestClassloadingProps() {
- throw new UnsupportedOperationException();
+ return classloadingPropsSupplier.get();
}
@Override
public CompletableFuture<Collection<SlotOffer>> offerSlots(ResourceID taskManagerId, Collection<SlotOffer> slots, Time timeout) {
- throw new UnsupportedOperationException();
+ return offerSlotsFunction.apply(taskManagerId, slots);
}
@Override
public void failSlot(ResourceID taskManagerId, AllocationID allocationId, Exception cause) {
- throw new UnsupportedOperationException();
+ failSlotConsumer.accept(taskManagerId, allocationId, cause);
}
@Override
public CompletableFuture<RegistrationResponse> registerTaskManager(String taskManagerRpcAddress, TaskManagerLocation taskManagerLocation, Time timeout) {
- throw new UnsupportedOperationException();
+ return registerTaskManagerFunction.apply(taskManagerRpcAddress, taskManagerLocation);
}
@Override
public void heartbeatFromTaskManager(ResourceID resourceID, AccumulatorReport accumulatorReport) {
- throw new UnsupportedOperationException();
+ taskManagerHeartbeatConsumer.accept(resourceID, accumulatorReport);
}
@Override
public void heartbeatFromResourceManager(ResourceID resourceID) {
- throw new UnsupportedOperationException();
+ resourceManagerHeartbeatConsumer.accept(resourceID);
}
@Override
public CompletableFuture<JobDetails> requestJobDetails(Time timeout) {
- throw new UnsupportedOperationException();
+ return requestJobDetailsSupplier.get();
}
@Override
public CompletableFuture<JobStatus> requestJobStatus(Time timeout) {
- throw new UnsupportedOperationException();
+ return requestJobDetailsSupplier.get().thenApply(JobDetails::getStatus);
}
@Override
public CompletableFuture<ArchivedExecutionGraph> requestJob(Time timeout) {
- throw new UnsupportedOperationException();
+ return requestJobSupplier.get();
}
@Override
public CompletableFuture<String> triggerSavepoint(@Nullable final String targetDirectory, final boolean cancelJob, final Time timeout) {
- throw new UnsupportedOperationException();
+ return triggerSavepointFunction.apply(targetDirectory, cancelJob);
}
@Override
public CompletableFuture<OperatorBackPressureStatsResponse> requestOperatorBackPressureStats(JobVertexID jobVertexId) {
- throw new UnsupportedOperationException();
+ return requestOperatorBackPressureStatsFunction.apply(jobVertexId);
}
@Override
public void notifyAllocationFailure(AllocationID allocationID, Exception cause) {
- throw new UnsupportedOperationException();
+ notifyAllocationFailureConsumer.accept(allocationID, cause);
}
@Override
public void acknowledgeCheckpoint(JobID jobID, ExecutionAttemptID executionAttemptID, long checkpointId, CheckpointMetrics checkpointMetrics, TaskStateSnapshot subtaskState) {
- throw new UnsupportedOperationException();
+ acknowledgeCheckpointConsumer.accept(Tuple5.of(jobID, executionAttemptID, checkpointId, checkpointMetrics, subtaskState));
}
@Override
public void declineCheckpoint(JobID jobID, ExecutionAttemptID executionAttemptID, long checkpointId, Throwable cause) {
- throw new UnsupportedOperationException();
+ declineCheckpointConsumer.accept(Tuple4.of(jobID, executionAttemptID, checkpointId, cause));
}
@Override
public JobMasterId getFencingToken() {
- throw new UnsupportedOperationException();
+ return fencingTokenSupplier.get();
}
@Override
public CompletableFuture<KvStateLocation> requestKvStateLocation(JobID jobId, String registrationName) {
- throw new UnsupportedOperationException();
+ return requestKvStateLocationFunction.apply(jobId, registrationName);
}
@Override
public CompletableFuture<Acknowledge> notifyKvStateRegistered(JobID jobId, JobVertexID jobVertexId, KeyGroupRange keyGroupRange, String registrationName, KvStateID kvStateId, InetSocketAddress kvStateServerAddress) {
- throw new UnsupportedOperationException();
+ return notifyKvStateRegisteredFunction.apply(Tuple6.of(jobId, jobVertexId, keyGroupRange, registrationName, kvStateId, kvStateServerAddress));
}
@Override
public CompletableFuture<Acknowledge> notifyKvStateUnregistered(JobID jobId, JobVertexID jobVertexId, KeyGroupRange keyGroupRange, String registrationName) {
- throw new UnsupportedOperationException();
+ return notifyKvStateUnregisteredFunction.apply(Tuple4.of(jobId, jobVertexId, keyGroupRange, registrationName));
}
@Override
public String getAddress() {
- return null;
+ return address;
}
@Override
public String getHostname() {
- return null;
+ return hostname;
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGatewayBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGatewayBuilder.java
new file mode 100644
index 0000000..e40b752
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/utils/TestingJobMasterGatewayBuilder.java
@@ -0,0 +1,253 @@
+/*
+ * 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.jobmaster.utils;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.tuple.Tuple5;
+import org.apache.flink.api.java.tuple.Tuple6;
+import org.apache.flink.queryablestate.KvStateID;
+import org.apache.flink.runtime.checkpoint.CheckpointMetrics;
+import org.apache.flink.runtime.checkpoint.TaskStateSnapshot;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobmaster.JMTMRegistrationSuccess;
+import org.apache.flink.runtime.jobmaster.JobMasterId;
+import org.apache.flink.runtime.jobmaster.RescalingBehaviour;
+import org.apache.flink.runtime.jobmaster.SerializedInputSplit;
+import org.apache.flink.runtime.jobmaster.message.ClassloadingProps;
+import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.messages.webmonitor.JobDetails;
+import org.apache.flink.runtime.query.KvStateLocation;
+import org.apache.flink.runtime.query.UnknownKvStateLocation;
+import org.apache.flink.runtime.registration.RegistrationResponse;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
+import org.apache.flink.runtime.rest.handler.legacy.backpressure.OperatorBackPressureStatsResponse;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.taskexecutor.AccumulatorReport;
+import org.apache.flink.runtime.taskexecutor.slot.SlotOffer;
+import org.apache.flink.runtime.taskmanager.TaskExecutionState;
+import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+import org.apache.flink.util.function.TriConsumer;
+import org.apache.flink.util.function.TriFunction;
+
+import java.net.InetSocketAddress;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+/**
+ * Builder for {@link TestingJobMasterGateway}.
+ */
+public class TestingJobMasterGatewayBuilder {
+ private static final ResourceID RESOURCE_MANAGER_ID = ResourceID.generate();
+ private static final JobMasterId JOB_MASTER_ID = JobMasterId.generate();
+
+ private String address = "akka.tcp://flink@localhost:6130/user/jobmanager";
+ private String hostname = "localhost";
+ private Supplier<CompletableFuture<Acknowledge>> cancelFunction = () -> CompletableFuture.completedFuture(Acknowledge.get());
+ private Supplier<CompletableFuture<Acknowledge>> stopFunction = () -> CompletableFuture.completedFuture(Acknowledge.get());
+ private BiFunction<Integer, RescalingBehaviour, CompletableFuture<Acknowledge>> rescalingJobFunction = (ignoredA, ignoredB) -> CompletableFuture.completedFuture(Acknowledge.get());
+ private TriFunction<Collection<JobVertexID>, Integer, RescalingBehaviour, CompletableFuture<Acknowledge>> rescalingOperatorsFunction = (ignoredA, ignoredB, ignoredC) -> CompletableFuture.completedFuture(Acknowledge.get());
+ private Function<TaskExecutionState, CompletableFuture<Acknowledge>> updateTaskExecutionStateFunction = ignored -> CompletableFuture.completedFuture(Acknowledge.get());
+ private BiFunction<JobVertexID, ExecutionAttemptID, CompletableFuture<SerializedInputSplit>> requestNextInputSplitFunction = (ignoredA, ignoredB) -> CompletableFuture.completedFuture(new SerializedInputSplit(null));
+ private BiFunction<IntermediateDataSetID, ResultPartitionID, CompletableFuture<ExecutionState>> requestPartitionStateFunction = (ignoredA, ignoredB) -> CompletableFuture.completedFuture(ExecutionState.RUNNING);
+ private Function<ResultPartitionID, CompletableFuture<Acknowledge>> scheduleOrUpdateConsumersFunction = ignored -> CompletableFuture.completedFuture(Acknowledge.get());
+ private Function<ResourceID, CompletableFuture<Acknowledge>> disconnectTaskManagerFunction = ignored -> CompletableFuture.completedFuture(Acknowledge.get());
+ private Consumer<ResourceManagerId> disconnectResourceManagerConsumer = ignored -> {};
+ private Supplier<CompletableFuture<ClassloadingProps>> classloadingPropsSupplier = () -> CompletableFuture.completedFuture(new ClassloadingProps(6124, Collections.emptyList(), Collections.emptyList()));
+ private BiFunction<ResourceID, Collection<SlotOffer>, CompletableFuture<Collection<SlotOffer>>> offerSlotsFunction = (ignoredA, ignoredB) -> CompletableFuture.completedFuture(Collections.emptyList());
+ private TriConsumer<ResourceID, AllocationID, Throwable> failSlotConsumer = (ignoredA, ignoredB, ignoredC) -> {};
+ private BiFunction<String, TaskManagerLocation, CompletableFuture<RegistrationResponse>> registerTaskManagerFunction = (ignoredA, ignoredB) -> CompletableFuture.completedFuture(new JMTMRegistrationSuccess(RESOURCE_MANAGER_ID));
+ private BiConsumer<ResourceID, AccumulatorReport> taskManagerHeartbeatConsumer = (ignoredA, ignoredB) -> {};
+ private Consumer<ResourceID> resourceManagerHeartbeatConsumer = ignored -> {};
+ private Supplier<CompletableFuture<JobDetails>> requestJobDetailsSupplier = () -> FutureUtils.completedExceptionally(new UnsupportedOperationException());
+ private Supplier<CompletableFuture<ArchivedExecutionGraph>> requestJobSupplier = () -> FutureUtils.completedExceptionally(new UnsupportedOperationException());
+ private BiFunction<String, Boolean, CompletableFuture<String>> triggerSavepointFunction = (targetDirectory, ignoredB) -> CompletableFuture.completedFuture(targetDirectory != null ? targetDirectory : UUID.randomUUID().toString());
+ private Function<JobVertexID, CompletableFuture<OperatorBackPressureStatsResponse>> requestOperatorBackPressureStatsFunction = ignored -> CompletableFuture.completedFuture(OperatorBackPressureStatsResponse.of(null));
+ private BiConsumer<AllocationID, Throwable> notifyAllocationFailureConsumer = (ignoredA, ignoredB) -> {};
+ private Consumer<Tuple5<JobID, ExecutionAttemptID, Long, CheckpointMetrics, TaskStateSnapshot>> acknowledgeCheckpointConsumer = ignored -> {};
+ private Consumer<Tuple4<JobID, ExecutionAttemptID, Long, Throwable>> declineCheckpointConsumer = ignored -> {};
+ private Supplier<JobMasterId> fencingTokenSupplier = () -> JOB_MASTER_ID;
+ private BiFunction<JobID, String, CompletableFuture<KvStateLocation>> requestKvStateLocationFunction = (ignoredA, registrationName) -> FutureUtils.completedExceptionally(new UnknownKvStateLocation(registrationName));
+ private Function<Tuple6<JobID, JobVertexID, KeyGroupRange, String, KvStateID, InetSocketAddress>, CompletableFuture<Acknowledge>> notifyKvStateRegisteredFunction = ignored -> CompletableFuture.completedFuture(Acknowledge.get());
+ private Function<Tuple4<JobID, JobVertexID, KeyGroupRange, String>, CompletableFuture<Acknowledge>> notifyKvStateUnregisteredFunction = ignored -> CompletableFuture.completedFuture(Acknowledge.get());
+
+ public TestingJobMasterGatewayBuilder setAddress(String address) {
+ this.address = address;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setHostname(String hostname) {
+ this.hostname = hostname;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setCancelFunction(Supplier<CompletableFuture<Acknowledge>> cancelFunction) {
+ this.cancelFunction = cancelFunction;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setStopFunction(Supplier<CompletableFuture<Acknowledge>> stopFunction) {
+ this.stopFunction = stopFunction;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setRescalingJobFunction(BiFunction<Integer, RescalingBehaviour, CompletableFuture<Acknowledge>> rescalingJobFunction) {
+ this.rescalingJobFunction = rescalingJobFunction;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setRescalingOperatorsFunction(TriFunction<Collection<JobVertexID>, Integer, RescalingBehaviour, CompletableFuture<Acknowledge>> rescalingOperatorsFunction) {
+ this.rescalingOperatorsFunction = rescalingOperatorsFunction;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setUpdateTaskExecutionStateFunction(Function<TaskExecutionState, CompletableFuture<Acknowledge>> updateTaskExecutionStateFunction) {
+ this.updateTaskExecutionStateFunction = updateTaskExecutionStateFunction;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setRequestNextInputSplitFunction(BiFunction<JobVertexID, ExecutionAttemptID, CompletableFuture<SerializedInputSplit>> requestNextInputSplitFunction) {
+ this.requestNextInputSplitFunction = requestNextInputSplitFunction;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setRequestPartitionStateFunction(BiFunction<IntermediateDataSetID, ResultPartitionID, CompletableFuture<ExecutionState>> requestPartitionStateFunction) {
+ this.requestPartitionStateFunction = requestPartitionStateFunction;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setScheduleOrUpdateConsumersFunction(Function<ResultPartitionID, CompletableFuture<Acknowledge>> scheduleOrUpdateConsumersFunction) {
+ this.scheduleOrUpdateConsumersFunction = scheduleOrUpdateConsumersFunction;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setDisconnectTaskManagerFunction(Function<ResourceID, CompletableFuture<Acknowledge>> disconnectTaskManagerFunction) {
+ this.disconnectTaskManagerFunction = disconnectTaskManagerFunction;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setDisconnectResourceManagerConsumer(Consumer<ResourceManagerId> disconnectResourceManagerConsumer) {
+ this.disconnectResourceManagerConsumer = disconnectResourceManagerConsumer;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setClassloadingPropsSupplier(Supplier<CompletableFuture<ClassloadingProps>> classloadingPropsSupplier) {
+ this.classloadingPropsSupplier = classloadingPropsSupplier;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setOfferSlotsFunction(BiFunction<ResourceID, Collection<SlotOffer>, CompletableFuture<Collection<SlotOffer>>> offerSlotsFunction) {
+ this.offerSlotsFunction = offerSlotsFunction;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setFailSlotConsumer(TriConsumer<ResourceID, AllocationID, Throwable> failSlotConsumer) {
+ this.failSlotConsumer = failSlotConsumer;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setRegisterTaskManagerFunction(BiFunction<String, TaskManagerLocation, CompletableFuture<RegistrationResponse>> registerTaskManagerFunction) {
+ this.registerTaskManagerFunction = registerTaskManagerFunction;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setTaskManagerHeartbeatConsumer(BiConsumer<ResourceID, AccumulatorReport> taskManagerHeartbeatConsumer) {
+ this.taskManagerHeartbeatConsumer = taskManagerHeartbeatConsumer;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setResourceManagerHeartbeatConsumer(Consumer<ResourceID> resourceManagerHeartbeatConsumer) {
+ this.resourceManagerHeartbeatConsumer = resourceManagerHeartbeatConsumer;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setRequestJobDetailsSupplier(Supplier<CompletableFuture<JobDetails>> requestJobDetailsSupplier) {
+ this.requestJobDetailsSupplier = requestJobDetailsSupplier;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setRequestJobSupplier(Supplier<CompletableFuture<ArchivedExecutionGraph>> requestJobSupplier) {
+ this.requestJobSupplier = requestJobSupplier;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setTriggerSavepointFunction(BiFunction<String, Boolean, CompletableFuture<String>> triggerSavepointFunction) {
+ this.triggerSavepointFunction = triggerSavepointFunction;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setRequestOperatorBackPressureStatsFunction(Function<JobVertexID, CompletableFuture<OperatorBackPressureStatsResponse>> requestOperatorBackPressureStatsFunction) {
+ this.requestOperatorBackPressureStatsFunction = requestOperatorBackPressureStatsFunction;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setNotifyAllocationFailureConsumer(BiConsumer<AllocationID, Throwable> notifyAllocationFailureConsumer) {
+ this.notifyAllocationFailureConsumer = notifyAllocationFailureConsumer;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setAcknowledgeCheckpointConsumer(Consumer<Tuple5<JobID, ExecutionAttemptID, Long, CheckpointMetrics, TaskStateSnapshot>> acknowledgeCheckpointConsumer) {
+ this.acknowledgeCheckpointConsumer = acknowledgeCheckpointConsumer;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setDeclineCheckpointConsumer(Consumer<Tuple4<JobID, ExecutionAttemptID, Long, Throwable>> declineCheckpointConsumer) {
+ this.declineCheckpointConsumer = declineCheckpointConsumer;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setFencingTokenSupplier(Supplier<JobMasterId> fencingTokenSupplier) {
+ this.fencingTokenSupplier = fencingTokenSupplier;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setRequestKvStateLocationFunction(BiFunction<JobID, String, CompletableFuture<KvStateLocation>> requestKvStateLocationFunction) {
+ this.requestKvStateLocationFunction = requestKvStateLocationFunction;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setNotifyKvStateRegisteredFunction(Function<Tuple6<JobID, JobVertexID, KeyGroupRange, String, KvStateID, InetSocketAddress>, CompletableFuture<Acknowledge>> notifyKvStateRegisteredFunction) {
+ this.notifyKvStateRegisteredFunction = notifyKvStateRegisteredFunction;
+ return this;
+ }
+
+ public TestingJobMasterGatewayBuilder setNotifyKvStateUnregisteredFunction(Function<Tuple4<JobID, JobVertexID, KeyGroupRange, String>, CompletableFuture<Acknowledge>> notifyKvStateUnregisteredFunction) {
+ this.notifyKvStateUnregisteredFunction = notifyKvStateUnregisteredFunction;
+ return this;
+ }
+
+ public TestingJobMasterGateway build() {
+ return new TestingJobMasterGateway(address, hostname, cancelFunction, stopFunction, rescalingJobFunction, rescalingOperatorsFunction, updateTaskExecutionStateFunction, requestNextInputSplitFunction, requestPartitionStateFunction, scheduleOrUpdateConsumersFunction, disconnectTaskManagerFunction, disconnectResourceManagerConsumer, classloadingPropsSupplier, offerSlotsFunction, failSlotConsumer, registerTaskManagerFunction, taskManagerHeartbeatConsumer, resourceManagerHeartbeatConsumer, r [...]
+ }
+}
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 4ad02fc..a372a4f 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
@@ -64,6 +64,7 @@ import org.apache.flink.runtime.jobmaster.JMTMRegistrationSuccess;
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.LeaderRetrievalListener;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.leaderretrieval.SettableLeaderRetrievalService;
@@ -248,8 +249,20 @@ public class TaskExecutorTest extends TestLogger {
final UUID jmLeaderId = UUID.randomUUID();
final ResourceID jmResourceId = ResourceID.generate();
- final SimpleJobMasterGateway jobMasterGateway = new SimpleJobMasterGateway(
- CompletableFuture.completedFuture(new JMTMRegistrationSuccess(jmResourceId)));
+ final CompletableFuture<TaskManagerLocation> taskManagerLocationFuture = new CompletableFuture<>();
+ final CompletableFuture<ResourceID> disconnectTaskManagerFuture = new CompletableFuture<>();
+ final TestingJobMasterGateway jobMasterGateway = new TestingJobMasterGatewayBuilder()
+ .setRegisterTaskManagerFunction((s, taskManagerLocation) -> {
+ taskManagerLocationFuture.complete(taskManagerLocation);
+ return CompletableFuture.completedFuture(new JMTMRegistrationSuccess(jmResourceId));
+ })
+ .setDisconnectTaskManagerFunction(
+ resourceID -> {
+ disconnectTaskManagerFuture.complete(resourceID);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ }
+ )
+ .build();
TaskExecutorLocalStateStoresManager localStateStoresManager = new TaskExecutorLocalStateStoresManager(
false,
@@ -287,11 +300,11 @@ public class TaskExecutorTest extends TestLogger {
jobManagerLeaderRetriever.notifyListener(jobMasterAddress, jmLeaderId);
// register task manager success will trigger monitoring heartbeat target between tm and jm
- final TaskManagerLocation taskManagerLocation1 = jobMasterGateway.getRegisterTaskManagerFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+ final TaskManagerLocation taskManagerLocation1 = taskManagerLocationFuture.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
assertThat(taskManagerLocation1, equalTo(taskManagerLocation));
// the timeout should trigger disconnecting from the JobManager
- final ResourceID resourceID = jobMasterGateway.getDisconnectTaskManagerFuture().get(heartbeatTimeout * 50L, TimeUnit.MILLISECONDS);
+ final ResourceID resourceID = disconnectTaskManagerFuture.get(heartbeatTimeout * 50L, TimeUnit.MILLISECONDS);
assertThat(resourceID, equalTo(taskManagerLocation.getResourceID()));
} finally {
@@ -385,41 +398,6 @@ public class TaskExecutorTest extends TestLogger {
}
}
- private static final class SimpleJobMasterGateway extends TestingJobMasterGateway {
-
- private final CompletableFuture<TaskManagerLocation> registerTaskManagerFuture = new CompletableFuture<>();
-
- private final CompletableFuture<ResourceID> disconnectTaskManagerFuture = new CompletableFuture<>();
-
- private final CompletableFuture<RegistrationResponse> registerTaskManagerResponseFuture;
-
- private SimpleJobMasterGateway(CompletableFuture<RegistrationResponse> registerTaskManagerResponseFuture) {
- this.registerTaskManagerResponseFuture = registerTaskManagerResponseFuture;
- }
-
- @Override
- public CompletableFuture<RegistrationResponse> registerTaskManager(String taskManagerRpcAddress, TaskManagerLocation taskManagerLocation, Time timeout) {
- registerTaskManagerFuture.complete(taskManagerLocation);
-
- return registerTaskManagerResponseFuture;
- }
-
- @Override
- public CompletableFuture<Acknowledge> disconnectTaskManager(ResourceID resourceID, Exception cause) {
- disconnectTaskManagerFuture.complete(resourceID);
-
- return CompletableFuture.completedFuture(Acknowledge.get());
- }
-
- CompletableFuture<TaskManagerLocation> getRegisterTaskManagerFuture() {
- return registerTaskManagerFuture;
- }
-
- CompletableFuture<ResourceID> getDisconnectTaskManagerFuture() {
- return disconnectTaskManagerFuture;
- }
- }
-
/**
* Tests that the correct slot report is sent as part of the heartbeat response.
*/
[flink] 03/04: [FLINK-9932] Harden slot allocation protocol
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c60fdc4e0d1885a25145995fdc51b3a4acce7500
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Fri Oct 12 16:53:59 2018 +0200
[FLINK-9932] Harden slot allocation protocol
Harden slot allocation protocol by accepting task submissions for slots which are only
allocated. Before, it was necessary that the slot was marked as active. This, however,
required that task submissions come strictly after completing the slot offering which
is not the case. With this change, we mark all allocated and active slots as active
if the TaskExecutor receives a task submission.
---
.../flink/runtime/taskexecutor/TaskExecutor.java | 32 ++++---
.../runtime/taskexecutor/slot/TaskSlotTable.java | 10 +-
.../runtime/taskexecutor/TaskExecutorTest.java | 106 +++++++--------------
.../taskexecutor/slot/TaskSlotTableTest.java | 95 ++++++++++++++++++
.../taskexecutor/slot/TestingSlotActions.java | 54 +++++++++++
.../slot/TestingSlotActionsBuilder.java | 47 +++++++++
6 files changed, 255 insertions(+), 89 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 9352c47..f90e939 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
@@ -451,7 +451,7 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
throw new TaskSubmissionException(message);
}
- if (!taskSlotTable.existsActiveSlot(jobId, tdd.getAllocationId())) {
+ if (!taskSlotTable.tryMarkSlotActive(jobId, tdd.getAllocationId())) {
final String message = "No task slot allocated for job ID " + jobId +
" and allocation ID " + tdd.getAllocationId() + '.';
log.debug(message);
@@ -1062,18 +1062,6 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
while (reservedSlotsIterator.hasNext()) {
SlotOffer offer = reservedSlotsIterator.next().generateSlotOffer();
- try {
- if (!taskSlotTable.markSlotActive(offer.getAllocationId())) {
- // the slot is either free or releasing at the moment
- final String message = "Could not mark slot " + jobId + " active.";
- log.debug(message);
- jobMasterGateway.failSlot(getResourceID(), offer.getAllocationId(), new Exception(message));
- }
- } catch (SlotNotFoundException e) {
- final String message = "Could not mark slot " + jobId + " active.";
- jobMasterGateway.failSlot(getResourceID(), offer.getAllocationId(), new Exception(message));
- continue;
- }
reservedSlots.add(offer);
}
@@ -1103,6 +1091,24 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
if (isJobManagerConnectionValid(jobId, jobMasterId)) {
// mark accepted slots active
for (SlotOffer acceptedSlot : acceptedSlots) {
+ try {
+ if (!taskSlotTable.markSlotActive(acceptedSlot.getAllocationId())) {
+ // the slot is either free or releasing at the moment
+ final String message = "Could not mark slot " + jobId + " active.";
+ log.debug(message);
+ jobMasterGateway.failSlot(
+ getResourceID(),
+ acceptedSlot.getAllocationId(),
+ new FlinkException(message));
+ }
+ } catch (SlotNotFoundException e) {
+ final String message = "Could not mark slot " + jobId + " active.";
+ jobMasterGateway.failSlot(
+ getResourceID(),
+ acceptedSlot.getAllocationId(),
+ new FlinkException(message));
+ }
+
reservedSlots.remove(acceptedSlot);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/slot/TaskSlotTable.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/slot/TaskSlotTable.java
index c94e278..867203c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/slot/TaskSlotTable.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/slot/TaskSlotTable.java
@@ -379,17 +379,17 @@ public class TaskSlotTable implements TimeoutListener<AllocationID> {
}
/**
- * Check whether there exists an active slot for the given job and allocation id.
+ * Try to mark the specified slot as active if it has been allocated by the given job.
*
* @param jobId of the allocated slot
* @param allocationId identifying the allocation
- * @return True if there exists a task slot which is active for the given job and allocation id.
+ * @return True if the task slot could be marked active.
*/
- public boolean existsActiveSlot(JobID jobId, AllocationID allocationId) {
+ public boolean tryMarkSlotActive(JobID jobId, AllocationID allocationId) {
TaskSlot taskSlot = getTaskSlot(allocationId);
- if (taskSlot != null) {
- return taskSlot.isActive(jobId, allocationId);
+ if (taskSlot != null && taskSlot.isAllocated(jobId, allocationId)) {
+ return taskSlot.markActive();
} else {
return false;
}
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 a372a4f..550f31a 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
@@ -35,8 +35,6 @@ import org.apache.flink.runtime.clusterframework.types.SlotID;
import org.apache.flink.runtime.concurrent.Executors;
import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.concurrent.ScheduledExecutor;
-import org.apache.flink.runtime.deployment.InputGateDeploymentDescriptor;
-import org.apache.flink.runtime.deployment.ResultPartitionDeploymentDescriptor;
import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.execution.Environment;
@@ -69,9 +67,6 @@ import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalListener;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
import org.apache.flink.runtime.leaderretrieval.SettableLeaderRetrievalService;
import org.apache.flink.runtime.messages.Acknowledge;
-import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
-import org.apache.flink.runtime.metrics.groups.TaskManagerMetricGroup;
-import org.apache.flink.runtime.metrics.groups.TaskMetricGroup;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.query.TaskKvStateRegistry;
import org.apache.flink.runtime.registration.RegistrationResponse;
@@ -90,10 +85,10 @@ import org.apache.flink.runtime.taskexecutor.slot.TimerService;
import org.apache.flink.runtime.taskmanager.CheckpointResponder;
import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation;
import org.apache.flink.runtime.taskmanager.Task;
-import org.apache.flink.runtime.taskmanager.TaskExecutionState;
import org.apache.flink.runtime.taskmanager.TaskManagerActions;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
import org.apache.flink.runtime.testingUtils.TestingUtils;
+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;
@@ -728,7 +723,7 @@ public class TaskExecutorTest extends TestLogger {
jobManagerTable.put(jobId, jobManagerConnection);
final TaskSlotTable taskSlotTable = mock(TaskSlotTable.class);
- when(taskSlotTable.existsActiveSlot(eq(jobId), eq(allocationId))).thenReturn(true);
+ when(taskSlotTable.tryMarkSlotActive(eq(jobId), eq(allocationId))).thenReturn(true);
when(taskSlotTable.addTask(any(Task.class))).thenReturn(true);
TaskEventDispatcher taskEventDispatcher = new TaskEventDispatcher();
@@ -993,8 +988,8 @@ public class TaskExecutorTest extends TestLogger {
assertThat(instanceIDSlotIDAllocationIDTuple3, equalTo(expectedResult));
- assertTrue(taskSlotTable.existsActiveSlot(jobId, allocationId1));
- assertFalse(taskSlotTable.existsActiveSlot(jobId, allocationId2));
+ assertTrue(taskSlotTable.tryMarkSlotActive(jobId, allocationId1));
+ assertFalse(taskSlotTable.tryMarkSlotActive(jobId, allocationId2));
assertTrue(taskSlotTable.isSlotFree(1));
} finally {
taskManager.shutDown();
@@ -1011,61 +1006,37 @@ public class TaskExecutorTest extends TestLogger {
final JobManagerTable jobManagerTable = new JobManagerTable();
final JobLeaderService jobLeaderService = new JobLeaderService(taskManagerLocation);
- final String resourceManagerAddress = "rm";
- final UUID resourceManagerLeaderId = UUID.randomUUID();
-
- final String jobManagerAddress = "jm";
- final JobMasterId jobMasterId = JobMasterId.generate();
-
- resourceManagerLeaderRetriever.notifyListener(resourceManagerAddress, resourceManagerLeaderId);
- jobManagerLeaderRetriever.notifyListener(jobManagerAddress, jobMasterId.toUUID());
-
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 ResourceID jmResourceId = new ResourceID(jobManagerAddress);
-
final AllocationID allocationId1 = new AllocationID();
final AllocationID allocationId2 = new AllocationID();
final SlotOffer offer1 = new SlotOffer(allocationId1, 0, ResourceProfile.UNKNOWN);
- final JobMasterGateway jobMasterGateway = mock(JobMasterGateway.class);
-
- when(jobMasterGateway.registerTaskManager(
- any(String.class),
- eq(taskManagerLocation),
- any(Time.class)
- )).thenReturn(CompletableFuture.completedFuture(new JMTMRegistrationSuccess(jmResourceId)));
- when(jobMasterGateway.getHostname()).thenReturn(jobManagerAddress);
- when(jobMasterGateway.updateTaskExecutionState(any(TaskExecutionState.class))).thenReturn(CompletableFuture.completedFuture(Acknowledge.get()));
-
- rpc.registerGateway(resourceManagerAddress, resourceManagerGateway);
- rpc.registerGateway(jobManagerAddress, jobMasterGateway);
-
- final LibraryCacheManager libraryCacheManager = mock(LibraryCacheManager.class);
- when(libraryCacheManager.getClassLoader(eq(jobId))).thenReturn(getClass().getClassLoader());
-
- final JobManagerConnection jobManagerConnection = new JobManagerConnection(
- jobId,
- jmResourceId,
- jobMasterGateway,
- mock(TaskManagerActions.class),
- mock(CheckpointResponder.class),
- libraryCacheManager,
- mock(ResultPartitionConsumableNotifier.class),
- mock(PartitionProducerStateChecker.class));
+ 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 TaskManagerMetricGroup taskManagerMetricGroup = mock(TaskManagerMetricGroup.class);
- TaskMetricGroup taskMetricGroup = mock(TaskMetricGroup.class);
- when(taskMetricGroup.getIOMetricGroup()).thenReturn(mock(TaskIOMetricGroup.class));
+ jobManagerLeaderRetriever.notifyListener(jobMasterGateway.getAddress(), jobMasterGateway.getFencingToken().toUUID());
- when(taskManagerMetricGroup.addTaskForJob(
- any(JobID.class), anyString(), any(JobVertexID.class), any(ExecutionAttemptID.class),
- anyString(), anyInt(), anyInt())
- ).thenReturn(taskMetricGroup);
+ rpc.registerGateway(resourceManagerGateway.getAddress(), resourceManagerGateway);
+ rpc.registerGateway(jobMasterGateway.getAddress(), jobMasterGateway);
final NetworkEnvironment networkMock = mock(NetworkEnvironment.class, Mockito.RETURNS_MOCKS);
@@ -1089,7 +1060,7 @@ public class TaskExecutorTest extends TestLogger {
haServices,
taskManagerServices,
new HeartbeatServices(1000L, 1000L),
- taskManagerMetricGroup,
+ UnregisteredMetricGroups.createUnregisteredTaskManagerMetricGroup(),
null,
dummyBlobCacheService,
testingFatalErrorHandler);
@@ -1117,7 +1088,7 @@ public class TaskExecutorTest extends TestLogger {
"test task",
1,
1,
- TestInvokable.class.getName(),
+ NoOpInvokable.class.getName(),
new Configuration());
SerializedValue<JobInformation> serializedJobInformation = new SerializedValue<>(jobInformation);
@@ -1133,27 +1104,17 @@ public class TaskExecutorTest extends TestLogger {
0,
0,
null,
- Collections.<ResultPartitionDeploymentDescriptor>emptyList(),
- Collections.<InputGateDeploymentDescriptor>emptyList());
-
- CompletableFuture<Collection<SlotOffer>> offerResultFuture = new CompletableFuture<>();
-
- // submit task first and then return acceptance response
- when(
- jobMasterGateway.offerSlots(
- any(ResourceID.class),
- any(Collection.class),
- any(Time.class)))
- .thenReturn(offerResultFuture);
+ Collections.emptyList(),
+ Collections.emptyList());
// we have to add the job after the TaskExecutor, because otherwise the service has not
// been properly started. This will also offer the slots to the job master
- jobLeaderService.addJob(jobId, jobManagerAddress);
+ jobLeaderService.addJob(jobId, jobMasterGateway.getAddress());
- verify(jobMasterGateway, Mockito.timeout(timeout.toMilliseconds())).offerSlots(any(ResourceID.class), any(Collection.class), any(Time.class));
+ offerSlotsLatch.await();
// submit the task without having acknowledge the offered slots
- tmGateway.submitTask(tdd, jobMasterId, timeout);
+ tmGateway.submitTask(tdd, jobMasterGateway.getFencingToken(), timeout).get();
// acknowledge the offered slots
offerResultFuture.complete(Collections.singleton(offer1));
@@ -1161,9 +1122,12 @@ public class TaskExecutorTest extends TestLogger {
final Tuple3<InstanceID, SlotID, AllocationID> instanceIDSlotIDAllocationIDTuple3 = availableSlotFuture.get();
assertThat(instanceIDSlotIDAllocationIDTuple3.f1, equalTo(new SlotID(taskManagerLocation.getResourceID(), 1)));
- assertTrue(taskSlotTable.existsActiveSlot(jobId, allocationId1));
- assertFalse(taskSlotTable.existsActiveSlot(jobId, allocationId2));
+ assertTrue(taskSlotTable.tryMarkSlotActive(jobId, allocationId1));
+ assertFalse(taskSlotTable.tryMarkSlotActive(jobId, allocationId2));
assertTrue(taskSlotTable.isSlotFree(1));
+
+ // wait for the task completion
+ taskInTerminalState.await();
} finally {
taskManager.shutDown();
taskManager.getTerminationFuture().get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/slot/TaskSlotTableTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/slot/TaskSlotTableTest.java
new file mode 100644
index 0000000..ef3fc06
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/slot/TaskSlotTableTest.java
@@ -0,0 +1,95 @@
+/*
+ * 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.slot;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.flink.shaded.guava18.com.google.common.collect.Sets;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.junit.Test;
+
+import javax.annotation.Nonnull;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for the {@link TaskSlotTable}.
+ */
+public class TaskSlotTableTest extends TestLogger {
+
+ private static final Time SLOT_TIMEOUT = Time.seconds(100L);
+
+ /**
+ * Tests that one can can mark allocated slots as active.
+ */
+ @Test
+ public void testTryMarkSlotActive() throws SlotNotFoundException {
+ final TaskSlotTable taskSlotTable = createTaskSlotTable(Collections.nCopies(3, ResourceProfile.UNKNOWN));
+
+ try {
+ taskSlotTable.start(new TestingSlotActionsBuilder().build());
+
+ final JobID jobId1 = new JobID();
+ final AllocationID allocationId1 = new AllocationID();
+ taskSlotTable.allocateSlot(0, jobId1, allocationId1, SLOT_TIMEOUT);
+ final AllocationID allocationId2 = new AllocationID();
+ taskSlotTable.allocateSlot(1, jobId1, allocationId2, SLOT_TIMEOUT);
+ final AllocationID allocationId3 = new AllocationID();
+ final JobID jobId2 = new JobID();
+ taskSlotTable.allocateSlot(2, jobId2, allocationId3, SLOT_TIMEOUT);
+
+ taskSlotTable.markSlotActive(allocationId1);
+
+ assertThat(taskSlotTable.isAllocated(0, jobId1, allocationId1), is(true));
+ assertThat(taskSlotTable.isAllocated(1, jobId1, allocationId2), is(true));
+ assertThat(taskSlotTable.isAllocated(2, jobId2, allocationId3), is(true));
+
+ assertThat(IteratorUtils.toList(taskSlotTable.getActiveSlots(jobId1)), is(equalTo(Arrays.asList(allocationId1))));
+
+ assertThat(taskSlotTable.tryMarkSlotActive(jobId1, allocationId1), is(true));
+ assertThat(taskSlotTable.tryMarkSlotActive(jobId1, allocationId2), is(true));
+ assertThat(taskSlotTable.tryMarkSlotActive(jobId1, allocationId3), is(false));
+
+ assertThat(Sets.newHashSet(taskSlotTable.getActiveSlots(jobId1)), is(equalTo(new HashSet<>(Arrays.asList(allocationId2, allocationId1)))));
+ } finally {
+ taskSlotTable.stop();
+ }
+ }
+
+ @Nonnull
+ private TaskSlotTable createTaskSlotTable(final Collection<ResourceProfile> resourceProfiles) {
+ return new TaskSlotTable(
+ resourceProfiles,
+ new TimerService<>(TestingUtils.defaultExecutor(), 10000L));
+ }
+
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/slot/TestingSlotActions.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/slot/TestingSlotActions.java
new file mode 100644
index 0000000..ad24189
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/slot/TestingSlotActions.java
@@ -0,0 +1,54 @@
+/*
+ * 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.slot;
+
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+
+import javax.annotation.Nonnull;
+
+import java.util.UUID;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+
+/**
+ * {@link SlotActions} implementation for testing purposes.
+ */
+public class TestingSlotActions implements SlotActions {
+
+ @Nonnull
+ private final Consumer<AllocationID> freeSlotConsumer;
+
+ @Nonnull
+ private final BiConsumer<AllocationID, UUID> timeoutSlotConsumer;
+
+ public TestingSlotActions(@Nonnull Consumer<AllocationID> freeSlotConsumer, @Nonnull BiConsumer<AllocationID, UUID> timeoutSlotConsumer) {
+ this.freeSlotConsumer = freeSlotConsumer;
+ this.timeoutSlotConsumer = timeoutSlotConsumer;
+ }
+
+ @Override
+ public void freeSlot(AllocationID allocationId) {
+ freeSlotConsumer.accept(allocationId);
+ }
+
+ @Override
+ public void timeoutSlot(AllocationID allocationId, UUID ticket) {
+ timeoutSlotConsumer.accept(allocationId, ticket);
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/slot/TestingSlotActionsBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/slot/TestingSlotActionsBuilder.java
new file mode 100644
index 0000000..f1bef55
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/slot/TestingSlotActionsBuilder.java
@@ -0,0 +1,47 @@
+/*
+ * 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.slot;
+
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+
+import java.util.UUID;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+
+/**
+ * Builder for the {@link TestingSlotActions}.
+ */
+public class TestingSlotActionsBuilder {
+ private Consumer<AllocationID> freeSlotConsumer = ignored -> {};
+ private BiConsumer<AllocationID, UUID> timeoutSlotConsumer = (ignoredA, ignoredB) -> {};
+
+ public TestingSlotActionsBuilder setFreeSlotConsumer(Consumer<AllocationID> freeSlotConsumer) {
+ this.freeSlotConsumer = freeSlotConsumer;
+ return this;
+ }
+
+ public TestingSlotActionsBuilder setTimeoutSlotConsumer(BiConsumer<AllocationID, UUID> timeoutSlotConsumer) {
+ this.timeoutSlotConsumer = timeoutSlotConsumer;
+ return this;
+ }
+
+ public TestingSlotActions build() {
+ return new TestingSlotActions(freeSlotConsumer, timeoutSlotConsumer);
+ }
+}
[flink] 01/04: [hotfix] Fix checkstyle violations in
TaskExecutorTest
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2b44b33ca2ed12e7988715d80858106beac75fce
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Mon Oct 15 11:27:39 2018 +0200
[hotfix] Fix checkstyle violations in TaskExecutorTest
---
.../runtime/taskexecutor/TaskExecutorTest.java | 90 ++++++++++------------
1 file changed, 42 insertions(+), 48 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 22256ee..4ad02fc 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
@@ -442,15 +442,15 @@ public class TaskExecutorTest extends TestLogger {
new ClusterInformation("localhost", 1234)));
rmGateway.setRegisterTaskExecutorFunction(stringResourceIDIntegerHardwareDescriptionTuple4 -> {
- taskExecutorRegistrationFuture.complete(stringResourceIDIntegerHardwareDescriptionTuple4.f1);
- return registrationResponse;
- });
+ taskExecutorRegistrationFuture.complete(stringResourceIDIntegerHardwareDescriptionTuple4.f1);
+ return registrationResponse;
+ });
final CompletableFuture<SlotReport> initialSlotReportFuture = new CompletableFuture<>();
rmGateway.setSendSlotReportFunction(resourceIDInstanceIDSlotReportTuple3 -> {
- initialSlotReportFuture.complete(resourceIDInstanceIDSlotReportTuple3.f2);
- return CompletableFuture.completedFuture(Acknowledge.get());
- });
+ initialSlotReportFuture.complete(resourceIDInstanceIDSlotReportTuple3.f2);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ });
final CompletableFuture<SlotReport> heartbeatSlotReportFuture = new CompletableFuture<>();
rmGateway.setTaskExecutorHeartbeatConsumer((resourceID, slotReport) -> heartbeatSlotReportFuture.complete(slotReport));
@@ -486,10 +486,10 @@ public class TaskExecutorTest extends TestLogger {
return spy(new HeartbeatManagerImpl<>(
heartbeatTimeout,
taskManagerLocation.getResourceID(),
- (HeartbeatListener<SlotReport, Void>)invocation.getArguments()[1],
- (Executor)invocation.getArguments()[2],
- (ScheduledExecutor)invocation.getArguments()[2],
- (Logger)invocation.getArguments()[3]));
+ (HeartbeatListener<SlotReport, Void>) invocation.getArguments()[1],
+ (Executor) invocation.getArguments()[2],
+ (ScheduledExecutor) invocation.getArguments()[2],
+ (Logger) invocation.getArguments()[3]));
}
}
);
@@ -673,10 +673,10 @@ public class TaskExecutorTest extends TestLogger {
eq(taskManagerAddress), eq(taskManagerLocation.getResourceID()), anyInt(), any(HardwareDescription.class), any(Time.class));
assertNotNull(taskManager.getResourceManagerConnection());
- // cancel the leader
+ // cancel the leader
resourceManagerLeaderRetriever.notifyListener(null, null);
- // set a new leader, see that a registration happens
+ // set a new leader, see that a registration happens
resourceManagerLeaderRetriever.notifyListener(address2, leaderId2);
verify(rmGateway2, Mockito.timeout(timeout.toMilliseconds())).registerTaskExecutor(
@@ -789,7 +789,7 @@ public class TaskExecutorTest extends TestLogger {
tmGateway.submitTask(tdd, jobMasterId, timeout);
- CompletableFuture<Boolean> completionFuture = TestInvokable.completableFuture;
+ CompletableFuture<Boolean> completionFuture = TestInvokable.COMPLETABLE_FUTURE;
completionFuture.get();
} finally {
@@ -803,7 +803,7 @@ public class TaskExecutorTest extends TestLogger {
*/
public static class TestInvokable extends AbstractInvokable {
- static final CompletableFuture<Boolean> completableFuture = new CompletableFuture<>();
+ static final CompletableFuture<Boolean> COMPLETABLE_FUTURE = new CompletableFuture<>();
public TestInvokable(Environment environment) {
super(environment);
@@ -811,7 +811,7 @@ public class TaskExecutorTest extends TestLogger {
@Override
public void invoke() throws Exception {
- completableFuture.complete(true);
+ COMPLETABLE_FUTURE.complete(true);
}
}
@@ -831,10 +831,9 @@ public class TaskExecutorTest extends TestLogger {
final TestingResourceManagerGateway resourceManagerGateway = new TestingResourceManagerGateway();
CompletableFuture<Void> initialSlotReportFuture = new CompletableFuture<>();
resourceManagerGateway.setSendSlotReportFunction(resourceIDInstanceIDSlotReportTuple3 -> {
- initialSlotReportFuture.complete(null);
- return CompletableFuture.completedFuture(Acknowledge.get());
-
- });
+ initialSlotReportFuture.complete(null);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ });
final String jobManagerAddress = "jm";
final UUID jobManagerLeaderId = UUID.randomUUID();
@@ -944,10 +943,9 @@ public class TaskExecutorTest extends TestLogger {
final CompletableFuture<ResourceID> registrationFuture = new CompletableFuture<>();
resourceManagerGateway.setRegisterTaskExecutorFunction(
stringResourceIDIntegerHardwareDescriptionTuple4 -> {
- registrationFuture.complete(stringResourceIDIntegerHardwareDescriptionTuple4.f1);
- return CompletableFuture.completedFuture(new TaskExecutorRegistrationSuccess(registrationId, resourceManagerResourceId, 1000L, new ClusterInformation("localhost", 1234)));
- }
- );
+ registrationFuture.complete(stringResourceIDIntegerHardwareDescriptionTuple4.f1);
+ return CompletableFuture.completedFuture(new TaskExecutorRegistrationSuccess(registrationId, resourceManagerResourceId, 1000L, new ClusterInformation("localhost", 1234)));
+ });
final CompletableFuture<Tuple3<InstanceID, SlotID, AllocationID>> availableSlotFuture = new CompletableFuture<>();
resourceManagerGateway.setNotifySlotAvailableConsumer(availableSlotFuture::complete);
@@ -970,7 +968,7 @@ public class TaskExecutorTest extends TestLogger {
when(jobMasterGateway.offerSlots(
any(ResourceID.class), any(Collection.class), any(Time.class)))
- .thenReturn(CompletableFuture.completedFuture((Collection<SlotOffer>)Collections.singleton(offer1)));
+ .thenReturn(CompletableFuture.completedFuture((Collection<SlotOffer>) Collections.singleton(offer1)));
rpc.registerGateway(resourceManagerAddress, resourceManagerGateway);
rpc.registerGateway(jobManagerAddress, jobMasterGateway);
@@ -1066,7 +1064,6 @@ public class TaskExecutorTest extends TestLogger {
when(jobMasterGateway.getHostname()).thenReturn(jobManagerAddress);
when(jobMasterGateway.updateTaskExecutionState(any(TaskExecutionState.class))).thenReturn(CompletableFuture.completedFuture(Acknowledge.get()));
-
rpc.registerGateway(resourceManagerAddress, resourceManagerGateway);
rpc.registerGateway(jobManagerAddress, jobMasterGateway);
@@ -1197,9 +1194,7 @@ public class TaskExecutorTest extends TestLogger {
/**
* This tests makes sure that duplicate JobMaster gained leadership messages are filtered out
- * by the TaskExecutor.
- *
- * See FLINK-7526
+ * by the TaskExecutor. See FLINK-7526.
*/
@Test
public void testFilterOutDuplicateJobMasterRegistrations() throws Exception {
@@ -1370,9 +1365,9 @@ public class TaskExecutorTest extends TestLogger {
final TestingResourceManagerGateway resourceManagerGateway = new TestingResourceManagerGateway();
final CompletableFuture<Void> initialSlotReport = new CompletableFuture<>();
resourceManagerGateway.setSendSlotReportFunction(resourceIDInstanceIDSlotReportTuple3 -> {
- initialSlotReport.complete(null);
- return CompletableFuture.completedFuture(Acknowledge.get());
- });
+ initialSlotReport.complete(null);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ });
final ResourceManagerId resourceManagerId = resourceManagerGateway.getFencingToken();
rpc.registerGateway(resourceManagerGateway.getAddress(), resourceManagerGateway);
@@ -1519,9 +1514,9 @@ public class TaskExecutorTest extends TestLogger {
final CompletableFuture<ResourceID> taskExecutorResourceIdFuture = new CompletableFuture<>();
testingResourceManagerGateway.setRegisterTaskExecutorFunction(stringResourceIDSlotReportIntegerHardwareDescriptionTuple5 -> {
- taskExecutorResourceIdFuture.complete(stringResourceIDSlotReportIntegerHardwareDescriptionTuple5.f1);
- return registrationFuture;
- });
+ taskExecutorResourceIdFuture.complete(stringResourceIDSlotReportIntegerHardwareDescriptionTuple5.f1);
+ return registrationFuture;
+ });
rpc.registerGateway(testingResourceManagerGateway.getAddress(), testingResourceManagerGateway);
resourceManagerLeaderRetriever.notifyListener(testingResourceManagerGateway.getAddress(), testingResourceManagerGateway.getFencingToken().toUUID());
@@ -1624,10 +1619,9 @@ public class TaskExecutorTest extends TestLogger {
testingResourceManagerGateway.setSendSlotReportFunction(
resourceIDInstanceIDSlotReportTuple3 -> {
- initialSlotReportFuture.complete(resourceIDInstanceIDSlotReportTuple3.f0);
- return CompletableFuture.completedFuture(Acknowledge.get());
- }
- );
+ initialSlotReportFuture.complete(resourceIDInstanceIDSlotReportTuple3.f0);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ });
rpc.registerGateway(testingResourceManagerGateway.getAddress(), testingResourceManagerGateway);
resourceManagerLeaderRetriever.notifyListener(testingResourceManagerGateway.getAddress(), testingResourceManagerGateway.getFencingToken().toUUID());
@@ -1672,7 +1666,7 @@ public class TaskExecutorTest extends TestLogger {
new InstanceID(),
testingResourceManagerGateway.getOwnResourceId(),
1000L,
- new ClusterInformation("foobar", 1234) ));
+ new ClusterInformation("foobar", 1234)));
final CountDownLatch numberRegistrations = new CountDownLatch(2);
@@ -1700,15 +1694,15 @@ public class TaskExecutorTest extends TestLogger {
@Nonnull
private TaskExecutor createTaskExecutor(TaskManagerServices taskManagerServices) {
return new TaskExecutor(
- rpc,
- TaskManagerConfiguration.fromConfiguration(configuration),
- haServices,
- taskManagerServices,
- new HeartbeatServices(1000L, 1000L),
- UnregisteredMetricGroups.createUnregisteredTaskManagerMetricGroup(),
- null,
- dummyBlobCacheService,
- testingFatalErrorHandler);
+ rpc,
+ TaskManagerConfiguration.fromConfiguration(configuration),
+ haServices,
+ taskManagerServices,
+ new HeartbeatServices(1000L, 1000L),
+ UnregisteredMetricGroups.createUnregisteredTaskManagerMetricGroup(),
+ null,
+ dummyBlobCacheService,
+ testingFatalErrorHandler);
}
private static final class StartStopNotifyingLeaderRetrievalService implements LeaderRetrievalService {
[flink] 04/04: [FLINK-9932] Add
TaskExecutorTest#testOfferSlotToJobMasterAfterTimeout
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b64cbd6dfe4445aa45f9f8982de1ec83cea9c040
Author: shuai-xu <sh...@foxmail.com>
AuthorDate: Mon Oct 15 16:55:06 2018 +0200
[FLINK-9932] Add TaskExecutorTest#testOfferSlotToJobMasterAfterTimeout
---
.../runtime/taskexecutor/TaskExecutorTest.java | 71 ++++++++++++++++++++++
1 file changed, 71 insertions(+)
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 550f31a..179dea2 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
@@ -123,6 +123,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.function.Function;
import static org.hamcrest.Matchers.contains;
@@ -1633,6 +1634,76 @@ public class TaskExecutorTest extends TestLogger {
}
}
+ /**
+ * Tests that offers slots to job master timeout and retry.
+ */
+ @Test
+ public void testOfferSlotToJobMasterAfterTimeout() throws Exception {
+ final TaskSlotTable taskSlotTable = new TaskSlotTable(
+ Arrays.asList(ResourceProfile.UNKNOWN, ResourceProfile.UNKNOWN),
+ timerService);
+ final TaskManagerServices taskManagerServices = new TaskManagerServicesBuilder()
+ .setTaskSlotTable(taskSlotTable)
+ .build();
+
+ final TaskExecutor taskExecutor = createTaskExecutor(taskManagerServices);
+
+ final AllocationID allocationId = new AllocationID();
+
+ final CompletableFuture<ResourceID> initialSlotReportFuture = new CompletableFuture<>();
+
+ final TestingResourceManagerGateway testingResourceManagerGateway = new TestingResourceManagerGateway();
+ testingResourceManagerGateway.setSendSlotReportFunction(resourceIDInstanceIDSlotReportTuple3 -> {
+ initialSlotReportFuture.complete(null);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+
+ });
+ rpc.registerGateway(testingResourceManagerGateway.getAddress(), testingResourceManagerGateway);
+ resourceManagerLeaderRetriever.notifyListener(testingResourceManagerGateway.getAddress(), testingResourceManagerGateway.getFencingToken().toUUID());
+
+ final CountDownLatch slotOfferings = new CountDownLatch(3);
+ final CompletableFuture<AllocationID> offeredSlotFuture = new CompletableFuture<>();
+ final TestingJobMasterGateway jobMasterGateway = new TestingJobMasterGatewayBuilder()
+ .setOfferSlotsFunction((resourceID, slotOffers) -> {
+ assertThat(slotOffers.size(), is(1));
+ slotOfferings.countDown();
+
+ if (slotOfferings.getCount() == 0) {
+ offeredSlotFuture.complete(slotOffers.iterator().next().getAllocationId());
+ return CompletableFuture.completedFuture(slotOffers);
+ } else {
+ return FutureUtils.completedExceptionally(new TimeoutException());
+ }
+ })
+ .build();
+ final String jobManagerAddress = jobMasterGateway.getAddress();
+ rpc.registerGateway(jobManagerAddress, jobMasterGateway);
+ jobManagerLeaderRetriever.notifyListener(jobManagerAddress, jobMasterGateway.getFencingToken().toUUID());
+
+ try {
+ taskExecutor.start();
+ final TaskExecutorGateway taskExecutorGateway = taskExecutor.getSelfGateway(TaskExecutorGateway.class);
+
+ // wait for the connection to the ResourceManager
+ initialSlotReportFuture.get();
+
+ taskExecutorGateway.requestSlot(
+ new SlotID(taskExecutor.getResourceID(), 0),
+ jobId,
+ allocationId,
+ jobManagerAddress,
+ testingResourceManagerGateway.getFencingToken(),
+ timeout).get();
+
+ slotOfferings.await();
+
+ assertThat(offeredSlotFuture.get(), is(allocationId));
+ assertTrue(taskSlotTable.isSlotFree(1));
+ } finally {
+ RpcUtils.terminateRpcEndpoint(taskExecutor, timeout);
+ }
+ }
+
@Nonnull
private TaskExecutor createTaskExecutor(TaskManagerServices taskManagerServices) {
return new TaskExecutor(