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 2017/11/29 13:56:12 UTC
[3/9] flink git commit: [FLINK-8150] [flip6] Expose TaskExecutor's
ResourceID as TaskExecutor id
[FLINK-8150] [flip6] Expose TaskExecutor's ResourceID as TaskExecutor id
Before, the TaskExecutor's InstanceID was exposed as TaskExecutor id. This was wrong
since the InstanceID is bound the registration of a TaskExecutor whereas the
ResourceID is bound to the lifetime of the TaskExecutor. Thus, it is better to identify
the TaskExecutor by its ResourceID which does not change.
This commit changes the behaviour accordingly on the ResourceManager and the
TaskManagerDetailsHandler.
This closes #5093.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/dc7ab135
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/dc7ab135
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/dc7ab135
Branch: refs/heads/master
Commit: dc7ab1355e5e84d50313eec50b930e09d6bd759f
Parents: 44d973d
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue Nov 28 12:43:39 2017 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Nov 29 14:52:53 2017 +0100
----------------------------------------------------------------------
.../resourcemanager/ResourceManager.java | 29 ++--
.../resourcemanager/ResourceManagerGateway.java | 4 +-
.../taskmanager/TaskManagerDetailsHandler.java | 4 +-
.../messages/json/InstanceIDDeserializer.java | 45 ------
.../messages/json/InstanceIDSerializer.java | 44 ------
.../messages/json/ResourceIDDeserializer.java | 44 ++++++
.../messages/json/ResourceIDSerializer.java | 45 ++++++
.../taskmanager/TaskManagerDetailsInfo.java | 10 +-
.../taskmanager/TaskManagerIdPathParameter.java | 13 +-
.../messages/taskmanager/TaskManagerInfo.java | 22 +--
.../clusterframework/ResourceManagerTest.java | 13 +-
.../resourcemanager/ResourceManagerTest.java | 153 +++++++++++++++++++
.../resourcemanager/TestingResourceManager.java | 76 +++++++++
.../utils/TestingResourceManagerGateway.java | 2 +-
.../taskmanager/TaskManagerInfoTest.java | 4 +-
.../TestingTaskExecutorGateway.java | 124 +++++++++++++++
16 files changed, 489 insertions(+), 143 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
index 3ba8808..4220ebf 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
@@ -500,14 +500,14 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
@Override
public CompletableFuture<Collection<TaskManagerInfo>> requestTaskManagerInfo(Time timeout) {
- ArrayList<TaskManagerInfo> taskManagerInfos = new ArrayList<>(taskExecutors.size());
+ final ArrayList<TaskManagerInfo> taskManagerInfos = new ArrayList<>(taskExecutors.size());
for (Map.Entry<ResourceID, WorkerRegistration<WorkerType>> taskExecutorEntry : taskExecutors.entrySet()) {
final WorkerRegistration<WorkerType> taskExecutor = taskExecutorEntry.getValue();
taskManagerInfos.add(
new TaskManagerInfo(
- taskExecutor.getInstanceID(),
+ taskExecutorEntry.getKey(),
taskExecutor.getTaskExecutorGateway().getAddress(),
taskExecutor.getDataPort(),
taskManagerHeartbeatManager.getLastHeartbeatFrom(taskExecutorEntry.getKey()),
@@ -520,29 +520,20 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
}
@Override
- public CompletableFuture<TaskManagerInfo> requestTaskManagerInfo(InstanceID instanceId, Time timeout) {
+ public CompletableFuture<TaskManagerInfo> requestTaskManagerInfo(ResourceID resourceId, Time timeout) {
- Map.Entry<ResourceID, WorkerRegistration<WorkerType>> taskExecutorEntry = null;
+ WorkerRegistration<WorkerType> taskExecutor = taskExecutors.get(resourceId);
- // TODO: Implement more efficiently
- for (Map.Entry<ResourceID, WorkerRegistration<WorkerType>> workerEntry : taskExecutors.entrySet()) {
- if (Objects.equals(workerEntry.getValue().getInstanceID(), instanceId)) {
- taskExecutorEntry = workerEntry;
- break;
- }
- }
-
- if (taskExecutorEntry == null) {
- return FutureUtils.completedExceptionally(new FlinkException("Requested TaskManager " + instanceId + " is not registered."));
+ if (taskExecutor == null) {
+ return FutureUtils.completedExceptionally(new FlinkException("Requested TaskManager " + resourceId + " is not registered."));
} else {
- WorkerRegistration<?> taskExecutor = taskExecutorEntry.getValue();
final TaskManagerInfo taskManagerInfo = new TaskManagerInfo(
- taskExecutor.getInstanceID(),
+ resourceId,
taskExecutor.getTaskExecutorGateway().getAddress(),
taskExecutor.getDataPort(),
- taskManagerHeartbeatManager.getLastHeartbeatFrom(taskExecutorEntry.getKey()),
- slotManager.getNumberRegisteredSlotsOf(instanceId),
- slotManager.getNumberFreeSlotsOf(instanceId),
+ taskManagerHeartbeatManager.getLastHeartbeatFrom(resourceId),
+ slotManager.getNumberRegisteredSlotsOf(taskExecutor.getInstanceID()),
+ slotManager.getNumberFreeSlotsOf(taskExecutor.getInstanceID()),
taskExecutor.getHardwareDescription());
return CompletableFuture.completedFuture(taskManagerInfo);
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java
index 88538ff..41a0794 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java
@@ -185,11 +185,11 @@ public interface ResourceManagerGateway extends FencedRpcGateway<ResourceManager
/**
* Requests information about the given {@link TaskExecutor}.
*
- * @param instanceId identifying the TaskExecutor for which to return information
+ * @param taskManagerId identifying the TaskExecutor for which to return information
* @param timeout of the request
* @return Future TaskManager information
*/
- CompletableFuture<TaskManagerInfo> requestTaskManagerInfo(InstanceID instanceId, @RpcTimeout Time timeout);
+ CompletableFuture<TaskManagerInfo> requestTaskManagerInfo(ResourceID taskManagerId, @RpcTimeout Time timeout);
/**
* Requests the resource overview. The resource overview provides information about the
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerDetailsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerDetailsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerDetailsHandler.java
index bca2961..b8c1a60 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerDetailsHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerDetailsHandler.java
@@ -19,7 +19,7 @@
package org.apache.flink.runtime.rest.handler.taskmanager;
import org.apache.flink.api.common.time.Time;
-import org.apache.flink.runtime.instance.InstanceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.RestHandlerException;
@@ -71,7 +71,7 @@ public class TaskManagerDetailsHandler<T extends RestfulGateway> extends Abstrac
protected CompletableFuture<TaskManagerDetailsInfo> handleRequest(
@Nonnull HandlerRequest<EmptyRequestBody, TaskManagerMessageParameters> request,
@Nonnull ResourceManagerGateway gateway) throws RestHandlerException {
- final InstanceID taskManagerInstanceId = request.getPathParameter(TaskManagerIdPathParameter.class);
+ final ResourceID taskManagerInstanceId = request.getPathParameter(TaskManagerIdPathParameter.class);
CompletableFuture<TaskManagerInfo> taskManagerInfoFuture = gateway.requestTaskManagerInfo(taskManagerInstanceId, timeout);
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/InstanceIDDeserializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/InstanceIDDeserializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/InstanceIDDeserializer.java
deleted file mode 100644
index 1c53dcd..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/InstanceIDDeserializer.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.rest.messages.json;
-
-import org.apache.flink.runtime.instance.InstanceID;
-import org.apache.flink.util.StringUtils;
-
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationContext;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.std.StdDeserializer;
-
-import java.io.IOException;
-
-/**
- * Json deserializer for {@link InstanceID}.
- */
-public class InstanceIDDeserializer extends StdDeserializer<InstanceID> {
-
- private static final long serialVersionUID = -9058463293913469849L;
-
- protected InstanceIDDeserializer() {
- super(InstanceID.class);
- }
-
- @Override
- public InstanceID deserialize(JsonParser p, DeserializationContext ctxt) throws IOException {
- return new InstanceID(StringUtils.hexStringToByte(p.getValueAsString()));
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/InstanceIDSerializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/InstanceIDSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/InstanceIDSerializer.java
deleted file mode 100644
index f3c0dc5..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/InstanceIDSerializer.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.rest.messages.json;
-
-import org.apache.flink.runtime.instance.InstanceID;
-
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.StdSerializer;
-
-import java.io.IOException;
-
-/**
- * Json serializer for {@link InstanceID}.
- */
-public class InstanceIDSerializer extends StdSerializer<InstanceID> {
-
- private static final long serialVersionUID = 5798852092159615938L;
-
- protected InstanceIDSerializer() {
- super(InstanceID.class);
- }
-
- @Override
- public void serialize(InstanceID value, JsonGenerator gen, SerializerProvider provider) throws IOException {
- gen.writeString(value.toString());
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/ResourceIDDeserializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/ResourceIDDeserializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/ResourceIDDeserializer.java
new file mode 100644
index 0000000..cc0d086
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/ResourceIDDeserializer.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.runtime.rest.messages.json;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationContext;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.std.StdDeserializer;
+
+import java.io.IOException;
+
+/**
+ * Json deserializer for {@link ResourceID}.
+ */
+public class ResourceIDDeserializer extends StdDeserializer<ResourceID> {
+
+ private static final long serialVersionUID = -9058463293913469849L;
+
+ protected ResourceIDDeserializer() {
+ super(ResourceID.class);
+ }
+
+ @Override
+ public ResourceID deserialize(JsonParser p, DeserializationContext ctxt) throws IOException {
+ return new ResourceID(p.getValueAsString());
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/ResourceIDSerializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/ResourceIDSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/ResourceIDSerializer.java
new file mode 100644
index 0000000..fa95451
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/ResourceIDSerializer.java
@@ -0,0 +1,45 @@
+/*
+ * 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.rest.messages.json;
+
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.instance.InstanceID;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.SerializerProvider;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.StdSerializer;
+
+import java.io.IOException;
+
+/**
+ * Json serializer for {@link InstanceID}.
+ */
+public class ResourceIDSerializer extends StdSerializer<ResourceID> {
+
+ private static final long serialVersionUID = 5798852092159615938L;
+
+ protected ResourceIDSerializer() {
+ super(ResourceID.class);
+ }
+
+ @Override
+ public void serialize(ResourceID value, JsonGenerator gen, SerializerProvider provider) throws IOException {
+ gen.writeString(value.toString());
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerDetailsInfo.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerDetailsInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerDetailsInfo.java
index 3764a26..3b0a68f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerDetailsInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerDetailsInfo.java
@@ -18,9 +18,9 @@
package org.apache.flink.runtime.rest.messages.taskmanager;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.instance.HardwareDescription;
-import org.apache.flink.runtime.instance.InstanceID;
-import org.apache.flink.runtime.rest.messages.json.InstanceIDDeserializer;
+import org.apache.flink.runtime.rest.messages.json.ResourceIDDeserializer;
import org.apache.flink.runtime.taskexecutor.TaskExecutor;
import org.apache.flink.util.Preconditions;
@@ -43,7 +43,7 @@ public class TaskManagerDetailsInfo extends TaskManagerInfo {
@JsonCreator
public TaskManagerDetailsInfo(
- @JsonDeserialize(using = InstanceIDDeserializer.class) @JsonProperty(FIELD_NAME_INSTANCE_ID) InstanceID instanceId,
+ @JsonDeserialize(using = ResourceIDDeserializer.class) @JsonProperty(FIELD_NAME_INSTANCE_ID) ResourceID resourceId,
@JsonProperty(FIELD_NAME_ADDRESS) String address,
@JsonProperty(FIELD_NAME_DATA_PORT) int dataPort,
@JsonProperty(FIELD_NAME_LAST_HEARTBEAT) long lastHeartbeat,
@@ -52,7 +52,7 @@ public class TaskManagerDetailsInfo extends TaskManagerInfo {
@JsonProperty(FIELD_NAME_HARDWARE) HardwareDescription hardwareDescription,
@JsonProperty(FIELD_NAME_METRICS) TaskManagerMetricsInfo taskManagerMetrics) {
super(
- instanceId,
+ resourceId,
address,
dataPort,
lastHeartbeat,
@@ -65,7 +65,7 @@ public class TaskManagerDetailsInfo extends TaskManagerInfo {
public TaskManagerDetailsInfo(TaskManagerInfo taskManagerInfo, TaskManagerMetricsInfo taskManagerMetrics) {
this(
- taskManagerInfo.getInstanceId(),
+ taskManagerInfo.getResourceId(),
taskManagerInfo.getAddress(),
taskManagerInfo.getDataPort(),
taskManagerInfo.getLastHeartbeat(),
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerIdPathParameter.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerIdPathParameter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerIdPathParameter.java
index 2ff7909..b0cd2b5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerIdPathParameter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerIdPathParameter.java
@@ -18,15 +18,14 @@
package org.apache.flink.runtime.rest.messages.taskmanager;
-import org.apache.flink.runtime.instance.InstanceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.rest.messages.ConversionException;
import org.apache.flink.runtime.rest.messages.MessagePathParameter;
-import org.apache.flink.util.StringUtils;
/**
* TaskManager id path parameter used by TaskManager related handlers.
*/
-public class TaskManagerIdPathParameter extends MessagePathParameter<InstanceID> {
+public class TaskManagerIdPathParameter extends MessagePathParameter<ResourceID> {
public static final String KEY = "taskmanagerid";
@@ -35,12 +34,12 @@ public class TaskManagerIdPathParameter extends MessagePathParameter<InstanceID>
}
@Override
- protected InstanceID convertFromString(String value) throws ConversionException {
- return new InstanceID(StringUtils.hexStringToByte(value));
+ protected ResourceID convertFromString(String value) throws ConversionException {
+ return new ResourceID(value);
}
@Override
- protected String convertToString(InstanceID value) {
- return StringUtils.byteToHexString(value.getBytes());
+ protected String convertToString(ResourceID value) {
+ return value.getResourceIdString();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerInfo.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerInfo.java
index 1a07273..1ad7e7d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerInfo.java
@@ -18,11 +18,11 @@
package org.apache.flink.runtime.rest.messages.taskmanager;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.instance.HardwareDescription;
-import org.apache.flink.runtime.instance.InstanceID;
import org.apache.flink.runtime.rest.messages.ResponseBody;
-import org.apache.flink.runtime.rest.messages.json.InstanceIDDeserializer;
-import org.apache.flink.runtime.rest.messages.json.InstanceIDSerializer;
+import org.apache.flink.runtime.rest.messages.json.ResourceIDDeserializer;
+import org.apache.flink.runtime.rest.messages.json.ResourceIDSerializer;
import org.apache.flink.runtime.taskexecutor.TaskExecutor;
import org.apache.flink.util.Preconditions;
@@ -53,8 +53,8 @@ public class TaskManagerInfo implements ResponseBody {
public static final String FIELD_NAME_HARDWARE = "hardware";
@JsonProperty(FIELD_NAME_INSTANCE_ID)
- @JsonSerialize(using = InstanceIDSerializer.class)
- private final InstanceID instanceId;
+ @JsonSerialize(using = ResourceIDSerializer.class)
+ private final ResourceID resourceId;
@JsonProperty(FIELD_NAME_ADDRESS)
private final String address;
@@ -76,14 +76,14 @@ public class TaskManagerInfo implements ResponseBody {
@JsonCreator
public TaskManagerInfo(
- @JsonDeserialize(using = InstanceIDDeserializer.class) @JsonProperty(FIELD_NAME_INSTANCE_ID) InstanceID instanceId,
+ @JsonDeserialize(using = ResourceIDDeserializer.class) @JsonProperty(FIELD_NAME_INSTANCE_ID) ResourceID resourceId,
@JsonProperty(FIELD_NAME_ADDRESS) String address,
@JsonProperty(FIELD_NAME_DATA_PORT) int dataPort,
@JsonProperty(FIELD_NAME_LAST_HEARTBEAT) long lastHeartbeat,
@JsonProperty(FIELD_NAME_NUMBER_SLOTS) int numberSlots,
@JsonProperty(FIELD_NAME_NUMBER_AVAILABLE_SLOTS) int numberAvailableSlots,
@JsonProperty(FIELD_NAME_HARDWARE) HardwareDescription hardwareDescription) {
- this.instanceId = Preconditions.checkNotNull(instanceId);
+ this.resourceId = Preconditions.checkNotNull(resourceId);
this.address = Preconditions.checkNotNull(address);
this.dataPort = dataPort;
this.lastHeartbeat = lastHeartbeat;
@@ -92,8 +92,8 @@ public class TaskManagerInfo implements ResponseBody {
this.hardwareDescription = Preconditions.checkNotNull(hardwareDescription);
}
- public InstanceID getInstanceId() {
- return instanceId;
+ public ResourceID getResourceId() {
+ return resourceId;
}
public String getAddress() {
@@ -133,7 +133,7 @@ public class TaskManagerInfo implements ResponseBody {
lastHeartbeat == that.lastHeartbeat &&
numberSlots == that.numberSlots &&
numberAvailableSlots == that.numberAvailableSlots &&
- Objects.equals(instanceId, that.instanceId) &&
+ Objects.equals(resourceId, that.resourceId) &&
Objects.equals(address, that.address) &&
Objects.equals(hardwareDescription, that.hardwareDescription);
}
@@ -141,7 +141,7 @@ public class TaskManagerInfo implements ResponseBody {
@Override
public int hashCode() {
return Objects.hash(
- instanceId,
+ resourceId,
address,
dataPort,
lastHeartbeat,
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ResourceManagerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ResourceManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ResourceManagerTest.java
index 2b58b6f..c238e33 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ResourceManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ResourceManagerTest.java
@@ -18,8 +18,6 @@
package org.apache.flink.runtime.clusterframework;
-import akka.actor.ActorSystem;
-import akka.testkit.JavaTestKit;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.AkkaOptions;
@@ -62,6 +60,9 @@ import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.apache.flink.runtime.testutils.TestingResourceManager;
import org.apache.flink.runtime.util.TestingFatalErrorHandler;
import org.apache.flink.util.TestLogger;
+
+import akka.actor.ActorSystem;
+import akka.testkit.JavaTestKit;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
@@ -70,8 +71,6 @@ import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
-import scala.Option;
-
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -80,7 +79,11 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import static org.junit.Assert.*;
+import scala.Option;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTest.java
new file mode 100644
index 0000000..3050718
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTest.java
@@ -0,0 +1,153 @@
+/*
+ * 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.resourcemanager;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.heartbeat.HeartbeatServices;
+import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
+import org.apache.flink.runtime.instance.HardwareDescription;
+import org.apache.flink.runtime.leaderelection.TestingLeaderElectionService;
+import org.apache.flink.runtime.metrics.NoOpMetricRegistry;
+import org.apache.flink.runtime.registration.RegistrationResponse;
+import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager;
+import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
+import org.apache.flink.runtime.rpc.RpcUtils;
+import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGateway;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.runtime.util.TestingFatalErrorHandler;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+
+public class ResourceManagerTest extends TestLogger {
+
+ private TestingRpcService rpcService;
+
+ @Before
+ public void setUp() {
+ if (rpcService != null) {
+ rpcService.stopService();
+ rpcService = null;
+ }
+
+ rpcService = new TestingRpcService();
+ }
+
+ @After
+ public void tearDown() {
+ if (rpcService != null) {
+ rpcService.stopService();
+ rpcService = null;
+ }
+ }
+
+ /**
+ * Tests that we can retrieve the correct {@link TaskManagerInfo} from the {@link ResourceManager}.
+ */
+ @Test
+ public void testRequestTaskManagerInfo() throws Exception {
+ final Configuration configuration = new Configuration();
+ final ResourceManagerConfiguration resourceManagerConfiguration = ResourceManagerConfiguration.fromConfiguration(configuration);
+ final TestingHighAvailabilityServices highAvailabilityServices = new TestingHighAvailabilityServices();
+ final SlotManager slotManager = new SlotManager(
+ rpcService.getScheduledExecutor(),
+ TestingUtils.infiniteTime(),
+ TestingUtils.infiniteTime(),
+ TestingUtils.infiniteTime());
+ final JobLeaderIdService jobLeaderIdService = new JobLeaderIdService(
+ highAvailabilityServices,
+ rpcService.getScheduledExecutor(),
+ TestingUtils.infiniteTime());
+ final TestingFatalErrorHandler testingFatalErrorHandler = new TestingFatalErrorHandler();
+
+ final TestingLeaderElectionService resourceManagerLeaderElectionService = new TestingLeaderElectionService();
+ highAvailabilityServices.setResourceManagerLeaderElectionService(resourceManagerLeaderElectionService);
+
+ final TestingResourceManager resourceManager = new TestingResourceManager(
+ rpcService,
+ ResourceManager.RESOURCE_MANAGER_NAME,
+ ResourceID.generate(),
+ resourceManagerConfiguration,
+ highAvailabilityServices,
+ new HeartbeatServices(1000L, 10000L),
+ slotManager,
+ new NoOpMetricRegistry(),
+ jobLeaderIdService,
+ testingFatalErrorHandler);
+
+ resourceManager.start();
+
+ try {
+ final ResourceID taskManagerId = ResourceID.generate();
+ final ResourceManagerGateway resourceManagerGateway = resourceManager.getSelfGateway(ResourceManagerGateway.class);
+ final TaskExecutorGateway taskExecutorGateway = new TestingTaskExecutorGateway();
+
+ // first make the ResourceManager the leader
+ resourceManagerLeaderElectionService.isLeader(UUID.randomUUID()).get();
+
+ rpcService.registerGateway(taskExecutorGateway.getAddress(), taskExecutorGateway);
+
+ final HardwareDescription hardwareDescription = new HardwareDescription(
+ 42,
+ 1337L,
+ 1337L,
+ 0L);
+
+ final int dataPort = 1234;
+
+ CompletableFuture<RegistrationResponse> registrationResponseFuture = resourceManagerGateway.registerTaskExecutor(
+ taskExecutorGateway.getAddress(),
+ taskManagerId,
+ new SlotReport(),
+ dataPort,
+ hardwareDescription,
+ TestingUtils.TIMEOUT());
+
+ Assert.assertTrue(registrationResponseFuture.get() instanceof RegistrationResponse.Success);
+
+ CompletableFuture<TaskManagerInfo> taskManagerInfoFuture = resourceManagerGateway.requestTaskManagerInfo(
+ taskManagerId,
+ TestingUtils.TIMEOUT());
+
+ TaskManagerInfo taskManagerInfo = taskManagerInfoFuture.get();
+
+ Assert.assertEquals(taskManagerId, taskManagerInfo.getResourceId());
+ Assert.assertEquals(hardwareDescription, taskManagerInfo.getHardwareDescription());
+ Assert.assertEquals(taskExecutorGateway.getAddress(), taskManagerInfo.getAddress());
+ Assert.assertEquals(dataPort, taskManagerInfo.getDataPort());
+ Assert.assertEquals(0, taskManagerInfo.getNumberSlots());
+ Assert.assertEquals(0, taskManagerInfo.getNumberAvailableSlots());
+
+ testingFatalErrorHandler.rethrowError();
+ } finally {
+ RpcUtils.terminateRpcEndpoint(resourceManager, TestingUtils.TIMEOUT());
+ highAvailabilityServices.close();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/TestingResourceManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/TestingResourceManager.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/TestingResourceManager.java
new file mode 100644
index 0000000..0d30822
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/TestingResourceManager.java
@@ -0,0 +1,76 @@
+/*
+ * 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.resourcemanager;
+
+import org.apache.flink.runtime.clusterframework.ApplicationStatus;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.heartbeat.HeartbeatServices;
+import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
+import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.resourcemanager.exceptions.ResourceManagerException;
+import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager;
+import org.apache.flink.runtime.rpc.FatalErrorHandler;
+import org.apache.flink.runtime.rpc.RpcService;
+
+/**
+ * Simple {@link ResourceManager} implementation for testing purposes.
+ */
+public class TestingResourceManager extends ResourceManager<ResourceID> {
+
+ public TestingResourceManager(
+ RpcService rpcService,
+ String resourceManagerEndpointId,
+ ResourceID resourceId,
+ ResourceManagerConfiguration resourceManagerConfiguration,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ SlotManager slotManager,
+ MetricRegistry metricRegistry,
+ JobLeaderIdService jobLeaderIdService,
+ FatalErrorHandler fatalErrorHandler) {
+ super(rpcService, resourceManagerEndpointId, resourceId, resourceManagerConfiguration, highAvailabilityServices, heartbeatServices, slotManager, metricRegistry, jobLeaderIdService, fatalErrorHandler);
+ }
+
+ @Override
+ protected void initialize() throws ResourceManagerException {
+ // noop
+ }
+
+ @Override
+ protected void shutDownApplication(ApplicationStatus finalStatus, String optionalDiagnostics) throws ResourceManagerException {
+ // noop
+ }
+
+ @Override
+ public void startNewWorker(ResourceProfile resourceProfile) {
+ // noop
+ }
+
+ @Override
+ protected ResourceID workerStarted(ResourceID resourceID) {
+ return resourceID;
+ }
+
+ @Override
+ public boolean stopWorker(ResourceID worker) {
+ // cannot stop workers
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/utils/TestingResourceManagerGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/utils/TestingResourceManagerGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/utils/TestingResourceManagerGateway.java
index f11a1eb..0f8b5fa 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/utils/TestingResourceManagerGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/utils/TestingResourceManagerGateway.java
@@ -200,7 +200,7 @@ public class TestingResourceManagerGateway implements ResourceManagerGateway {
}
@Override
- public CompletableFuture<TaskManagerInfo> requestTaskManagerInfo(InstanceID instanceId, Time timeout) {
+ public CompletableFuture<TaskManagerInfo> requestTaskManagerInfo(ResourceID resourceId, Time timeout) {
return FutureUtils.completedExceptionally(new UnsupportedOperationException("Not yet implemented"));
}
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerInfoTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerInfoTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerInfoTest.java
index 12514b7..9a1e85d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerInfoTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerInfoTest.java
@@ -18,8 +18,8 @@
package org.apache.flink.runtime.rest.messages.taskmanager;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.instance.HardwareDescription;
-import org.apache.flink.runtime.instance.InstanceID;
import org.apache.flink.runtime.rest.messages.RestResponseMarshallingTestBase;
import java.util.Random;
@@ -44,7 +44,7 @@ public class TaskManagerInfoTest extends RestResponseMarshallingTestBase<TaskMan
static TaskManagerInfo createRandomTaskManagerInfo() {
return new TaskManagerInfo(
- new InstanceID(),
+ ResourceID.generate(),
UUID.randomUUID().toString(),
random.nextInt(),
random.nextLong(),
http://git-wip-us.apache.org/repos/asf/flink/blob/dc7ab135/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
new file mode 100644
index 0000000..a9b676e
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.taskexecutor;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
+import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.executiongraph.PartitionInfo;
+import org.apache.flink.runtime.jobmaster.JobMasterId;
+import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
+import org.apache.flink.util.Preconditions;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Simple {@link TaskExecutorGateway} implementation for testing purposes.
+ */
+public class TestingTaskExecutorGateway implements TaskExecutorGateway {
+
+ private final String address;
+
+ private final String hostname;
+
+ public TestingTaskExecutorGateway() {
+ this("foobar:1234", "foobar");
+ }
+
+ public TestingTaskExecutorGateway(String address, String hostname) {
+ this.address = Preconditions.checkNotNull(address);
+ this.hostname = Preconditions.checkNotNull(hostname);
+ }
+
+ @Override
+ public CompletableFuture<Acknowledge> requestSlot(SlotID slotId, JobID jobId, AllocationID allocationId, String targetAddress, ResourceManagerId resourceManagerId, Time timeout) {
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ }
+
+ @Override
+ public CompletableFuture<Acknowledge> submitTask(TaskDeploymentDescriptor tdd, JobMasterId jobMasterId, Time timeout) {
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ }
+
+ @Override
+ public CompletableFuture<Acknowledge> updatePartitions(ExecutionAttemptID executionAttemptID, Iterable<PartitionInfo> partitionInfos, Time timeout) {
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ }
+
+ @Override
+ public void failPartition(ExecutionAttemptID executionAttemptID) {
+ // noop
+ }
+
+ @Override
+ public CompletableFuture<Acknowledge> triggerCheckpoint(ExecutionAttemptID executionAttemptID, long checkpointID, long checkpointTimestamp, CheckpointOptions checkpointOptions) {
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ }
+
+ @Override
+ public CompletableFuture<Acknowledge> confirmCheckpoint(ExecutionAttemptID executionAttemptID, long checkpointId, long checkpointTimestamp) {
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ }
+
+ @Override
+ public CompletableFuture<Acknowledge> stopTask(ExecutionAttemptID executionAttemptID, Time timeout) {
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ }
+
+ @Override
+ public CompletableFuture<Acknowledge> cancelTask(ExecutionAttemptID executionAttemptID, Time timeout) {
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ }
+
+ @Override
+ public void heartbeatFromJobManager(ResourceID heartbeatOrigin) {
+ // noop
+ }
+
+ @Override
+ public void heartbeatFromResourceManager(ResourceID heartbeatOrigin) {
+ // noop
+ }
+
+ @Override
+ public void disconnectJobManager(JobID jobId, Exception cause) {
+ // nooop
+ }
+
+ @Override
+ public void disconnectResourceManager(Exception cause) {
+ // noop
+ }
+
+ @Override
+ public String getAddress() {
+ return address;
+ }
+
+ @Override
+ public String getHostname() {
+ return hostname;
+ }
+}