You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by dr...@apache.org on 2016/08/01 15:55:36 UTC
[45/50] [abbrv] hadoop git commit: YARN-5113. Refactoring and other
clean-up for distributed scheduling. (Konstantinos Karanasos via asuresh)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5766b1d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/RegisterDistributedSchedulingAMResponse.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/RegisterDistributedSchedulingAMResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/RegisterDistributedSchedulingAMResponse.java
new file mode 100644
index 0000000..a0a0e38
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/RegisterDistributedSchedulingAMResponse.java
@@ -0,0 +1,108 @@
+/**
+ * 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.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;
+
+import java.util.List;
+
+/**
+ * This is the response to registering an Application Master when distributed
+ * scheduling is enabled. Apart from the
+ * {@link RegisterApplicationMasterResponse}, it includes various parameters
+ * to be used during distributed scheduling, such as the min and max resources
+ * that can be requested by containers.
+ */
+@Public
+@Unstable
+public abstract class RegisterDistributedSchedulingAMResponse {
+
+ @Public
+ @Unstable
+ public static RegisterDistributedSchedulingAMResponse newInstance
+ (RegisterApplicationMasterResponse regAMResp) {
+ RegisterDistributedSchedulingAMResponse response =
+ Records.newRecord(RegisterDistributedSchedulingAMResponse.class);
+ response.setRegisterResponse(regAMResp);
+ return response;
+ }
+
+ @Public
+ @Unstable
+ public abstract void setRegisterResponse(
+ RegisterApplicationMasterResponse resp);
+
+ @Public
+ @Unstable
+ public abstract RegisterApplicationMasterResponse getRegisterResponse();
+
+ @Public
+ @Unstable
+ public abstract void setMinContainerResource(Resource minResource);
+
+ @Public
+ @Unstable
+ public abstract Resource getMinContainerResource();
+
+ @Public
+ @Unstable
+ public abstract void setMaxContainerResource(Resource maxResource);
+
+ @Public
+ @Unstable
+ public abstract Resource getMaxContainerResource();
+
+ @Public
+ @Unstable
+ public abstract void setIncrContainerResource(Resource maxResource);
+
+ @Public
+ @Unstable
+ public abstract Resource getIncrContainerResource();
+
+ @Public
+ @Unstable
+ public abstract void setContainerTokenExpiryInterval(int interval);
+
+ @Public
+ @Unstable
+ public abstract int getContainerTokenExpiryInterval();
+
+ @Public
+ @Unstable
+ public abstract void setContainerIdStart(long containerIdStart);
+
+ @Public
+ @Unstable
+ public abstract long getContainerIdStart();
+
+ @Public
+ @Unstable
+ public abstract void setNodesForScheduling(List<NodeId> nodesForScheduling);
+
+ @Public
+ @Unstable
+ public abstract List<NodeId> getNodesForScheduling();
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5766b1d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistSchedAllocateRequestPBImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistSchedAllocateRequestPBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistSchedAllocateRequestPBImpl.java
deleted file mode 100644
index be386b6..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistSchedAllocateRequestPBImpl.java
+++ /dev/null
@@ -1,185 +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.hadoop.yarn.server.api.protocolrecords.impl.pb;
-
-import java.util.ArrayList;
-
-import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.AllocateRequestPBImpl;
-import org.apache.hadoop.yarn.api.records.Container;
-import org.apache.hadoop.yarn.api.records.impl.pb.ContainerPBImpl;
-import org.apache.hadoop.yarn.api.records.impl.pb.ProtoUtils;
-import org.apache.hadoop.yarn.proto.YarnProtos.ContainerProto;
-import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.DistSchedAllocateRequestProto;
-import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.DistSchedAllocateRequestProtoOrBuilder;
-import org.apache.hadoop.yarn.proto.YarnServiceProtos.AllocateRequestProto;
-import org.apache.hadoop.yarn.server.api.protocolrecords.DistSchedAllocateRequest;
-
-import java.util.Iterator;
-import java.util.List;
-
-/**
- * Implementation of {@link DistSchedAllocateRequest} for a distributed
- * scheduler to notify about the allocation of containers to the Resource
- * Manager.
- */
-public class DistSchedAllocateRequestPBImpl extends DistSchedAllocateRequest {
- private DistSchedAllocateRequestProto.Builder builder = null;
- private boolean viaProto = false;
-
- private DistSchedAllocateRequestProto proto;
- private AllocateRequest allocateRequest;
- private List<Container> containers;
-
- public DistSchedAllocateRequestPBImpl() {
- builder = DistSchedAllocateRequestProto.newBuilder();
- }
-
- public DistSchedAllocateRequestPBImpl(DistSchedAllocateRequestProto proto) {
- this.proto = proto;
- this.viaProto = true;
- }
-
- @Override
- public AllocateRequest getAllocateRequest() {
- DistSchedAllocateRequestProtoOrBuilder p = viaProto ? proto : builder;
- if (this.allocateRequest != null) {
- return this.allocateRequest;
- }
- if (!p.hasAllocateRequest()) {
- return null;
- }
- this.allocateRequest = convertFromProtoFormat(p.getAllocateRequest());
- return this.allocateRequest;
- }
-
- @Override
- public void setAllocateRequest(AllocateRequest pAllocateRequest) {
- maybeInitBuilder();
- if (allocateRequest == null) {
- builder.clearAllocateRequest();
- }
- this.allocateRequest = pAllocateRequest;
- }
-
- @Override
- public List<Container> getAllocatedContainers() {
- if (this.containers != null) {
- return this.containers;
- }
- initAllocatedContainers();
- return containers;
- }
-
- private void initAllocatedContainers() {
- DistSchedAllocateRequestProtoOrBuilder p = viaProto ? proto : builder;
- List<ContainerProto> list = p.getAllocatedContainersList();
- this.containers = new ArrayList<Container>();
- for (ContainerProto c : list) {
- this.containers.add(convertFromProtoFormat(c));
- }
- }
-
- @Override
- public void setAllocatedContainers(List<Container> pContainers) {
- maybeInitBuilder();
- if (pContainers == null || pContainers.isEmpty()) {
- if (this.containers != null) {
- this.containers.clear();
- }
- builder.clearAllocatedContainers();
- return;
- }
- this.containers = new ArrayList<>();
- this.containers.addAll(pContainers);
- }
-
- public DistSchedAllocateRequestProto getProto() {
- mergeLocalToProto();
- proto = viaProto ? proto : builder.build();
- viaProto = true;
- return proto;
- }
-
- private void maybeInitBuilder() {
- if (viaProto || builder == null) {
- builder = DistSchedAllocateRequestProto.newBuilder(proto);
- }
- viaProto = false;
- }
-
- private void mergeLocalToProto() {
- if (viaProto) {
- maybeInitBuilder();
- }
- mergeLocalToBuilder();
- proto = builder.build();
- viaProto = true;
- }
-
- private void mergeLocalToBuilder() {
- if (this.containers != null) {
- builder.clearAllocatedContainers();
- Iterable<ContainerProto> iterable =
- getContainerProtoIterable(this.containers);
- builder.addAllAllocatedContainers(iterable);
- }
- if (this.allocateRequest != null) {
- builder.setAllocateRequest(
- ((AllocateRequestPBImpl)this.allocateRequest).getProto());
- }
- }
-
- private Iterable<ContainerProto> getContainerProtoIterable(
- final List<Container> newContainersList) {
- maybeInitBuilder();
- return new Iterable<ContainerProto>() {
- @Override
- public synchronized Iterator<ContainerProto> iterator() {
- return new Iterator<ContainerProto>() {
- Iterator<Container> iter = newContainersList.iterator();
-
- @Override
- public synchronized boolean hasNext() {
- return iter.hasNext();
- }
-
- @Override
- public synchronized ContainerProto next() {
- return ProtoUtils.convertToProtoFormat(iter.next());
- }
-
- @Override
- public synchronized void remove() {
- throw new UnsupportedOperationException();
-
- }
- };
- }
- };
- }
-
- private ContainerPBImpl convertFromProtoFormat(ContainerProto p) {
- return new ContainerPBImpl(p);
- }
-
- private AllocateRequestPBImpl convertFromProtoFormat(AllocateRequestProto p) {
- return new AllocateRequestPBImpl(p);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5766b1d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistSchedAllocateResponsePBImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistSchedAllocateResponsePBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistSchedAllocateResponsePBImpl.java
deleted file mode 100644
index 3ea4965..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistSchedAllocateResponsePBImpl.java
+++ /dev/null
@@ -1,180 +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.hadoop.yarn.server.api.protocolrecords.impl.pb;
-
-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.YarnServiceProtos;
-import org.apache.hadoop.yarn.server.api.protocolrecords
- .DistSchedAllocateResponse;
-
-
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-
-public class DistSchedAllocateResponsePBImpl extends DistSchedAllocateResponse {
-
- YarnServerCommonServiceProtos.DistSchedAllocateResponseProto proto =
- YarnServerCommonServiceProtos.DistSchedAllocateResponseProto.getDefaultInstance();
- YarnServerCommonServiceProtos.DistSchedAllocateResponseProto.Builder builder = null;
- boolean viaProto = false;
-
- private AllocateResponse allocateResponse;
- private List<NodeId> nodesForScheduling;
-
- public DistSchedAllocateResponsePBImpl() {
- builder = YarnServerCommonServiceProtos.DistSchedAllocateResponseProto.newBuilder();
- }
-
- public DistSchedAllocateResponsePBImpl(YarnServerCommonServiceProtos.DistSchedAllocateResponseProto proto) {
- this.proto = proto;
- viaProto = true;
- }
-
- public YarnServerCommonServiceProtos.DistSchedAllocateResponseProto getProto() {
- mergeLocalToProto();
- proto = viaProto ? proto : builder.build();
- viaProto = true;
- return proto;
- }
-
- private void maybeInitBuilder() {
- if (viaProto || builder == null) {
- builder = YarnServerCommonServiceProtos.DistSchedAllocateResponseProto.newBuilder(proto);
- }
- viaProto = false;
- }
-
- private synchronized void mergeLocalToProto() {
- if (viaProto)
- maybeInitBuilder();
- mergeLocalToBuilder();
- proto = builder.build();
- viaProto = true;
- }
-
- private synchronized void mergeLocalToBuilder() {
- if (this.nodesForScheduling != null) {
- builder.clearNodesForScheduling();
- Iterable<YarnProtos.NodeIdProto> iterable =
- getNodeIdProtoIterable(this.nodesForScheduling);
- builder.addAllNodesForScheduling(iterable);
- }
- if (this.allocateResponse != null) {
- builder.setAllocateResponse(
- ((AllocateResponsePBImpl)this.allocateResponse).getProto());
- }
- }
- @Override
- public void setAllocateResponse(AllocateResponse response) {
- maybeInitBuilder();
- if(allocateResponse == null) {
- builder.clearAllocateResponse();
- }
- this.allocateResponse = response;
- }
-
- @Override
- public AllocateResponse getAllocateResponse() {
- if (this.allocateResponse != null) {
- return this.allocateResponse;
- }
-
- YarnServerCommonServiceProtos.DistSchedAllocateResponseProtoOrBuilder p =
- viaProto ? proto : builder;
- if (!p.hasAllocateResponse()) {
- return null;
- }
-
- this.allocateResponse =
- new AllocateResponsePBImpl(p.getAllocateResponse());
- return this.allocateResponse;
- }
-
- @Override
- public void setNodesForScheduling(List<NodeId> nodesForScheduling) {
- maybeInitBuilder();
- if (nodesForScheduling == null || nodesForScheduling.isEmpty()) {
- if (this.nodesForScheduling != null) {
- this.nodesForScheduling.clear();
- }
- builder.clearNodesForScheduling();
- return;
- }
- this.nodesForScheduling = new ArrayList<>();
- this.nodesForScheduling.addAll(nodesForScheduling);
- }
-
- @Override
- public List<NodeId> getNodesForScheduling() {
- if (nodesForScheduling != null) {
- return nodesForScheduling;
- }
- initLocalNodesForSchedulingList();
- return nodesForScheduling;
- }
-
- private synchronized void initLocalNodesForSchedulingList() {
- YarnServerCommonServiceProtos.DistSchedAllocateResponseProtoOrBuilder p =
- viaProto ? proto : builder;
- List<YarnProtos.NodeIdProto> list = p.getNodesForSchedulingList();
- nodesForScheduling = new ArrayList<>();
- if (list != null) {
- for (YarnProtos.NodeIdProto t : list) {
- nodesForScheduling.add(ProtoUtils.convertFromProtoFormat(t));
- }
- }
- }
- private synchronized Iterable<YarnProtos.NodeIdProto> getNodeIdProtoIterable(
- final List<NodeId> nodeList) {
- maybeInitBuilder();
- return new Iterable<YarnProtos.NodeIdProto>() {
- @Override
- public synchronized Iterator<YarnProtos.NodeIdProto> iterator() {
- return new Iterator<YarnProtos.NodeIdProto>() {
-
- Iterator<NodeId> iter = nodeList.iterator();
-
- @Override
- public boolean hasNext() {
- return iter.hasNext();
- }
-
- @Override
- public YarnProtos.NodeIdProto next() {
- return ProtoUtils.convertToProtoFormat(iter.next());
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
- }
- };
- }
- };
- }
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5766b1d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistSchedRegisterResponsePBImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistSchedRegisterResponsePBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistSchedRegisterResponsePBImpl.java
deleted file mode 100644
index eec62da..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistSchedRegisterResponsePBImpl.java
+++ /dev/null
@@ -1,308 +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.hadoop.yarn.server.api.protocolrecords.impl.pb;
-
-import com.google.protobuf.TextFormat;
-
-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.server.api.protocolrecords
- .DistSchedRegisterResponse;
-
-
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-
-public class DistSchedRegisterResponsePBImpl extends DistSchedRegisterResponse {
-
- YarnServerCommonServiceProtos.DistSchedRegisterResponseProto proto =
- YarnServerCommonServiceProtos.DistSchedRegisterResponseProto.getDefaultInstance();
- YarnServerCommonServiceProtos.DistSchedRegisterResponseProto.Builder builder = null;
- boolean viaProto = false;
-
- private Resource maxAllocatableCapability;
- private Resource minAllocatableCapability;
- private Resource incrAllocatableCapability;
- private List<NodeId> nodesForScheduling;
- private RegisterApplicationMasterResponse registerApplicationMasterResponse;
-
- public DistSchedRegisterResponsePBImpl() {
- builder = YarnServerCommonServiceProtos.DistSchedRegisterResponseProto.newBuilder();
- }
-
- public DistSchedRegisterResponsePBImpl(YarnServerCommonServiceProtos.DistSchedRegisterResponseProto proto) {
- this.proto = proto;
- viaProto = true;
- }
-
- public YarnServerCommonServiceProtos.DistSchedRegisterResponseProto getProto() {
- mergeLocalToProto();
- proto = viaProto ? proto : builder.build();
- viaProto = true;
- return proto;
- }
-
- private void maybeInitBuilder() {
- if (viaProto || builder == null) {
- builder = YarnServerCommonServiceProtos.DistSchedRegisterResponseProto.newBuilder(proto);
- }
- viaProto = false;
- }
-
- private synchronized void mergeLocalToProto() {
- if (viaProto)
- maybeInitBuilder();
- mergeLocalToBuilder();
- proto = builder.build();
- viaProto = true;
- }
-
- private synchronized void mergeLocalToBuilder() {
- if (this.nodesForScheduling != null) {
- builder.clearNodesForScheduling();
- Iterable<YarnProtos.NodeIdProto> iterable =
- getNodeIdProtoIterable(this.nodesForScheduling);
- builder.addAllNodesForScheduling(iterable);
- }
- if (this.maxAllocatableCapability != null) {
- builder.setMaxAllocCapability(
- ProtoUtils.convertToProtoFormat(this.maxAllocatableCapability));
- }
- if (this.minAllocatableCapability != null) {
- builder.setMinAllocCapability(
- ProtoUtils.convertToProtoFormat(this.minAllocatableCapability));
- }
- if (this.incrAllocatableCapability != null) {
- builder.setIncrAllocCapability(
- ProtoUtils.convertToProtoFormat(this.incrAllocatableCapability));
- }
- if (this.registerApplicationMasterResponse != null) {
- builder.setRegisterResponse(
- ((RegisterApplicationMasterResponsePBImpl)
- this.registerApplicationMasterResponse).getProto());
- }
- }
-
- @Override
- public void setRegisterResponse(RegisterApplicationMasterResponse resp) {
- maybeInitBuilder();
- if(registerApplicationMasterResponse == null) {
- builder.clearRegisterResponse();
- }
- this.registerApplicationMasterResponse = resp;
- }
-
- @Override
- public RegisterApplicationMasterResponse getRegisterResponse() {
- if (this.registerApplicationMasterResponse != null) {
- return this.registerApplicationMasterResponse;
- }
-
- YarnServerCommonServiceProtos.DistSchedRegisterResponseProtoOrBuilder p = viaProto ? proto : builder;
- if (!p.hasRegisterResponse()) {
- return null;
- }
-
- this.registerApplicationMasterResponse =
- new RegisterApplicationMasterResponsePBImpl(p.getRegisterResponse());
- return this.registerApplicationMasterResponse;
- }
-
- @Override
- public void setMaxAllocatableCapabilty(Resource maxResource) {
- maybeInitBuilder();
- if(maxAllocatableCapability == null) {
- builder.clearMaxAllocCapability();
- }
- this.maxAllocatableCapability = maxResource;
- }
-
- @Override
- public Resource getMaxAllocatableCapabilty() {
- if (this.maxAllocatableCapability != null) {
- return this.maxAllocatableCapability;
- }
-
- YarnServerCommonServiceProtos.DistSchedRegisterResponseProtoOrBuilder p = viaProto ? proto : builder;
- if (!p.hasMaxAllocCapability()) {
- return null;
- }
-
- this.maxAllocatableCapability =
- ProtoUtils.convertFromProtoFormat(p.getMaxAllocCapability());
- return this.maxAllocatableCapability;
- }
-
- @Override
- public void setMinAllocatableCapabilty(Resource minResource) {
- maybeInitBuilder();
- if(minAllocatableCapability == null) {
- builder.clearMinAllocCapability();
- }
- this.minAllocatableCapability = minResource;
- }
-
- @Override
- public Resource getMinAllocatableCapabilty() {
- if (this.minAllocatableCapability != null) {
- return this.minAllocatableCapability;
- }
-
- YarnServerCommonServiceProtos.DistSchedRegisterResponseProtoOrBuilder p = viaProto ? proto : builder;
- if (!p.hasMinAllocCapability()) {
- return null;
- }
-
- this.minAllocatableCapability =
- ProtoUtils.convertFromProtoFormat(p.getMinAllocCapability());
- return this.minAllocatableCapability;
- }
-
- @Override
- public void setIncrAllocatableCapabilty(Resource incrResource) {
- maybeInitBuilder();
- if(incrAllocatableCapability == null) {
- builder.clearIncrAllocCapability();
- }
- this.incrAllocatableCapability = incrResource;
- }
-
- @Override
- public Resource getIncrAllocatableCapabilty() {
- if (this.incrAllocatableCapability != null) {
- return this.incrAllocatableCapability;
- }
-
- YarnServerCommonServiceProtos.DistSchedRegisterResponseProtoOrBuilder p = viaProto ? proto : builder;
- if (!p.hasIncrAllocCapability()) {
- return null;
- }
-
- this.incrAllocatableCapability =
- ProtoUtils.convertFromProtoFormat(p.getIncrAllocCapability());
- return this.incrAllocatableCapability;
- }
-
- @Override
- public void setContainerTokenExpiryInterval(int interval) {
- maybeInitBuilder();
- builder.setContainerTokenExpiryInterval(interval);
- }
-
- @Override
- public int getContainerTokenExpiryInterval() {
- YarnServerCommonServiceProtos.DistSchedRegisterResponseProtoOrBuilder p = viaProto ? proto : builder;
- if (!p.hasContainerTokenExpiryInterval()) {
- return 0;
- }
- return p.getContainerTokenExpiryInterval();
- }
-
- @Override
- public void setContainerIdStart(long containerIdStart) {
- maybeInitBuilder();
- builder.setContainerIdStart(containerIdStart);
- }
-
- @Override
- public long getContainerIdStart() {
- YarnServerCommonServiceProtos.DistSchedRegisterResponseProtoOrBuilder p = viaProto ? proto : builder;
- if (!p.hasContainerIdStart()) {
- return 0;
- }
- return p.getContainerIdStart();
- }
-
-
- @Override
- public void setNodesForScheduling(List<NodeId> nodesForScheduling) {
- maybeInitBuilder();
- if (nodesForScheduling == null || nodesForScheduling.isEmpty()) {
- if (this.nodesForScheduling != null) {
- this.nodesForScheduling.clear();
- }
- builder.clearNodesForScheduling();
- return;
- }
- this.nodesForScheduling = new ArrayList<>();
- this.nodesForScheduling.addAll(nodesForScheduling);
- }
-
- @Override
- public List<NodeId> getNodesForScheduling() {
- if (nodesForScheduling != null) {
- return nodesForScheduling;
- }
- initLocalNodesForSchedulingList();
- return nodesForScheduling;
- }
-
- private synchronized void initLocalNodesForSchedulingList() {
- YarnServerCommonServiceProtos.DistSchedRegisterResponseProtoOrBuilder p = viaProto ? proto : builder;
- List<YarnProtos.NodeIdProto> list = p.getNodesForSchedulingList();
- nodesForScheduling = new ArrayList<>();
- if (list != null) {
- for (YarnProtos.NodeIdProto t : list) {
- nodesForScheduling.add(ProtoUtils.convertFromProtoFormat(t));
- }
- }
- }
- private synchronized Iterable<YarnProtos.NodeIdProto> getNodeIdProtoIterable(
- final List<NodeId> nodeList) {
- maybeInitBuilder();
- return new Iterable<YarnProtos.NodeIdProto>() {
- @Override
- public synchronized Iterator<YarnProtos.NodeIdProto> iterator() {
- return new Iterator<YarnProtos.NodeIdProto>() {
-
- Iterator<NodeId> iter = nodeList.iterator();
-
- @Override
- public boolean hasNext() {
- return iter.hasNext();
- }
-
- @Override
- public YarnProtos.NodeIdProto next() {
- return ProtoUtils.convertToProtoFormat(iter.next());
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException();
- }
- };
- }
- };
- }
-
- @Override
- public String toString() {
- return TextFormat.shortDebugString(getProto());
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5766b1d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistributedSchedulingAllocateRequestPBImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistributedSchedulingAllocateRequestPBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistributedSchedulingAllocateRequestPBImpl.java
new file mode 100644
index 0000000..d99c85e
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/impl/pb/DistributedSchedulingAllocateRequestPBImpl.java
@@ -0,0 +1,187 @@
+/**
+ * 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.impl.pb;
+
+import java.util.ArrayList;
+
+import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.AllocateRequestPBImpl;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.impl.pb.ContainerPBImpl;
+import org.apache.hadoop.yarn.api.records.impl.pb.ProtoUtils;
+import org.apache.hadoop.yarn.proto.YarnProtos.ContainerProto;
+import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.DistributedSchedulingAllocateRequestProto;
+import org.apache.hadoop.yarn.proto.YarnServerCommonServiceProtos.DistributedSchedulingAllocateRequestProtoOrBuilder;
+import org.apache.hadoop.yarn.proto.YarnServiceProtos.AllocateRequestProto;
+import org.apache.hadoop.yarn.server.api.protocolrecords.DistributedSchedulingAllocateRequest;
+
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * Implementation of {@link DistributedSchedulingAllocateRequest}.
+ */
+public class DistributedSchedulingAllocateRequestPBImpl
+ extends DistributedSchedulingAllocateRequest {
+ private DistributedSchedulingAllocateRequestProto.Builder builder = null;
+ private boolean viaProto = false;
+
+ private DistributedSchedulingAllocateRequestProto proto;
+ private AllocateRequest allocateRequest;
+ private List<Container> containers;
+
+ public DistributedSchedulingAllocateRequestPBImpl() {
+ builder = DistributedSchedulingAllocateRequestProto.newBuilder();
+ }
+
+ public DistributedSchedulingAllocateRequestPBImpl(
+ DistributedSchedulingAllocateRequestProto proto) {
+ this.proto = proto;
+ this.viaProto = true;
+ }
+
+ @Override
+ public AllocateRequest getAllocateRequest() {
+ DistributedSchedulingAllocateRequestProtoOrBuilder p =
+ viaProto ? proto : builder;
+ if (this.allocateRequest != null) {
+ return this.allocateRequest;
+ }
+ if (!p.hasAllocateRequest()) {
+ return null;
+ }
+ this.allocateRequest = convertFromProtoFormat(p.getAllocateRequest());
+ return this.allocateRequest;
+ }
+
+ @Override
+ public void setAllocateRequest(AllocateRequest pAllocateRequest) {
+ maybeInitBuilder();
+ if (allocateRequest == null) {
+ builder.clearAllocateRequest();
+ }
+ this.allocateRequest = pAllocateRequest;
+ }
+
+ @Override
+ public List<Container> getAllocatedContainers() {
+ if (this.containers != null) {
+ return this.containers;
+ }
+ initAllocatedContainers();
+ return containers;
+ }
+
+ private void initAllocatedContainers() {
+ DistributedSchedulingAllocateRequestProtoOrBuilder p =
+ viaProto ? proto : builder;
+ List<ContainerProto> list = p.getAllocatedContainersList();
+ this.containers = new ArrayList<Container>();
+ for (ContainerProto c : list) {
+ this.containers.add(convertFromProtoFormat(c));
+ }
+ }
+
+ @Override
+ public void setAllocatedContainers(List<Container> pContainers) {
+ maybeInitBuilder();
+ if (pContainers == null || pContainers.isEmpty()) {
+ if (this.containers != null) {
+ this.containers.clear();
+ }
+ builder.clearAllocatedContainers();
+ return;
+ }
+ this.containers = new ArrayList<>();
+ this.containers.addAll(pContainers);
+ }
+
+ public DistributedSchedulingAllocateRequestProto getProto() {
+ mergeLocalToProto();
+ proto = viaProto ? proto : builder.build();
+ viaProto = true;
+ return proto;
+ }
+
+ private void maybeInitBuilder() {
+ if (viaProto || builder == null) {
+ builder = DistributedSchedulingAllocateRequestProto.newBuilder(proto);
+ }
+ viaProto = false;
+ }
+
+ private void mergeLocalToProto() {
+ if (viaProto) {
+ maybeInitBuilder();
+ }
+ mergeLocalToBuilder();
+ proto = builder.build();
+ viaProto = true;
+ }
+
+ private void mergeLocalToBuilder() {
+ if (this.containers != null) {
+ builder.clearAllocatedContainers();
+ Iterable<ContainerProto> iterable =
+ getContainerProtoIterable(this.containers);
+ builder.addAllAllocatedContainers(iterable);
+ }
+ if (this.allocateRequest != null) {
+ builder.setAllocateRequest(
+ ((AllocateRequestPBImpl)this.allocateRequest).getProto());
+ }
+ }
+
+ private Iterable<ContainerProto> getContainerProtoIterable(
+ final List<Container> newContainersList) {
+ maybeInitBuilder();
+ return new Iterable<ContainerProto>() {
+ @Override
+ public synchronized Iterator<ContainerProto> iterator() {
+ return new Iterator<ContainerProto>() {
+ Iterator<Container> iter = newContainersList.iterator();
+
+ @Override
+ public synchronized boolean hasNext() {
+ return iter.hasNext();
+ }
+
+ @Override
+ public synchronized ContainerProto next() {
+ return ProtoUtils.convertToProtoFormat(iter.next());
+ }
+
+ @Override
+ public synchronized void remove() {
+ throw new UnsupportedOperationException();
+
+ }
+ };
+ }
+ };
+ }
+
+ private ContainerPBImpl convertFromProtoFormat(ContainerProto p) {
+ return new ContainerPBImpl(p);
+ }
+
+ private AllocateRequestPBImpl convertFromProtoFormat(AllocateRequestProto p) {
+ return new AllocateRequestPBImpl(p);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5766b1d/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
----------------------------------------------------------------------
diff --git a/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 b/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
new file mode 100644
index 0000000..18d5073
--- /dev/null
+++ b/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
@@ -0,0 +1,190 @@
+/**
+ * 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.impl.pb;
+
+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.server.api.protocolrecords.DistributedSchedulingAllocateResponse;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * Implementation of {@link DistributedSchedulingAllocateResponse}.
+ */
+public class DistributedSchedulingAllocateResponsePBImpl extends
+ DistributedSchedulingAllocateResponse {
+
+ YarnServerCommonServiceProtos.DistributedSchedulingAllocateResponseProto
+ proto = YarnServerCommonServiceProtos.
+ DistributedSchedulingAllocateResponseProto.getDefaultInstance();
+ YarnServerCommonServiceProtos.DistributedSchedulingAllocateResponseProto.
+ Builder builder = null;
+ boolean viaProto = false;
+
+ private AllocateResponse allocateResponse;
+ private List<NodeId> nodesForScheduling;
+
+ public DistributedSchedulingAllocateResponsePBImpl() {
+ builder = YarnServerCommonServiceProtos.
+ DistributedSchedulingAllocateResponseProto.newBuilder();
+ }
+
+ public DistributedSchedulingAllocateResponsePBImpl(
+ YarnServerCommonServiceProtos.
+ DistributedSchedulingAllocateResponseProto proto) {
+ this.proto = proto;
+ viaProto = true;
+ }
+
+ public YarnServerCommonServiceProtos.
+ DistributedSchedulingAllocateResponseProto getProto() {
+ mergeLocalToProto();
+ proto = viaProto ? proto : builder.build();
+ viaProto = true;
+ return proto;
+ }
+
+ private void maybeInitBuilder() {
+ if (viaProto || builder == null) {
+ builder = YarnServerCommonServiceProtos.
+ DistributedSchedulingAllocateResponseProto.newBuilder(proto);
+ }
+ viaProto = false;
+ }
+
+ private synchronized void mergeLocalToProto() {
+ if (viaProto)
+ maybeInitBuilder();
+ mergeLocalToBuilder();
+ proto = builder.build();
+ viaProto = true;
+ }
+
+ private synchronized void mergeLocalToBuilder() {
+ if (this.nodesForScheduling != null) {
+ builder.clearNodesForScheduling();
+ Iterable<YarnProtos.NodeIdProto> iterable = getNodeIdProtoIterable(
+ this.nodesForScheduling);
+ builder.addAllNodesForScheduling(iterable);
+ }
+ if (this.allocateResponse != null) {
+ builder.setAllocateResponse(
+ ((AllocateResponsePBImpl) this.allocateResponse).getProto());
+ }
+ }
+
+ @Override
+ public void setAllocateResponse(AllocateResponse response) {
+ maybeInitBuilder();
+ if (allocateResponse == null) {
+ builder.clearAllocateResponse();
+ }
+ this.allocateResponse = response;
+ }
+
+ @Override
+ public AllocateResponse getAllocateResponse() {
+ if (this.allocateResponse != null) {
+ return this.allocateResponse;
+ }
+
+ YarnServerCommonServiceProtos.
+ DistributedSchedulingAllocateResponseProtoOrBuilder p =
+ viaProto ? proto : builder;
+ if (!p.hasAllocateResponse()) {
+ return null;
+ }
+
+ this.allocateResponse = new AllocateResponsePBImpl(p.getAllocateResponse());
+ return this.allocateResponse;
+ }
+
+ @Override
+ public void setNodesForScheduling(List<NodeId> nodesForScheduling) {
+ maybeInitBuilder();
+ if (nodesForScheduling == null || nodesForScheduling.isEmpty()) {
+ if (this.nodesForScheduling != null) {
+ this.nodesForScheduling.clear();
+ }
+ builder.clearNodesForScheduling();
+ return;
+ }
+ this.nodesForScheduling = new ArrayList<>();
+ this.nodesForScheduling.addAll(nodesForScheduling);
+ }
+
+ @Override
+ public List<NodeId> getNodesForScheduling() {
+ if (nodesForScheduling != null) {
+ return nodesForScheduling;
+ }
+ initLocalNodesForSchedulingList();
+ return nodesForScheduling;
+ }
+
+ private synchronized void initLocalNodesForSchedulingList() {
+ YarnServerCommonServiceProtos.
+ DistributedSchedulingAllocateResponseProtoOrBuilder p =
+ viaProto ? proto : builder;
+ List<YarnProtos.NodeIdProto> list = p.getNodesForSchedulingList();
+ nodesForScheduling = new ArrayList<>();
+ if (list != null) {
+ for (YarnProtos.NodeIdProto t : list) {
+ nodesForScheduling.add(ProtoUtils.convertFromProtoFormat(t));
+ }
+ }
+ }
+
+ private synchronized Iterable<YarnProtos.NodeIdProto> getNodeIdProtoIterable(
+ final List<NodeId> nodeList) {
+ maybeInitBuilder();
+ return new Iterable<YarnProtos.NodeIdProto>() {
+ @Override
+ public synchronized Iterator<YarnProtos.NodeIdProto> iterator() {
+ return new Iterator<YarnProtos.NodeIdProto>() {
+
+ Iterator<NodeId> iter = nodeList.iterator();
+
+ @Override
+ public boolean hasNext() {
+ return iter.hasNext();
+ }
+
+ @Override
+ public YarnProtos.NodeIdProto next() {
+ return ProtoUtils.convertToProtoFormat(iter.next());
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+ };
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5766b1d/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
----------------------------------------------------------------------
diff --git a/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 b/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
new file mode 100644
index 0000000..4aaf99c
--- /dev/null
+++ b/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
@@ -0,0 +1,332 @@
+/**
+ * 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.impl.pb;
+
+import com.google.protobuf.TextFormat;
+
+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.server.api.protocolrecords.RegisterDistributedSchedulingAMResponse;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * Implementation of {@link RegisterDistributedSchedulingAMResponse}.
+ */
+public class RegisterDistributedSchedulingAMResponsePBImpl extends
+ RegisterDistributedSchedulingAMResponse {
+
+ YarnServerCommonServiceProtos.RegisterDistributedSchedulingAMResponseProto
+ proto =
+ YarnServerCommonServiceProtos.
+ RegisterDistributedSchedulingAMResponseProto
+ .getDefaultInstance();
+ YarnServerCommonServiceProtos.RegisterDistributedSchedulingAMResponseProto.
+ Builder builder = null;
+ boolean viaProto = false;
+
+ private Resource maxContainerResource;
+ private Resource minContainerResource;
+ private Resource incrContainerResource;
+ private List<NodeId> nodesForScheduling;
+ private RegisterApplicationMasterResponse registerApplicationMasterResponse;
+
+ public RegisterDistributedSchedulingAMResponsePBImpl() {
+ builder = YarnServerCommonServiceProtos.
+ RegisterDistributedSchedulingAMResponseProto.newBuilder();
+ }
+
+ public RegisterDistributedSchedulingAMResponsePBImpl(
+ YarnServerCommonServiceProtos.
+ RegisterDistributedSchedulingAMResponseProto proto) {
+ this.proto = proto;
+ viaProto = true;
+ }
+
+ public YarnServerCommonServiceProtos.
+ RegisterDistributedSchedulingAMResponseProto
+ getProto() {
+ mergeLocalToProto();
+ proto = viaProto ? proto : builder.build();
+ viaProto = true;
+ return proto;
+ }
+
+ private void maybeInitBuilder() {
+ if (viaProto || builder == null) {
+ builder = YarnServerCommonServiceProtos.
+ RegisterDistributedSchedulingAMResponseProto.newBuilder(proto);
+ }
+ viaProto = false;
+ }
+
+ private synchronized void mergeLocalToProto() {
+ if (viaProto)
+ maybeInitBuilder();
+ mergeLocalToBuilder();
+ proto = builder.build();
+ viaProto = true;
+ }
+
+ private synchronized void mergeLocalToBuilder() {
+ if (this.nodesForScheduling != null) {
+ builder.clearNodesForScheduling();
+ Iterable<YarnProtos.NodeIdProto> iterable = getNodeIdProtoIterable(
+ this.nodesForScheduling);
+ builder.addAllNodesForScheduling(iterable);
+ }
+ if (this.maxContainerResource != null) {
+ builder.setMaxContainerResource(ProtoUtils.convertToProtoFormat(
+ this.maxContainerResource));
+ }
+ if (this.minContainerResource != null) {
+ builder.setMinContainerResource(ProtoUtils.convertToProtoFormat(
+ this.minContainerResource));
+ }
+ if (this.incrContainerResource != null) {
+ builder.setIncrContainerResource(
+ ProtoUtils.convertToProtoFormat(this.incrContainerResource));
+ }
+ if (this.registerApplicationMasterResponse != null) {
+ builder.setRegisterResponse(
+ ((RegisterApplicationMasterResponsePBImpl)
+ this.registerApplicationMasterResponse).getProto());
+ }
+ }
+
+ @Override
+ public void setRegisterResponse(RegisterApplicationMasterResponse resp) {
+ maybeInitBuilder();
+ if (registerApplicationMasterResponse == null) {
+ builder.clearRegisterResponse();
+ }
+ this.registerApplicationMasterResponse = resp;
+ }
+
+ @Override
+ public RegisterApplicationMasterResponse getRegisterResponse() {
+ if (this.registerApplicationMasterResponse != null) {
+ return this.registerApplicationMasterResponse;
+ }
+
+ YarnServerCommonServiceProtos.
+ RegisterDistributedSchedulingAMResponseProtoOrBuilder p =
+ viaProto ? proto : builder;
+ if (!p.hasRegisterResponse()) {
+ return null;
+ }
+
+ this.registerApplicationMasterResponse =
+ new RegisterApplicationMasterResponsePBImpl(p.getRegisterResponse());
+ return this.registerApplicationMasterResponse;
+ }
+
+ @Override
+ public void setMaxContainerResource(Resource maxResource) {
+ maybeInitBuilder();
+ if (maxContainerResource == null) {
+ builder.clearMaxContainerResource();
+ }
+ this.maxContainerResource = maxResource;
+ }
+
+ @Override
+ public Resource getMaxContainerResource() {
+ if (this.maxContainerResource != null) {
+ return this.maxContainerResource;
+ }
+
+ YarnServerCommonServiceProtos.
+ RegisterDistributedSchedulingAMResponseProtoOrBuilder p =
+ viaProto ? proto : builder;
+ if (!p.hasMaxContainerResource()) {
+ return null;
+ }
+
+ this.maxContainerResource = ProtoUtils.convertFromProtoFormat(p
+ .getMaxContainerResource());
+ return this.maxContainerResource;
+ }
+
+ @Override
+ public void setMinContainerResource(Resource minResource) {
+ maybeInitBuilder();
+ if (minContainerResource == null) {
+ builder.clearMinContainerResource();
+ }
+ this.minContainerResource = minResource;
+ }
+
+ @Override
+ public Resource getMinContainerResource() {
+ if (this.minContainerResource != null) {
+ return this.minContainerResource;
+ }
+
+ YarnServerCommonServiceProtos.
+ RegisterDistributedSchedulingAMResponseProtoOrBuilder p =
+ viaProto ? proto : builder;
+ if (!p.hasMinContainerResource()) {
+ return null;
+ }
+
+ this.minContainerResource = ProtoUtils.convertFromProtoFormat(p
+ .getMinContainerResource());
+ return this.minContainerResource;
+ }
+
+ @Override
+ public void setIncrContainerResource(Resource incrResource) {
+ maybeInitBuilder();
+ if (incrContainerResource == null) {
+ builder.clearIncrContainerResource();
+ }
+ this.incrContainerResource = incrResource;
+ }
+
+ @Override
+ public Resource getIncrContainerResource() {
+ if (this.incrContainerResource != null) {
+ return this.incrContainerResource;
+ }
+
+ YarnServerCommonServiceProtos.
+ RegisterDistributedSchedulingAMResponseProtoOrBuilder p =
+ viaProto ? proto : builder;
+ if (!p.hasIncrContainerResource()) {
+ return null;
+ }
+
+ this.incrContainerResource = ProtoUtils.convertFromProtoFormat(p
+ .getIncrContainerResource());
+ return this.incrContainerResource;
+ }
+
+ @Override
+ public void setContainerTokenExpiryInterval(int interval) {
+ maybeInitBuilder();
+ builder.setContainerTokenExpiryInterval(interval);
+ }
+
+ @Override
+ public int getContainerTokenExpiryInterval() {
+ YarnServerCommonServiceProtos.
+ RegisterDistributedSchedulingAMResponseProtoOrBuilder p =
+ viaProto ? proto : builder;
+ if (!p.hasContainerTokenExpiryInterval()) {
+ return 0;
+ }
+ return p.getContainerTokenExpiryInterval();
+ }
+
+ @Override
+ public void setContainerIdStart(long containerIdStart) {
+ maybeInitBuilder();
+ builder.setContainerIdStart(containerIdStart);
+ }
+
+ @Override
+ public long getContainerIdStart() {
+ YarnServerCommonServiceProtos.
+ RegisterDistributedSchedulingAMResponseProtoOrBuilder p =
+ viaProto ? proto : builder;
+ if (!p.hasContainerIdStart()) {
+ return 0;
+ }
+ return p.getContainerIdStart();
+ }
+
+ @Override
+ public void setNodesForScheduling(List<NodeId> nodesForScheduling) {
+ maybeInitBuilder();
+ if (nodesForScheduling == null || nodesForScheduling.isEmpty()) {
+ if (this.nodesForScheduling != null) {
+ this.nodesForScheduling.clear();
+ }
+ builder.clearNodesForScheduling();
+ return;
+ }
+ this.nodesForScheduling = new ArrayList<>();
+ this.nodesForScheduling.addAll(nodesForScheduling);
+ }
+
+ @Override
+ public List<NodeId> getNodesForScheduling() {
+ if (nodesForScheduling != null) {
+ return nodesForScheduling;
+ }
+ initLocalNodesForSchedulingList();
+ return nodesForScheduling;
+ }
+
+ private synchronized void initLocalNodesForSchedulingList() {
+ YarnServerCommonServiceProtos.
+ RegisterDistributedSchedulingAMResponseProtoOrBuilder p =
+ viaProto ? proto : builder;
+ List<YarnProtos.NodeIdProto> list = p.getNodesForSchedulingList();
+ nodesForScheduling = new ArrayList<>();
+ if (list != null) {
+ for (YarnProtos.NodeIdProto t : list) {
+ nodesForScheduling.add(ProtoUtils.convertFromProtoFormat(t));
+ }
+ }
+ }
+
+ private synchronized Iterable<YarnProtos.NodeIdProto> getNodeIdProtoIterable(
+ final List<NodeId> nodeList) {
+ maybeInitBuilder();
+ return new Iterable<YarnProtos.NodeIdProto>() {
+ @Override
+ public synchronized Iterator<YarnProtos.NodeIdProto> iterator() {
+ return new Iterator<YarnProtos.NodeIdProto>() {
+
+ Iterator<NodeId> iter = nodeList.iterator();
+
+ @Override
+ public boolean hasNext() {
+ return iter.hasNext();
+ }
+
+ @Override
+ public YarnProtos.NodeIdProto next() {
+ return ProtoUtils.convertToProtoFormat(iter.next());
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+ };
+ }
+
+ @Override
+ public String toString() {
+ return TextFormat.shortDebugString(getProto());
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5766b1d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/distributed_scheduler_protocol.proto
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/distributed_scheduler_protocol.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/distributed_scheduler_protocol.proto
deleted file mode 100644
index 818eb4a..0000000
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/distributed_scheduler_protocol.proto
+++ /dev/null
@@ -1,39 +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.
- */
-
-/**
- * These .proto interfaces are public and stable.
- * Please see http://wiki.apache.org/hadoop/Compatibility
- * for what changes are allowed for a *stable* .proto interface.
- */
-
-option java_package = "org.apache.hadoop.yarn.proto";
-option java_outer_classname = "DistributedSchedulerProtocol";
-option java_generic_services = true;
-option java_generate_equals_and_hash = true;
-package hadoop.yarn;
-
-import "yarn_service_protos.proto";
-import "yarn_server_common_service_protos.proto";
-
-
-service DistributedSchedulerProtocolService {
- rpc registerApplicationMasterForDistributedScheduling (RegisterApplicationMasterRequestProto) returns (DistSchedRegisterResponseProto);
- rpc finishApplicationMaster (FinishApplicationMasterRequestProto) returns (FinishApplicationMasterResponseProto);
- rpc allocateForDistributedScheduling (DistSchedAllocateRequestProto) returns (DistSchedAllocateResponseProto);
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5766b1d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/distributed_scheduling_am_protocol.proto
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/distributed_scheduling_am_protocol.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/distributed_scheduling_am_protocol.proto
new file mode 100644
index 0000000..274eaa2
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/distributed_scheduling_am_protocol.proto
@@ -0,0 +1,39 @@
+/**
+ * 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.
+ */
+
+
+/**
+ * These .proto interfaces are public and stable.
+ * Please see http://wiki.apache.org/hadoop/Compatibility
+ * for what changes are allowed for a *stable* .proto interface.
+ */
+
+option java_package = "org.apache.hadoop.yarn.proto";
+option java_outer_classname = "DistributedSchedulingAMProtocol";
+option java_generic_services = true;
+option java_generate_equals_and_hash = true;
+package hadoop.yarn;
+
+import "yarn_service_protos.proto";
+import "yarn_server_common_service_protos.proto";
+
+service DistributedSchedulingAMProtocolService {
+ rpc registerApplicationMasterForDistributedScheduling (RegisterApplicationMasterRequestProto) returns (RegisterDistributedSchedulingAMResponseProto);
+ rpc allocateForDistributedScheduling (DistributedSchedulingAllocateRequestProto) returns (DistributedSchedulingAllocateResponseProto);
+ rpc finishApplicationMaster (FinishApplicationMasterRequestProto) returns (FinishApplicationMasterResponseProto);
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5766b1d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/yarn_server_common_service_protos.proto
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/yarn_server_common_service_protos.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/yarn_server_common_service_protos.proto
index 3660252..55ac875 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/yarn_server_common_service_protos.proto
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/proto/yarn_server_common_service_protos.proto
@@ -26,22 +26,22 @@ import "yarn_protos.proto";
import "yarn_server_common_protos.proto";
import "yarn_service_protos.proto";
-message DistSchedRegisterResponseProto {
+message RegisterDistributedSchedulingAMResponseProto {
optional RegisterApplicationMasterResponseProto register_response = 1;
- optional ResourceProto max_alloc_capability = 2;
- optional ResourceProto min_alloc_capability = 3;
- optional ResourceProto incr_alloc_capability = 4;
+ optional ResourceProto max_container_resource = 2;
+ optional ResourceProto min_container_resource = 3;
+ optional ResourceProto incr_container_resource = 4;
optional int32 container_token_expiry_interval = 5;
optional int64 container_id_start = 6;
repeated NodeIdProto nodes_for_scheduling = 7;
}
-message DistSchedAllocateResponseProto {
+message DistributedSchedulingAllocateResponseProto {
optional AllocateResponseProto allocate_response = 1;
repeated NodeIdProto nodes_for_scheduling = 2;
}
-message DistSchedAllocateRequestProto {
+message DistributedSchedulingAllocateRequestProto {
optional AllocateRequestProto allocate_request = 1;
repeated ContainerProto allocated_containers = 2;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5766b1d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/AMRMProxyService.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/AMRMProxyService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/AMRMProxyService.java
index ac360f4..511db16 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/AMRMProxyService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/AMRMProxyService.java
@@ -65,7 +65,7 @@ import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.Ap
import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEvent;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationEventType;
-import org.apache.hadoop.yarn.server.nodemanager.scheduler.LocalScheduler;
+import org.apache.hadoop.yarn.server.nodemanager.scheduler.DistributedScheduler;
import org.apache.hadoop.yarn.server.security.MasterKeyData;
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.apache.hadoop.yarn.server.utils.YarnServerSecurityUtils;
@@ -467,10 +467,9 @@ public class AMRMProxyService extends AbstractService implements
interceptorClassNames.add(item.trim());
}
- // Make sure LocalScheduler is present at the beginning
- // of the chain..
+ // Make sure DistributedScheduler is present at the beginning of the chain.
if (this.nmContext.isDistributedSchedulingEnabled()) {
- interceptorClassNames.add(0, LocalScheduler.class.getName());
+ interceptorClassNames.add(0, DistributedScheduler.class.getName());
}
return interceptorClassNames;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5766b1d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/AbstractRequestInterceptor.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/AbstractRequestInterceptor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/AbstractRequestInterceptor.java
index 55c65f4..e6c9bbd 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/AbstractRequestInterceptor.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/AbstractRequestInterceptor.java
@@ -21,12 +21,11 @@ package org.apache.hadoop.yarn.server.nodemanager.amrmproxy;
import org.apache.hadoop.conf.Configuration;
import com.google.common.base.Preconditions;
-import org.apache.hadoop.yarn.api.protocolrecords
- .RegisterApplicationMasterRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterRequest;
import org.apache.hadoop.yarn.exceptions.YarnException;
-import org.apache.hadoop.yarn.server.api.protocolrecords.DistSchedAllocateRequest;
-import org.apache.hadoop.yarn.server.api.protocolrecords.DistSchedAllocateResponse;
-import org.apache.hadoop.yarn.server.api.protocolrecords.DistSchedRegisterResponse;
+import org.apache.hadoop.yarn.server.api.protocolrecords.DistributedSchedulingAllocateRequest;
+import org.apache.hadoop.yarn.server.api.protocolrecords.DistributedSchedulingAllocateResponse;
+import org.apache.hadoop.yarn.server.api.protocolrecords.RegisterDistributedSchedulingAMResponse;
import java.io.IOException;
@@ -118,8 +117,9 @@ public abstract class AbstractRequestInterceptor implements
* @throws IOException
*/
@Override
- public DistSchedAllocateResponse allocateForDistributedScheduling(
- DistSchedAllocateRequest request) throws YarnException, IOException {
+ public DistributedSchedulingAllocateResponse allocateForDistributedScheduling(
+ DistributedSchedulingAllocateRequest request)
+ throws YarnException, IOException {
return (this.nextInterceptor != null) ?
this.nextInterceptor.allocateForDistributedScheduling(request) : null;
}
@@ -134,10 +134,10 @@ public abstract class AbstractRequestInterceptor implements
* @throws IOException
*/
@Override
- public DistSchedRegisterResponse
+ public RegisterDistributedSchedulingAMResponse
registerApplicationMasterForDistributedScheduling(
- RegisterApplicationMasterRequest request)
- throws YarnException, IOException {
+ RegisterApplicationMasterRequest request)
+ throws YarnException, IOException {
return (this.nextInterceptor != null) ? this.nextInterceptor
.registerApplicationMasterForDistributedScheduling(request) : null;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5766b1d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/DefaultRequestInterceptor.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/DefaultRequestInterceptor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/DefaultRequestInterceptor.java
index debff76..75fe022 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/DefaultRequestInterceptor.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/DefaultRequestInterceptor.java
@@ -43,12 +43,11 @@ import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
import org.apache.hadoop.yarn.security.AMRMTokenIdentifier;
-import org.apache.hadoop.yarn.server.api.DistributedSchedulerProtocol;
+import org.apache.hadoop.yarn.server.api.DistributedSchedulingAMProtocol;
import org.apache.hadoop.yarn.server.api.ServerRMProxy;
-import org.apache.hadoop.yarn.server.api.protocolrecords
- .DistSchedAllocateResponse;
-import org.apache.hadoop.yarn.server.api.protocolrecords.DistSchedAllocateRequest;
-import org.apache.hadoop.yarn.server.api.protocolrecords.DistSchedRegisterResponse;
+import org.apache.hadoop.yarn.server.api.protocolrecords.DistributedSchedulingAllocateRequest;
+import org.apache.hadoop.yarn.server.api.protocolrecords.DistributedSchedulingAllocateResponse;
+import org.apache.hadoop.yarn.server.api.protocolrecords.RegisterDistributedSchedulingAMResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -63,7 +62,7 @@ public final class DefaultRequestInterceptor extends
AbstractRequestInterceptor {
private static final Logger LOG = LoggerFactory
.getLogger(DefaultRequestInterceptor.class);
- private DistributedSchedulerProtocol rmClient;
+ private DistributedSchedulingAMProtocol rmClient;
private UserGroupInformation user = null;
@Override
@@ -77,13 +76,13 @@ public final class DefaultRequestInterceptor extends
user.addToken(appContext.getAMRMToken());
final Configuration conf = this.getConf();
- rmClient =
- user.doAs(new PrivilegedExceptionAction<DistributedSchedulerProtocol>() {
+ rmClient = user.doAs(
+ new PrivilegedExceptionAction<DistributedSchedulingAMProtocol>() {
@Override
- public DistributedSchedulerProtocol run() throws Exception {
+ public DistributedSchedulingAMProtocol run() throws Exception {
setAMRMTokenService(conf);
return ServerRMProxy.createRMProxy(conf,
- DistributedSchedulerProtocol.class);
+ DistributedSchedulingAMProtocol.class);
}
});
} catch (IOException e) {
@@ -124,7 +123,7 @@ public final class DefaultRequestInterceptor extends
}
@Override
- public DistSchedRegisterResponse
+ public RegisterDistributedSchedulingAMResponse
registerApplicationMasterForDistributedScheduling
(RegisterApplicationMasterRequest request) throws YarnException,
IOException {
@@ -134,13 +133,14 @@ public final class DefaultRequestInterceptor extends
}
@Override
- public DistSchedAllocateResponse allocateForDistributedScheduling(
- DistSchedAllocateRequest request) throws YarnException, IOException {
+ public DistributedSchedulingAllocateResponse allocateForDistributedScheduling(
+ DistributedSchedulingAllocateRequest request)
+ throws YarnException, IOException {
if (LOG.isDebugEnabled()) {
LOG.debug("Forwarding allocateForDistributedScheduling request" +
"to the real YARN RM");
}
- DistSchedAllocateResponse allocateResponse =
+ DistributedSchedulingAllocateResponse allocateResponse =
rmClient.allocateForDistributedScheduling(request);
if (allocateResponse.getAllocateResponse().getAMRMToken() != null) {
updateAMRMToken(allocateResponse.getAllocateResponse().getAMRMToken());
@@ -180,10 +180,10 @@ public final class DefaultRequestInterceptor extends
@VisibleForTesting
public void setRMClient(final ApplicationMasterProtocol rmClient) {
- if (rmClient instanceof DistributedSchedulerProtocol) {
- this.rmClient = (DistributedSchedulerProtocol)rmClient;
+ if (rmClient instanceof DistributedSchedulingAMProtocol) {
+ this.rmClient = (DistributedSchedulingAMProtocol)rmClient;
} else {
- this.rmClient = new DistributedSchedulerProtocol() {
+ this.rmClient = new DistributedSchedulingAMProtocol() {
@Override
public RegisterApplicationMasterResponse registerApplicationMaster
(RegisterApplicationMasterRequest request) throws YarnException,
@@ -205,7 +205,7 @@ public final class DefaultRequestInterceptor extends
}
@Override
- public DistSchedRegisterResponse
+ public RegisterDistributedSchedulingAMResponse
registerApplicationMasterForDistributedScheduling
(RegisterApplicationMasterRequest request) throws YarnException,
IOException {
@@ -213,8 +213,9 @@ public final class DefaultRequestInterceptor extends
}
@Override
- public DistSchedAllocateResponse allocateForDistributedScheduling(
- DistSchedAllocateRequest request)
+ public DistributedSchedulingAllocateResponse
+ allocateForDistributedScheduling(
+ DistributedSchedulingAllocateRequest request)
throws YarnException, IOException {
throw new IOException("Not Supported !!");
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/e5766b1d/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/RequestInterceptor.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/RequestInterceptor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/RequestInterceptor.java
index 7a73563..5995af1 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/RequestInterceptor.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/RequestInterceptor.java
@@ -19,14 +19,14 @@
package org.apache.hadoop.yarn.server.nodemanager.amrmproxy;
import org.apache.hadoop.conf.Configurable;
-import org.apache.hadoop.yarn.server.api.DistributedSchedulerProtocol;
+import org.apache.hadoop.yarn.server.api.DistributedSchedulingAMProtocol;
/**
* Defines the contract to be implemented by the request intercepter classes,
* that can be used to intercept and inspect messages sent from the application
* master to the resource manager.
*/
-public interface RequestInterceptor extends DistributedSchedulerProtocol,
+public interface RequestInterceptor extends DistributedSchedulingAMProtocol,
Configurable {
/**
* This method is called for initializing the intercepter. This is guaranteed
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org