YARN-5799. Fix Opportunistic Allocation to set the correct value of N…

…ode Http Address. (asuresh)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/DistributedSchedulingAllocateResponse.java

@@ -21,7 +21,6 @@
 import org.apache.hadoop.classification.InterfaceAudience.Public;
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
 import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
-import org.apache.hadoop.yarn.api.records.NodeId;
 import org.apache.hadoop.yarn.util.Records;
 
 import java.util.List;
@@ -58,9 +57,10 @@ public static DistributedSchedulingAllocateResponse newInstance(
 
   @Public
   @Unstable
-  public abstract void setNodesForScheduling(List<NodeId> nodesForScheduling);
+  public abstract void setNodesForScheduling(
+      List<RemoteNode> nodesForScheduling);
 
   @Public
   @Unstable
-  public abstract List<NodeId> getNodesForScheduling();
+  public abstract List<RemoteNode> getNodesForScheduling();
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/RegisterDistributedSchedulingAMResponse.java

@@ -21,7 +21,6 @@
 import org.apache.hadoop.classification.InterfaceAudience.Public;
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
 import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
-import org.apache.hadoop.yarn.api.records.NodeId;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.util.Records;
 
@@ -99,10 +98,11 @@ public abstract void setRegisterResponse(
 
   @Public
   @Unstable
-  public abstract void setNodesForScheduling(List<NodeId> nodesForScheduling);
+  public abstract void setNodesForScheduling(
+      List<RemoteNode> nodesForScheduling);
 
   @Public
   @Unstable
-  public abstract List<NodeId> getNodesForScheduling();
+  public abstract List<RemoteNode> getNodesForScheduling();
 
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/RemoteNode.java

@@ -0,0 +1,90 @@
+/**
+ * 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.hadoop.yarn.server.api.protocolrecords;
+
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.util.Records;
+
+/**
+ * This class is used to encapsulate the {@link NodeId} as well as the HTTP
+ * address that can be used to communicate with the Node.
+ */
+@Private
+@Unstable
+public abstract class RemoteNode implements Comparable<RemoteNode> {
+
+  /**
+   * Create new Instance.
+   * @param nodeId NodeId.
+   * @param httpAddress Http address.
+   * @return RemoteNode instance.
+   */
+  @Private
+  @Unstable
+  public static RemoteNode newInstance(NodeId nodeId, String httpAddress) {
+    RemoteNode remoteNode = Records.newRecord(RemoteNode.class);
+    remoteNode.setNodeId(nodeId);
+    remoteNode.setHttpAddress(httpAddress);
+    return remoteNode;
+  }
+
+  /**
+   * Get {@link NodeId}.
+   * @return NodeId.
+   */
+  @Private
+  @Unstable
+  public abstract NodeId getNodeId();
+
+  /**
+   * Set {@link NodeId}.
+   * @param nodeId NodeId.
+   */
+  @Private
+  @Unstable
+  public abstract void setNodeId(NodeId nodeId);
+
+  /**
+   * Get HTTP address.
+   * @return Http Address.
+   */
+  @Private
+  @Unstable
+  public abstract String getHttpAddress();
+
+  /**
+   * Set HTTP address.
+   * @param httpAddress HTTP address.
+   */
+  @Private
+  @Unstable
+  public abstract void setHttpAddress(String httpAddress);
+
+  /**
+   * Use the underlying {@link NodeId} comparator.
+   * @param other RemoteNode.
+   * @return Comparison.
+   */
+  @Override
+  public int compareTo(RemoteNode other) {
+    return this.getNodeId().compareTo(other.getNodeId());
+  }
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistributedSchedulingAllocateResponsePBImpl.java

@@ -21,12 +21,13 @@
 import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
 
 import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.AllocateResponsePBImpl;
-import org.apache.hadoop.yarn.api.records.NodeId;
-import org.apache.hadoop.yarn.api.records.impl.pb.ProtoUtils;
-import org.apache.hadoop.yarn.proto.YarnProtos;
 import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos;
+import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.RemoteNodeProto;
 import org.apache.hadoop.yarn.server.api.protocolrecords.DistributedSchedulingAllocateResponse;
 
+
+import org.apache.hadoop.yarn.server.api.protocolrecords.RemoteNode;
+
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
@@ -45,7 +46,7 @@ public class DistributedSchedulingAllocateResponsePBImpl extends
   boolean viaProto = false;
 
   private AllocateResponse allocateResponse;
-  private List<NodeId> nodesForScheduling;
+  private List<RemoteNode> nodesForScheduling;
 
   public DistributedSchedulingAllocateResponsePBImpl() {
     builder = YarnServerCommonServiceProtos.
@@ -86,8 +87,8 @@ private synchronized void mergeLocalToProto() {
   private synchronized void mergeLocalToBuilder() {
     if (this.nodesForScheduling != null) {
       builder.clearNodesForScheduling();
-      Iterable<YarnProtos.NodeIdProto> iterable = getNodeIdProtoIterable(
-          this.nodesForScheduling);
+      Iterable<YarnServerCommonServiceProtos.RemoteNodeProto> iterable =
+          getNodeIdProtoIterable(this.nodesForScheduling);
       builder.addAllNodesForScheduling(iterable);
     }
     if (this.allocateResponse != null) {
@@ -123,7 +124,7 @@ public AllocateResponse getAllocateResponse() {
   }
 
   @Override
-  public void setNodesForScheduling(List<NodeId> nodesForScheduling) {
+  public void setNodesForScheduling(List<RemoteNode> nodesForScheduling) {
     maybeInitBuilder();
     if (nodesForScheduling == null || nodesForScheduling.isEmpty()) {
       if (this.nodesForScheduling != null) {
@@ -137,7 +138,7 @@ public void setNodesForScheduling(List<NodeId> nodesForScheduling) {
   }
 
   @Override
-  public List<NodeId> getNodesForScheduling() {
+  public List<RemoteNode> getNodesForScheduling() {
     if (nodesForScheduling != null) {
       return nodesForScheduling;
     }
@@ -149,33 +150,34 @@ private synchronized void initLocalNodesForSchedulingList() {
     YarnServerCommonServiceProtos.
         DistributedSchedulingAllocateResponseProtoOrBuilder p =
             viaProto ? proto : builder;
-    List<YarnProtos.NodeIdProto> list = p.getNodesForSchedulingList();
+    List<YarnServerCommonServiceProtos.RemoteNodeProto> list =
+        p.getNodesForSchedulingList();
     nodesForScheduling = new ArrayList<>();
     if (list != null) {
-      for (YarnProtos.NodeIdProto t : list) {
-        nodesForScheduling.add(ProtoUtils.convertFromProtoFormat(t));
+      for (YarnServerCommonServiceProtos.RemoteNodeProto t : list) {
+        nodesForScheduling.add(new RemoteNodePBImpl(t));
       }
     }
   }
 
-  private synchronized Iterable<YarnProtos.NodeIdProto> getNodeIdProtoIterable(
-      final List<NodeId> nodeList) {
+  private synchronized Iterable<RemoteNodeProto> getNodeIdProtoIterable(
+      final List<RemoteNode> nodeList) {
     maybeInitBuilder();
-    return new Iterable<YarnProtos.NodeIdProto>() {
+    return new Iterable<RemoteNodeProto>() {
       @Override
-      public synchronized Iterator<YarnProtos.NodeIdProto> iterator() {
-        return new Iterator<YarnProtos.NodeIdProto>() {
+      public synchronized Iterator<RemoteNodeProto> iterator() {
+        return new Iterator<RemoteNodeProto>() {
 
-          Iterator<NodeId> iter = nodeList.iterator();
+          Iterator<RemoteNode> iter = nodeList.iterator();
 
           @Override
           public boolean hasNext() {
             return iter.hasNext();
           }
 
           @Override
-          public YarnProtos.NodeIdProto next() {
-            return ProtoUtils.convertToProtoFormat(iter.next());
+          public RemoteNodeProto next() {
+            return ((RemoteNodePBImpl)iter.next()).getProto();
           }
 
           @Override
@@ -186,5 +188,4 @@ public void remove() {
       }
     };
   }
-
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/RegisterDistributedSchedulingAMResponsePBImpl.java

@@ -23,13 +23,15 @@
 import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
 
 import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.RegisterApplicationMasterResponsePBImpl;
-import org.apache.hadoop.yarn.api.records.NodeId;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.impl.pb.ProtoUtils;
-import org.apache.hadoop.yarn.proto.YarnProtos;
 import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos;
+import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.RemoteNodeProto;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RegisterDistributedSchedulingAMResponse;
 
+
+import org.apache.hadoop.yarn.server.api.protocolrecords.RemoteNode;
+
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
@@ -52,7 +54,7 @@ public class RegisterDistributedSchedulingAMResponsePBImpl extends
   private Resource maxContainerResource;
   private Resource minContainerResource;
   private Resource incrContainerResource;
-  private List<NodeId> nodesForScheduling;
+  private List<RemoteNode> nodesForScheduling;
   private RegisterApplicationMasterResponse registerApplicationMasterResponse;
 
   public RegisterDistributedSchedulingAMResponsePBImpl() {
@@ -95,8 +97,8 @@ private synchronized void mergeLocalToProto() {
   private synchronized void mergeLocalToBuilder() {
     if (this.nodesForScheduling != null) {
       builder.clearNodesForScheduling();
-      Iterable<YarnProtos.NodeIdProto> iterable = getNodeIdProtoIterable(
-          this.nodesForScheduling);
+      Iterable<YarnServerCommonServiceProtos.RemoteNodeProto> iterable =
+          getNodeIdProtoIterable(this.nodesForScheduling);
       builder.addAllNodesForScheduling(iterable);
     }
     if (this.maxContainerResource != null) {
@@ -261,7 +263,7 @@ public long getContainerIdStart() {
   }
 
   @Override
-  public void setNodesForScheduling(List<NodeId> nodesForScheduling) {
+  public void setNodesForScheduling(List<RemoteNode> nodesForScheduling) {
     maybeInitBuilder();
     if (nodesForScheduling == null || nodesForScheduling.isEmpty()) {
       if (this.nodesForScheduling != null) {
@@ -275,7 +277,7 @@ public void setNodesForScheduling(List<NodeId> nodesForScheduling) {
   }
 
   @Override
-  public List<NodeId> getNodesForScheduling() {
+  public List<RemoteNode> getNodesForScheduling() {
     if (nodesForScheduling != null) {
       return nodesForScheduling;
     }
@@ -287,33 +289,34 @@ private synchronized void initLocalNodesForSchedulingList() {
     YarnServerCommonServiceProtos.
         RegisterDistributedSchedulingAMResponseProtoOrBuilder p =
             viaProto ? proto : builder;
-    List<YarnProtos.NodeIdProto> list = p.getNodesForSchedulingList();
+    List<YarnServerCommonServiceProtos.RemoteNodeProto> list =
+        p.getNodesForSchedulingList();
     nodesForScheduling = new ArrayList<>();
     if (list != null) {
-      for (YarnProtos.NodeIdProto t : list) {
-        nodesForScheduling.add(ProtoUtils.convertFromProtoFormat(t));
+      for (YarnServerCommonServiceProtos.RemoteNodeProto t : list) {
+        nodesForScheduling.add(new RemoteNodePBImpl(t));
       }
     }
   }
 
-  private synchronized Iterable<YarnProtos.NodeIdProto> getNodeIdProtoIterable(
-      final List<NodeId> nodeList) {
+  private synchronized Iterable<RemoteNodeProto> getNodeIdProtoIterable(
+      final List<RemoteNode> nodeList) {
     maybeInitBuilder();
-    return new Iterable<YarnProtos.NodeIdProto>() {
+    return new Iterable<RemoteNodeProto>() {
       @Override
-      public synchronized Iterator<YarnProtos.NodeIdProto> iterator() {
-        return new Iterator<YarnProtos.NodeIdProto>() {
+      public synchronized Iterator<RemoteNodeProto> iterator() {
+        return new Iterator<RemoteNodeProto>() {
 
-          Iterator<NodeId> iter = nodeList.iterator();
+          Iterator<RemoteNode> iter = nodeList.iterator();
 
           @Override
           public boolean hasNext() {
             return iter.hasNext();
           }
 
           @Override
-          public YarnProtos.NodeIdProto next() {
-            return ProtoUtils.convertToProtoFormat(iter.next());
+          public RemoteNodeProto next() {
+            return ((RemoteNodePBImpl)iter.next()).getProto();
           }
 
           @Override