YARN-3866. AM-RM protocol changes to support container resizing. Contributed by Meng Ding
This commit is contained in:
Jian He
Wangda Tan
parent
692d51c09d
commit
dfe2cb849f
|
|
@ -46,8 +46,6 @@ import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterRespo
|
|||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationId;
|
||||
import org.apache.hadoop.yarn.api.records.Container;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceDecrease;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceIncrease;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerStatus;
|
||||
import org.apache.hadoop.yarn.api.records.NMToken;
|
||||
import org.apache.hadoop.yarn.api.records.NodeReport;
|
||||
|
|
@ -254,8 +252,8 @@ public class TestLocalContainerAllocator {
|
|||
Resources.none(), null, 1, null,
|
||||
Collections.<NMToken>emptyList(),
|
||||
yarnToken,
|
||||
Collections.<ContainerResourceIncrease>emptyList(),
|
||||
Collections.<ContainerResourceDecrease>emptyList());
|
||||
Collections.<Container>emptyList(),
|
||||
Collections.<Container>emptyList());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -200,6 +200,9 @@ Release 2.8.0 - UNRELEASED
|
|||
YARN-3212. RMNode State Transition Update with DECOMMISSIONING state.
|
||||
(Junping Du via wangda)
|
||||
|
||||
YARN-3866. AM-RM protocol changes to support container resizing. (Meng Ding
|
||||
via jianhe)
|
||||
|
||||
IMPROVEMENTS
|
||||
|
||||
YARN-644. Basic null check is not performed on passed in arguments before
|
||||
|
|
|
|||
|
|
@ -22,11 +22,12 @@ import java.util.List;
|
|||
|
||||
import org.apache.hadoop.classification.InterfaceAudience.Public;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Stable;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
||||
import org.apache.hadoop.yarn.api.ApplicationMasterProtocol;
|
||||
import org.apache.hadoop.yarn.api.records.Container;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.ResourceBlacklistRequest;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceIncreaseRequest;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceChangeRequest;
|
||||
import org.apache.hadoop.yarn.api.records.ResourceRequest;
|
||||
import org.apache.hadoop.yarn.util.Records;
|
||||
|
||||
|
|
@ -46,6 +47,16 @@ import org.apache.hadoop.yarn.util.Records;
|
|||
* <li>
|
||||
* A list of unused {@link Container} which are being returned.
|
||||
* </li>
|
||||
* <li>
|
||||
* A list of {@link ContainerResourceChangeRequest} to inform
|
||||
* the <code>ResourceManager</code> about the resource increase
|
||||
* requirements of running containers.
|
||||
* </li>
|
||||
* <li>
|
||||
* A list of {@link ContainerResourceChangeRequest} to inform
|
||||
* the <code>ResourceManager</code> about the resource decrease
|
||||
* requirements of running containers.
|
||||
* </li>
|
||||
* </ul>
|
||||
*
|
||||
* @see ApplicationMasterProtocol#allocate(AllocateRequest)
|
||||
|
|
@ -61,7 +72,7 @@ public abstract class AllocateRequest {
|
|||
List<ContainerId> containersToBeReleased,
|
||||
ResourceBlacklistRequest resourceBlacklistRequest) {
|
||||
return newInstance(responseID, appProgress, resourceAsk,
|
||||
containersToBeReleased, resourceBlacklistRequest, null);
|
||||
containersToBeReleased, resourceBlacklistRequest, null, null);
|
||||
}
|
||||
|
||||
@Public
|
||||
|
|
@ -70,7 +81,8 @@ public abstract class AllocateRequest {
|
|||
List<ResourceRequest> resourceAsk,
|
||||
List<ContainerId> containersToBeReleased,
|
||||
ResourceBlacklistRequest resourceBlacklistRequest,
|
||||
List<ContainerResourceIncreaseRequest> increaseRequests) {
|
||||
List<ContainerResourceChangeRequest> increaseRequests,
|
||||
List<ContainerResourceChangeRequest> decreaseRequests) {
|
||||
AllocateRequest allocateRequest = Records.newRecord(AllocateRequest.class);
|
||||
allocateRequest.setResponseId(responseID);
|
||||
allocateRequest.setProgress(appProgress);
|
||||
|
|
@ -78,6 +90,7 @@ public abstract class AllocateRequest {
|
|||
allocateRequest.setReleaseList(containersToBeReleased);
|
||||
allocateRequest.setResourceBlacklistRequest(resourceBlacklistRequest);
|
||||
allocateRequest.setIncreaseRequests(increaseRequests);
|
||||
allocateRequest.setDecreaseRequests(decreaseRequests);
|
||||
return allocateRequest;
|
||||
}
|
||||
|
||||
|
|
@ -184,20 +197,38 @@ public abstract class AllocateRequest {
|
|||
ResourceBlacklistRequest resourceBlacklistRequest);
|
||||
|
||||
/**
|
||||
* Get the <code>ContainerResourceIncreaseRequest</code> being sent by the
|
||||
* <code>ApplicationMaster</code>
|
||||
* Get the list of container resource increase requests being sent by the
|
||||
* <code>ApplicationMaster</code>.
|
||||
*/
|
||||
@Public
|
||||
@Stable
|
||||
public abstract List<ContainerResourceIncreaseRequest> getIncreaseRequests();
|
||||
@Unstable
|
||||
public abstract List<ContainerResourceChangeRequest> getIncreaseRequests();
|
||||
|
||||
/**
|
||||
* Set the <code>ContainerResourceIncreaseRequest</code> to inform the
|
||||
* <code>ResourceManager</code> about some container's resources need to be
|
||||
* increased
|
||||
* Set the list of container resource increase requests to inform the
|
||||
* <code>ResourceManager</code> about the containers whose resources need
|
||||
* to be increased.
|
||||
*/
|
||||
@Public
|
||||
@Stable
|
||||
@Unstable
|
||||
public abstract void setIncreaseRequests(
|
||||
List<ContainerResourceIncreaseRequest> increaseRequests);
|
||||
List<ContainerResourceChangeRequest> increaseRequests);
|
||||
|
||||
/**
|
||||
* Get the list of container resource decrease requests being sent by the
|
||||
* <code>ApplicationMaster</code>.
|
||||
*/
|
||||
@Public
|
||||
@Unstable
|
||||
public abstract List<ContainerResourceChangeRequest> getDecreaseRequests();
|
||||
|
||||
/**
|
||||
* Set the list of container resource decrease requests to inform the
|
||||
* <code>ResourceManager</code> about the containers whose resources need
|
||||
* to be decreased.
|
||||
*/
|
||||
@Public
|
||||
@Unstable
|
||||
public abstract void setDecreaseRequests(
|
||||
List<ContainerResourceChangeRequest> decreaseRequests);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,8 +28,6 @@ import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
|||
import org.apache.hadoop.yarn.api.ApplicationMasterProtocol;
|
||||
import org.apache.hadoop.yarn.api.records.AMCommand;
|
||||
import org.apache.hadoop.yarn.api.records.Container;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceDecrease;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceIncrease;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerStatus;
|
||||
import org.apache.hadoop.yarn.api.records.NMToken;
|
||||
import org.apache.hadoop.yarn.api.records.NodeReport;
|
||||
|
|
@ -59,6 +57,14 @@ import org.apache.hadoop.yarn.util.Records;
|
|||
* <li>The number of available nodes in a cluster.</li>
|
||||
* <li>A description of resources requested back by the cluster</li>
|
||||
* <li>AMRMToken, if AMRMToken has been rolled over</li>
|
||||
* <li>
|
||||
* A list of {@link Container} representing the containers
|
||||
* whose resource has been increased.
|
||||
* </li>
|
||||
* <li>
|
||||
* A list of {@link Container} representing the containers
|
||||
* whose resource has been decreased.
|
||||
* </li>
|
||||
* </ul>
|
||||
*
|
||||
* @see ApplicationMasterProtocol#allocate(AllocateRequest)
|
||||
|
|
@ -94,8 +100,8 @@ public abstract class AllocateResponse {
|
|||
List<Container> allocatedContainers, List<NodeReport> updatedNodes,
|
||||
Resource availResources, AMCommand command, int numClusterNodes,
|
||||
PreemptionMessage preempt, List<NMToken> nmTokens,
|
||||
List<ContainerResourceIncrease> increasedContainers,
|
||||
List<ContainerResourceDecrease> decreasedContainers) {
|
||||
List<Container> increasedContainers,
|
||||
List<Container> decreasedContainers) {
|
||||
AllocateResponse response = newInstance(responseId, completedContainers,
|
||||
allocatedContainers, updatedNodes, availResources, command,
|
||||
numClusterNodes, preempt, nmTokens);
|
||||
|
|
@ -111,8 +117,8 @@ public abstract class AllocateResponse {
|
|||
List<Container> allocatedContainers, List<NodeReport> updatedNodes,
|
||||
Resource availResources, AMCommand command, int numClusterNodes,
|
||||
PreemptionMessage preempt, List<NMToken> nmTokens, Token amRMToken,
|
||||
List<ContainerResourceIncrease> increasedContainers,
|
||||
List<ContainerResourceDecrease> decreasedContainers) {
|
||||
List<Container> increasedContainers,
|
||||
List<Container> decreasedContainers) {
|
||||
AllocateResponse response =
|
||||
newInstance(responseId, completedContainers, allocatedContainers,
|
||||
updatedNodes, availResources, command, numClusterNodes, preempt,
|
||||
|
|
@ -263,34 +269,38 @@ public abstract class AllocateResponse {
|
|||
public abstract void setNMTokens(List<NMToken> nmTokens);
|
||||
|
||||
/**
|
||||
* Get the list of newly increased containers by <code>ResourceManager</code>
|
||||
* Get the list of newly increased containers by
|
||||
* <code>ResourceManager</code>.
|
||||
*/
|
||||
@Public
|
||||
@Stable
|
||||
public abstract List<ContainerResourceIncrease> getIncreasedContainers();
|
||||
@Unstable
|
||||
public abstract List<Container> getIncreasedContainers();
|
||||
|
||||
/**
|
||||
* Set the list of newly increased containers by <code>ResourceManager</code>
|
||||
* Set the list of newly increased containers by
|
||||
* <code>ResourceManager</code>.
|
||||
*/
|
||||
@Private
|
||||
@Unstable
|
||||
public abstract void setIncreasedContainers(
|
||||
List<ContainerResourceIncrease> increasedContainers);
|
||||
List<Container> increasedContainers);
|
||||
|
||||
/**
|
||||
* Get the list of newly decreased containers by <code>NodeManager</code>
|
||||
* Get the list of newly decreased containers by
|
||||
* <code>ResourceManager</code>.
|
||||
*/
|
||||
@Public
|
||||
@Stable
|
||||
public abstract List<ContainerResourceDecrease> getDecreasedContainers();
|
||||
@Unstable
|
||||
public abstract List<Container> getDecreasedContainers();
|
||||
|
||||
/**
|
||||
* Set the list of newly decreased containers by <code>NodeManager</code>
|
||||
* Set the list of newly decreased containers by
|
||||
* <code>ResourceManager</code>.
|
||||
*/
|
||||
@Private
|
||||
@Unstable
|
||||
public abstract void setDecreasedContainers(
|
||||
List<ContainerResourceDecrease> decreasedContainers);
|
||||
List<Container> decreasedContainers);
|
||||
|
||||
/**
|
||||
* The AMRMToken that belong to this attempt
|
||||
|
|
|
|||
|
|
@ -19,34 +19,71 @@
|
|||
package org.apache.hadoop.yarn.api.records;
|
||||
|
||||
import org.apache.hadoop.classification.InterfaceAudience.Public;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
||||
import org.apache.hadoop.yarn.api.ApplicationMasterProtocol;
|
||||
import org.apache.hadoop.yarn.util.Records;
|
||||
|
||||
/**
|
||||
* Used by Application Master, send a container resource increase request to
|
||||
* Resource Manager
|
||||
* {@code ContainerResourceChangeRequest} represents the request made by an
|
||||
* application to the {@code ResourceManager} to change resource allocation of
|
||||
* a running {@code Container}.
|
||||
* <p>
|
||||
* It includes:
|
||||
* <ul>
|
||||
* <li>{@link ContainerId} for the container.</li>
|
||||
* <li>
|
||||
* {@link Resource} capability of the container after the resource change
|
||||
* is completed.
|
||||
* </li>
|
||||
* </ul>
|
||||
*
|
||||
* @see ApplicationMasterProtocol#allocate(org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest)
|
||||
*/
|
||||
@Public
|
||||
public abstract class ContainerResourceIncreaseRequest {
|
||||
@Unstable
|
||||
public abstract class ContainerResourceChangeRequest {
|
||||
|
||||
@Public
|
||||
public static ContainerResourceIncreaseRequest newInstance(
|
||||
@Unstable
|
||||
public static ContainerResourceChangeRequest newInstance(
|
||||
ContainerId existingContainerId, Resource targetCapability) {
|
||||
ContainerResourceIncreaseRequest context = Records
|
||||
.newRecord(ContainerResourceIncreaseRequest.class);
|
||||
ContainerResourceChangeRequest context = Records
|
||||
.newRecord(ContainerResourceChangeRequest.class);
|
||||
context.setContainerId(existingContainerId);
|
||||
context.setCapability(targetCapability);
|
||||
return context;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the <code>ContainerId</code> of the container.
|
||||
* @return <code>ContainerId</code> of the container
|
||||
*/
|
||||
@Public
|
||||
@Unstable
|
||||
public abstract ContainerId getContainerId();
|
||||
|
||||
/**
|
||||
* Set the <code>ContainerId</code> of the container.
|
||||
* @param containerId <code>ContainerId</code> of the container
|
||||
*/
|
||||
@Public
|
||||
@Unstable
|
||||
public abstract void setContainerId(ContainerId containerId);
|
||||
|
||||
/**
|
||||
* Get the <code>Resource</code> capability of the container.
|
||||
* @return <code>Resource</code> capability of the container
|
||||
*/
|
||||
@Public
|
||||
@Unstable
|
||||
public abstract Resource getCapability();
|
||||
|
||||
/**
|
||||
* Set the <code>Resource</code> capability of the container.
|
||||
* @param capability <code>Resource</code> capability of the container
|
||||
*/
|
||||
@Public
|
||||
@Unstable
|
||||
public abstract void setCapability(Resource capability);
|
||||
|
||||
@Override
|
||||
|
|
@ -56,9 +93,9 @@ public abstract class ContainerResourceIncreaseRequest {
|
|||
|
||||
@Override
|
||||
public boolean equals(Object other) {
|
||||
if (other instanceof ContainerResourceIncreaseRequest) {
|
||||
ContainerResourceIncreaseRequest ctx =
|
||||
(ContainerResourceIncreaseRequest) other;
|
||||
if (other instanceof ContainerResourceChangeRequest) {
|
||||
ContainerResourceChangeRequest ctx =
|
||||
(ContainerResourceChangeRequest) other;
|
||||
|
||||
if (getContainerId() == null && ctx.getContainerId() != null) {
|
||||
return false;
|
||||
|
|
@ -1,78 +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.api.records;
|
||||
|
||||
import org.apache.hadoop.classification.InterfaceAudience.Public;
|
||||
import org.apache.hadoop.yarn.util.Records;
|
||||
|
||||
/**
|
||||
* Used by Application Master to ask Node Manager reduce size of a specified
|
||||
* container
|
||||
*/
|
||||
public abstract class ContainerResourceDecrease {
|
||||
@Public
|
||||
public static ContainerResourceDecrease newInstance(
|
||||
ContainerId existingContainerId, Resource targetCapability) {
|
||||
ContainerResourceDecrease context = Records
|
||||
.newRecord(ContainerResourceDecrease.class);
|
||||
context.setContainerId(existingContainerId);
|
||||
context.setCapability(targetCapability);
|
||||
return context;
|
||||
}
|
||||
|
||||
@Public
|
||||
public abstract ContainerId getContainerId();
|
||||
|
||||
@Public
|
||||
public abstract void setContainerId(ContainerId containerId);
|
||||
|
||||
@Public
|
||||
public abstract Resource getCapability();
|
||||
|
||||
@Public
|
||||
public abstract void setCapability(Resource capability);
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return getCapability().hashCode() + getContainerId().hashCode();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object other) {
|
||||
if (other instanceof ContainerResourceDecrease) {
|
||||
ContainerResourceDecrease ctx = (ContainerResourceDecrease)other;
|
||||
|
||||
if (getContainerId() == null && ctx.getContainerId() != null) {
|
||||
return false;
|
||||
} else if (!getContainerId().equals(ctx.getContainerId())) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (getCapability() == null && ctx.getCapability() != null) {
|
||||
return false;
|
||||
} else if (!getCapability().equals(ctx.getCapability())) {
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,84 +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.api.records;
|
||||
|
||||
import org.apache.hadoop.classification.InterfaceAudience.Public;
|
||||
import org.apache.hadoop.yarn.util.Records;
|
||||
|
||||
/**
|
||||
* Represent a new increased container accepted by Resource Manager
|
||||
*/
|
||||
public abstract class ContainerResourceIncrease {
|
||||
@Public
|
||||
public static ContainerResourceIncrease newInstance(
|
||||
ContainerId existingContainerId, Resource targetCapability, Token token) {
|
||||
ContainerResourceIncrease context = Records
|
||||
.newRecord(ContainerResourceIncrease.class);
|
||||
context.setContainerId(existingContainerId);
|
||||
context.setCapability(targetCapability);
|
||||
context.setContainerToken(token);
|
||||
return context;
|
||||
}
|
||||
|
||||
@Public
|
||||
public abstract ContainerId getContainerId();
|
||||
|
||||
@Public
|
||||
public abstract void setContainerId(ContainerId containerId);
|
||||
|
||||
@Public
|
||||
public abstract Resource getCapability();
|
||||
|
||||
@Public
|
||||
public abstract void setCapability(Resource capability);
|
||||
|
||||
@Public
|
||||
public abstract Token getContainerToken();
|
||||
|
||||
@Public
|
||||
public abstract void setContainerToken(Token token);
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return getCapability().hashCode() + getContainerId().hashCode();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object other) {
|
||||
if (other instanceof ContainerResourceIncrease) {
|
||||
ContainerResourceIncrease ctx = (ContainerResourceIncrease)other;
|
||||
|
||||
if (getContainerId() == null && ctx.getContainerId() != null) {
|
||||
return false;
|
||||
} else if (!getContainerId().equals(ctx.getContainerId())) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (getCapability() == null && ctx.getCapability() != null) {
|
||||
return false;
|
||||
} else if (!getCapability().equals(ctx.getCapability())) {
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -34,6 +34,7 @@ import org.apache.hadoop.yarn.util.Records;
|
|||
* <li>{@code ContainerState} of the container.</li>
|
||||
* <li><em>Exit status</em> of a completed container.</li>
|
||||
* <li><em>Diagnostic</em> message for a failed container.</li>
|
||||
* <li>{@link Resource} allocated to the container.</li>
|
||||
* </ul>
|
||||
*/
|
||||
@Public
|
||||
|
|
@ -114,4 +115,16 @@ public abstract class ContainerStatus {
|
|||
@Private
|
||||
@Unstable
|
||||
public abstract void setDiagnostics(String diagnostics);
|
||||
|
||||
/**
|
||||
* Get the <code>Resource</code> allocated to the container.
|
||||
* @return <code>Resource</code> allocated to the container
|
||||
*/
|
||||
@Public
|
||||
@Unstable
|
||||
public abstract Resource getCapability();
|
||||
|
||||
@Private
|
||||
@Unstable
|
||||
public abstract void setCapability(Resource capability);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -470,6 +470,7 @@ message ContainerStatusProto {
|
|||
optional ContainerStateProto state = 2;
|
||||
optional string diagnostics = 3 [default = "N/A"];
|
||||
optional int32 exit_status = 4 [default = -1000];
|
||||
optional ResourceProto capability = 5;
|
||||
}
|
||||
|
||||
enum ContainerExitStatusProto {
|
||||
|
|
@ -479,18 +480,7 @@ enum ContainerExitStatusProto {
|
|||
DISKS_FAILED = -101;
|
||||
}
|
||||
|
||||
message ContainerResourceIncreaseRequestProto {
|
||||
optional ContainerIdProto container_id = 1;
|
||||
optional ResourceProto capability = 2;
|
||||
}
|
||||
|
||||
message ContainerResourceIncreaseProto {
|
||||
optional ContainerIdProto container_id = 1;
|
||||
optional ResourceProto capability = 2;
|
||||
optional hadoop.common.TokenProto container_token = 3;
|
||||
}
|
||||
|
||||
message ContainerResourceDecreaseProto {
|
||||
message ContainerResourceChangeRequestProto {
|
||||
optional ContainerIdProto container_id = 1;
|
||||
optional ResourceProto capability = 2;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -66,7 +66,8 @@ message AllocateRequestProto {
|
|||
optional ResourceBlacklistRequestProto blacklist_request = 3;
|
||||
optional int32 response_id = 4;
|
||||
optional float progress = 5;
|
||||
repeated ContainerResourceIncreaseRequestProto increase_request = 6;
|
||||
repeated ContainerResourceChangeRequestProto increase_request = 6;
|
||||
repeated ContainerResourceChangeRequestProto decrease_request = 7;
|
||||
}
|
||||
|
||||
message NMTokenProto {
|
||||
|
|
@ -84,8 +85,8 @@ message AllocateResponseProto {
|
|||
optional int32 num_cluster_nodes = 7;
|
||||
optional PreemptionMessageProto preempt = 8;
|
||||
repeated NMTokenProto nm_tokens = 9;
|
||||
repeated ContainerResourceIncreaseProto increased_containers = 10;
|
||||
repeated ContainerResourceDecreaseProto decreased_containers = 11;
|
||||
repeated ContainerProto increased_containers = 10;
|
||||
repeated ContainerProto decreased_containers = 11;
|
||||
optional hadoop.common.TokenProto am_rm_token = 12;
|
||||
}
|
||||
|
||||
|
|
@ -286,6 +287,15 @@ message GetContainerStatusesResponseProto {
|
|||
repeated ContainerExceptionMapProto failed_requests = 2;
|
||||
}
|
||||
|
||||
message IncreaseContainersResourceRequestProto {
|
||||
repeated hadoop.common.TokenProto increase_containers = 1;
|
||||
}
|
||||
|
||||
message IncreaseContainersResourceResponseProto {
|
||||
repeated ContainerIdProto succeeded_requests = 1;
|
||||
repeated ContainerExceptionMapProto failed_requests = 2;
|
||||
}
|
||||
|
||||
//////////////////////////////////////////////////////
|
||||
/////// Application_History_Protocol /////////////////
|
||||
//////////////////////////////////////////////////////
|
||||
|
|
|
|||
|
|
@ -27,15 +27,15 @@ import org.apache.hadoop.classification.InterfaceAudience.Private;
|
|||
import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
||||
import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceIncreaseRequest;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceChangeRequest;
|
||||
import org.apache.hadoop.yarn.api.records.ResourceBlacklistRequest;
|
||||
import org.apache.hadoop.yarn.api.records.ResourceRequest;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerIdPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerResourceIncreaseRequestPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerResourceChangeRequestPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ResourceBlacklistRequestPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ResourceRequestPBImpl;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerIdProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceIncreaseRequestProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceChangeRequestProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ResourceBlacklistRequestProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ResourceRequestProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnServiceProtos.AllocateRequestProto;
|
||||
|
|
@ -52,7 +52,8 @@ public class AllocateRequestPBImpl extends AllocateRequest {
|
|||
|
||||
private List<ResourceRequest> ask = null;
|
||||
private List<ContainerId> release = null;
|
||||
private List<ContainerResourceIncreaseRequest> increaseRequests = null;
|
||||
private List<ContainerResourceChangeRequest> increaseRequests = null;
|
||||
private List<ContainerResourceChangeRequest> decreaseRequests = null;
|
||||
private ResourceBlacklistRequest blacklistRequest = null;
|
||||
|
||||
public AllocateRequestPBImpl() {
|
||||
|
|
@ -101,6 +102,9 @@ public class AllocateRequestPBImpl extends AllocateRequest {
|
|||
if (this.increaseRequests != null) {
|
||||
addIncreaseRequestsToProto();
|
||||
}
|
||||
if (this.decreaseRequests != null) {
|
||||
addDecreaseRequestsToProto();
|
||||
}
|
||||
if (this.blacklistRequest != null) {
|
||||
builder.setBlacklistRequest(convertToProtoFormat(this.blacklistRequest));
|
||||
}
|
||||
|
|
@ -162,14 +166,14 @@ public class AllocateRequestPBImpl extends AllocateRequest {
|
|||
}
|
||||
|
||||
@Override
|
||||
public List<ContainerResourceIncreaseRequest> getIncreaseRequests() {
|
||||
public List<ContainerResourceChangeRequest> getIncreaseRequests() {
|
||||
initIncreaseRequests();
|
||||
return this.increaseRequests;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setIncreaseRequests(
|
||||
List<ContainerResourceIncreaseRequest> increaseRequests) {
|
||||
List<ContainerResourceChangeRequest> increaseRequests) {
|
||||
if (increaseRequests == null) {
|
||||
return;
|
||||
}
|
||||
|
|
@ -178,6 +182,23 @@ public class AllocateRequestPBImpl extends AllocateRequest {
|
|||
this.increaseRequests.addAll(increaseRequests);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<ContainerResourceChangeRequest> getDecreaseRequests() {
|
||||
initDecreaseRequests();
|
||||
return this.decreaseRequests;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setDecreaseRequests(
|
||||
List<ContainerResourceChangeRequest> decreaseRequests) {
|
||||
if (decreaseRequests == null) {
|
||||
return;
|
||||
}
|
||||
initDecreaseRequests();
|
||||
this.decreaseRequests.clear();
|
||||
this.decreaseRequests.addAll(decreaseRequests);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ResourceBlacklistRequest getResourceBlacklistRequest() {
|
||||
AllocateRequestProtoOrBuilder p = viaProto ? proto : builder;
|
||||
|
|
@ -252,28 +273,42 @@ public class AllocateRequestPBImpl extends AllocateRequest {
|
|||
return;
|
||||
}
|
||||
AllocateRequestProtoOrBuilder p = viaProto ? proto : builder;
|
||||
List<ContainerResourceIncreaseRequestProto> list =
|
||||
List<ContainerResourceChangeRequestProto> list =
|
||||
p.getIncreaseRequestList();
|
||||
this.increaseRequests = new ArrayList<ContainerResourceIncreaseRequest>();
|
||||
this.increaseRequests = new ArrayList<ContainerResourceChangeRequest>();
|
||||
|
||||
for (ContainerResourceIncreaseRequestProto c : list) {
|
||||
for (ContainerResourceChangeRequestProto c : list) {
|
||||
this.increaseRequests.add(convertFromProtoFormat(c));
|
||||
}
|
||||
}
|
||||
|
||||
private void initDecreaseRequests() {
|
||||
if (this.decreaseRequests != null) {
|
||||
return;
|
||||
}
|
||||
AllocateRequestProtoOrBuilder p = viaProto ? proto : builder;
|
||||
List<ContainerResourceChangeRequestProto> list =
|
||||
p.getDecreaseRequestList();
|
||||
this.decreaseRequests = new ArrayList<>();
|
||||
|
||||
for (ContainerResourceChangeRequestProto c : list) {
|
||||
this.decreaseRequests.add(convertFromProtoFormat(c));
|
||||
}
|
||||
}
|
||||
|
||||
private void addIncreaseRequestsToProto() {
|
||||
maybeInitBuilder();
|
||||
builder.clearIncreaseRequest();
|
||||
if (increaseRequests == null) {
|
||||
return;
|
||||
}
|
||||
Iterable<ContainerResourceIncreaseRequestProto> iterable =
|
||||
new Iterable<ContainerResourceIncreaseRequestProto>() {
|
||||
Iterable<ContainerResourceChangeRequestProto> iterable =
|
||||
new Iterable<ContainerResourceChangeRequestProto>() {
|
||||
@Override
|
||||
public Iterator<ContainerResourceIncreaseRequestProto> iterator() {
|
||||
return new Iterator<ContainerResourceIncreaseRequestProto>() {
|
||||
public Iterator<ContainerResourceChangeRequestProto> iterator() {
|
||||
return new Iterator<ContainerResourceChangeRequestProto>() {
|
||||
|
||||
Iterator<ContainerResourceIncreaseRequest> iter =
|
||||
Iterator<ContainerResourceChangeRequest> iter =
|
||||
increaseRequests.iterator();
|
||||
|
||||
@Override
|
||||
|
|
@ -282,7 +317,7 @@ public class AllocateRequestPBImpl extends AllocateRequest {
|
|||
}
|
||||
|
||||
@Override
|
||||
public ContainerResourceIncreaseRequestProto next() {
|
||||
public ContainerResourceChangeRequestProto next() {
|
||||
return convertToProtoFormat(iter.next());
|
||||
}
|
||||
|
||||
|
|
@ -297,6 +332,42 @@ public class AllocateRequestPBImpl extends AllocateRequest {
|
|||
builder.addAllIncreaseRequest(iterable);
|
||||
}
|
||||
|
||||
private void addDecreaseRequestsToProto() {
|
||||
maybeInitBuilder();
|
||||
builder.clearDecreaseRequest();
|
||||
if (decreaseRequests == null) {
|
||||
return;
|
||||
}
|
||||
Iterable<ContainerResourceChangeRequestProto> iterable =
|
||||
new Iterable<ContainerResourceChangeRequestProto>() {
|
||||
@Override
|
||||
public Iterator<ContainerResourceChangeRequestProto> iterator() {
|
||||
return new Iterator<ContainerResourceChangeRequestProto>() {
|
||||
|
||||
Iterator<ContainerResourceChangeRequest> iter =
|
||||
decreaseRequests.iterator();
|
||||
|
||||
@Override
|
||||
public boolean hasNext() {
|
||||
return iter.hasNext();
|
||||
}
|
||||
|
||||
@Override
|
||||
public ContainerResourceChangeRequestProto next() {
|
||||
return convertToProtoFormat(iter.next());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void remove() {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
};
|
||||
|
||||
}
|
||||
};
|
||||
builder.addAllDecreaseRequest(iterable);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<ContainerId> getReleaseList() {
|
||||
initReleases();
|
||||
|
|
@ -367,14 +438,14 @@ public class AllocateRequestPBImpl extends AllocateRequest {
|
|||
return ((ResourceRequestPBImpl)t).getProto();
|
||||
}
|
||||
|
||||
private ContainerResourceIncreaseRequestPBImpl convertFromProtoFormat(
|
||||
ContainerResourceIncreaseRequestProto p) {
|
||||
return new ContainerResourceIncreaseRequestPBImpl(p);
|
||||
private ContainerResourceChangeRequestPBImpl convertFromProtoFormat(
|
||||
ContainerResourceChangeRequestProto p) {
|
||||
return new ContainerResourceChangeRequestPBImpl(p);
|
||||
}
|
||||
|
||||
private ContainerResourceIncreaseRequestProto convertToProtoFormat(
|
||||
ContainerResourceIncreaseRequest t) {
|
||||
return ((ContainerResourceIncreaseRequestPBImpl) t).getProto();
|
||||
private ContainerResourceChangeRequestProto convertToProtoFormat(
|
||||
ContainerResourceChangeRequest t) {
|
||||
return ((ContainerResourceChangeRequestPBImpl) t).getProto();
|
||||
}
|
||||
|
||||
private ContainerIdPBImpl convertFromProtoFormat(ContainerIdProto p) {
|
||||
|
|
|
|||
|
|
@ -29,8 +29,6 @@ import org.apache.hadoop.security.proto.SecurityProtos.TokenProto;
|
|||
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
|
||||
import org.apache.hadoop.yarn.api.records.AMCommand;
|
||||
import org.apache.hadoop.yarn.api.records.Container;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceDecrease;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceIncrease;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerStatus;
|
||||
import org.apache.hadoop.yarn.api.records.NMToken;
|
||||
import org.apache.hadoop.yarn.api.records.NodeReport;
|
||||
|
|
@ -38,8 +36,6 @@ import org.apache.hadoop.yarn.api.records.PreemptionMessage;
|
|||
import org.apache.hadoop.yarn.api.records.Resource;
|
||||
import org.apache.hadoop.yarn.api.records.Token;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerResourceDecreasePBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerResourceIncreasePBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerStatusPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.NMTokenPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.NodeReportPBImpl;
|
||||
|
|
@ -48,8 +44,6 @@ import org.apache.hadoop.yarn.api.records.impl.pb.ProtoUtils;
|
|||
import org.apache.hadoop.yarn.api.records.impl.pb.ResourcePBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.TokenPBImpl;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceDecreaseProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceIncreaseProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerStatusProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.NodeReportProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.PreemptionMessageProto;
|
||||
|
|
@ -72,8 +66,8 @@ public class AllocateResponsePBImpl extends AllocateResponse {
|
|||
private List<Container> allocatedContainers = null;
|
||||
private List<NMToken> nmTokens = null;
|
||||
private List<ContainerStatus> completedContainersStatuses = null;
|
||||
private List<ContainerResourceIncrease> increasedContainers = null;
|
||||
private List<ContainerResourceDecrease> decreasedContainers = null;
|
||||
private List<Container> increasedContainers = null;
|
||||
private List<Container> decreasedContainers = null;
|
||||
|
||||
private List<NodeReport> updatedNodes = null;
|
||||
private PreemptionMessage preempt;
|
||||
|
|
@ -147,14 +141,14 @@ public class AllocateResponsePBImpl extends AllocateResponse {
|
|||
}
|
||||
if (this.increasedContainers != null) {
|
||||
builder.clearIncreasedContainers();
|
||||
Iterable<ContainerResourceIncreaseProto> iterable =
|
||||
getIncreaseProtoIterable(this.increasedContainers);
|
||||
Iterable<ContainerProto> iterable =
|
||||
getContainerProtoIterable(this.increasedContainers);
|
||||
builder.addAllIncreasedContainers(iterable);
|
||||
}
|
||||
if (this.decreasedContainers != null) {
|
||||
builder.clearDecreasedContainers();
|
||||
Iterable<ContainerResourceDecreaseProto> iterable =
|
||||
getChangeProtoIterable(this.decreasedContainers);
|
||||
Iterable<ContainerProto> iterable =
|
||||
getContainerProtoIterable(this.decreasedContainers);
|
||||
builder.addAllDecreasedContainers(iterable);
|
||||
}
|
||||
if (this.amrmToken != null) {
|
||||
|
|
@ -262,6 +256,36 @@ public class AllocateResponsePBImpl extends AllocateResponse {
|
|||
allocatedContainers.addAll(containers);
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized List<Container> getIncreasedContainers() {
|
||||
initLocalIncreasedContainerList();
|
||||
return this.increasedContainers;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void setIncreasedContainers(
|
||||
final List<Container> containers) {
|
||||
if (containers == null)
|
||||
return;
|
||||
initLocalIncreasedContainerList();
|
||||
increasedContainers.addAll(containers);
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized List<Container> getDecreasedContainers() {
|
||||
initLocalDecreasedContainerList();
|
||||
return this.decreasedContainers;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void setDecreasedContainers(
|
||||
final List<Container> containers) {
|
||||
if (containers == null)
|
||||
return;
|
||||
initLocalDecreasedContainerList();
|
||||
decreasedContainers.addAll(containers);
|
||||
}
|
||||
|
||||
//// Finished containers
|
||||
@Override
|
||||
public synchronized List<ContainerStatus> getCompletedContainersStatuses() {
|
||||
|
|
@ -332,37 +356,6 @@ public class AllocateResponsePBImpl extends AllocateResponse {
|
|||
this.preempt = preempt;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized List<ContainerResourceIncrease> getIncreasedContainers() {
|
||||
initLocalIncreasedContainerList();
|
||||
return increasedContainers;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void setIncreasedContainers(
|
||||
List<ContainerResourceIncrease> increasedContainers) {
|
||||
if (increasedContainers == null)
|
||||
return;
|
||||
initLocalIncreasedContainerList();
|
||||
this.increasedContainers.addAll(increasedContainers);
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized List<ContainerResourceDecrease> getDecreasedContainers() {
|
||||
initLocalDecreasedContainerList();
|
||||
return decreasedContainers;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void setDecreasedContainers(
|
||||
List<ContainerResourceDecrease> decreasedContainers) {
|
||||
if (decreasedContainers == null) {
|
||||
return;
|
||||
}
|
||||
initLocalDecreasedContainerList();
|
||||
this.decreasedContainers.addAll(decreasedContainers);
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized Token getAMRMToken() {
|
||||
AllocateResponseProtoOrBuilder p = viaProto ? proto : builder;
|
||||
|
|
@ -390,10 +383,10 @@ public class AllocateResponsePBImpl extends AllocateResponse {
|
|||
return;
|
||||
}
|
||||
AllocateResponseProtoOrBuilder p = viaProto ? proto : builder;
|
||||
List<ContainerResourceIncreaseProto> list = p.getIncreasedContainersList();
|
||||
increasedContainers = new ArrayList<ContainerResourceIncrease>();
|
||||
List<ContainerProto> list = p.getIncreasedContainersList();
|
||||
increasedContainers = new ArrayList<>();
|
||||
|
||||
for (ContainerResourceIncreaseProto c : list) {
|
||||
for (ContainerProto c : list) {
|
||||
increasedContainers.add(convertFromProtoFormat(c));
|
||||
}
|
||||
}
|
||||
|
|
@ -403,10 +396,10 @@ public class AllocateResponsePBImpl extends AllocateResponse {
|
|||
return;
|
||||
}
|
||||
AllocateResponseProtoOrBuilder p = viaProto ? proto : builder;
|
||||
List<ContainerResourceDecreaseProto> list = p.getDecreasedContainersList();
|
||||
decreasedContainers = new ArrayList<ContainerResourceDecrease>();
|
||||
List<ContainerProto> list = p.getDecreasedContainersList();
|
||||
decreasedContainers = new ArrayList<>();
|
||||
|
||||
for (ContainerResourceDecreaseProto c : list) {
|
||||
for (ContainerProto c : list) {
|
||||
decreasedContainers.add(convertFromProtoFormat(c));
|
||||
}
|
||||
}
|
||||
|
|
@ -453,70 +446,6 @@ public class AllocateResponsePBImpl extends AllocateResponse {
|
|||
}
|
||||
}
|
||||
|
||||
private synchronized Iterable<ContainerResourceIncreaseProto>
|
||||
getIncreaseProtoIterable(
|
||||
final List<ContainerResourceIncrease> newContainersList) {
|
||||
maybeInitBuilder();
|
||||
return new Iterable<ContainerResourceIncreaseProto>() {
|
||||
@Override
|
||||
public synchronized Iterator<ContainerResourceIncreaseProto> iterator() {
|
||||
return new Iterator<ContainerResourceIncreaseProto>() {
|
||||
|
||||
Iterator<ContainerResourceIncrease> iter = newContainersList
|
||||
.iterator();
|
||||
|
||||
@Override
|
||||
public synchronized boolean hasNext() {
|
||||
return iter.hasNext();
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized ContainerResourceIncreaseProto next() {
|
||||
return convertToProtoFormat(iter.next());
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void remove() {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
};
|
||||
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private synchronized Iterable<ContainerResourceDecreaseProto>
|
||||
getChangeProtoIterable(
|
||||
final List<ContainerResourceDecrease> newContainersList) {
|
||||
maybeInitBuilder();
|
||||
return new Iterable<ContainerResourceDecreaseProto>() {
|
||||
@Override
|
||||
public synchronized Iterator<ContainerResourceDecreaseProto> iterator() {
|
||||
return new Iterator<ContainerResourceDecreaseProto>() {
|
||||
|
||||
Iterator<ContainerResourceDecrease> iter = newContainersList
|
||||
.iterator();
|
||||
|
||||
@Override
|
||||
public synchronized boolean hasNext() {
|
||||
return iter.hasNext();
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized ContainerResourceDecreaseProto next() {
|
||||
return convertToProtoFormat(iter.next());
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void remove() {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
};
|
||||
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private synchronized Iterable<ContainerProto> getContainerProtoIterable(
|
||||
final List<Container> newContainersList) {
|
||||
maybeInitBuilder();
|
||||
|
|
@ -655,26 +584,6 @@ public class AllocateResponsePBImpl extends AllocateResponse {
|
|||
}
|
||||
}
|
||||
|
||||
private synchronized ContainerResourceIncrease convertFromProtoFormat(
|
||||
ContainerResourceIncreaseProto p) {
|
||||
return new ContainerResourceIncreasePBImpl(p);
|
||||
}
|
||||
|
||||
private synchronized ContainerResourceIncreaseProto convertToProtoFormat(
|
||||
ContainerResourceIncrease t) {
|
||||
return ((ContainerResourceIncreasePBImpl) t).getProto();
|
||||
}
|
||||
|
||||
private synchronized ContainerResourceDecrease convertFromProtoFormat(
|
||||
ContainerResourceDecreaseProto p) {
|
||||
return new ContainerResourceDecreasePBImpl(p);
|
||||
}
|
||||
|
||||
private synchronized ContainerResourceDecreaseProto convertToProtoFormat(
|
||||
ContainerResourceDecrease t) {
|
||||
return ((ContainerResourceDecreasePBImpl) t).getProto();
|
||||
}
|
||||
|
||||
private synchronized NodeReportPBImpl convertFromProtoFormat(
|
||||
NodeReportProto p) {
|
||||
return new NodeReportPBImpl(p);
|
||||
|
|
|
|||
|
|
@ -19,35 +19,35 @@
|
|||
package org.apache.hadoop.yarn.api.records.impl.pb;
|
||||
|
||||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceIncreaseRequest;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceChangeRequest;
|
||||
import org.apache.hadoop.yarn.api.records.Resource;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerIdProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceIncreaseRequestProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceIncreaseRequestProtoOrBuilder;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceChangeRequestProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceChangeRequestProtoOrBuilder;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ResourceProto;
|
||||
|
||||
|
||||
public class ContainerResourceIncreaseRequestPBImpl extends
|
||||
ContainerResourceIncreaseRequest {
|
||||
ContainerResourceIncreaseRequestProto proto =
|
||||
ContainerResourceIncreaseRequestProto.getDefaultInstance();
|
||||
ContainerResourceIncreaseRequestProto.Builder builder = null;
|
||||
public class ContainerResourceChangeRequestPBImpl extends
|
||||
ContainerResourceChangeRequest {
|
||||
ContainerResourceChangeRequestProto proto =
|
||||
ContainerResourceChangeRequestProto.getDefaultInstance();
|
||||
ContainerResourceChangeRequestProto.Builder builder = null;
|
||||
boolean viaProto = false;
|
||||
|
||||
private ContainerId existingContainerId = null;
|
||||
private Resource targetCapability = null;
|
||||
|
||||
public ContainerResourceIncreaseRequestPBImpl() {
|
||||
builder = ContainerResourceIncreaseRequestProto.newBuilder();
|
||||
public ContainerResourceChangeRequestPBImpl() {
|
||||
builder = ContainerResourceChangeRequestProto.newBuilder();
|
||||
}
|
||||
|
||||
public ContainerResourceIncreaseRequestPBImpl(
|
||||
ContainerResourceIncreaseRequestProto proto) {
|
||||
public ContainerResourceChangeRequestPBImpl(
|
||||
ContainerResourceChangeRequestProto proto) {
|
||||
this.proto = proto;
|
||||
viaProto = true;
|
||||
}
|
||||
|
||||
public ContainerResourceIncreaseRequestProto getProto() {
|
||||
public ContainerResourceChangeRequestProto getProto() {
|
||||
mergeLocalToProto();
|
||||
proto = viaProto ? proto : builder.build();
|
||||
viaProto = true;
|
||||
|
|
@ -56,7 +56,7 @@ public class ContainerResourceIncreaseRequestPBImpl extends
|
|||
|
||||
@Override
|
||||
public ContainerId getContainerId() {
|
||||
ContainerResourceIncreaseRequestProtoOrBuilder p = viaProto ? proto
|
||||
ContainerResourceChangeRequestProtoOrBuilder p = viaProto ? proto
|
||||
: builder;
|
||||
if (this.existingContainerId != null) {
|
||||
return this.existingContainerId;
|
||||
|
|
@ -78,7 +78,7 @@ public class ContainerResourceIncreaseRequestPBImpl extends
|
|||
|
||||
@Override
|
||||
public Resource getCapability() {
|
||||
ContainerResourceIncreaseRequestProtoOrBuilder p = viaProto ? proto
|
||||
ContainerResourceChangeRequestProtoOrBuilder p = viaProto ? proto
|
||||
: builder;
|
||||
if (this.targetCapability != null) {
|
||||
return this.targetCapability;
|
||||
|
|
@ -125,7 +125,7 @@ public class ContainerResourceIncreaseRequestPBImpl extends
|
|||
|
||||
private void maybeInitBuilder() {
|
||||
if (viaProto || builder == null) {
|
||||
builder = ContainerResourceIncreaseRequestProto.newBuilder(proto);
|
||||
builder = ContainerResourceChangeRequestProto.newBuilder(proto);
|
||||
}
|
||||
viaProto = false;
|
||||
}
|
||||
|
|
@ -1,136 +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.api.records.impl.pb;
|
||||
|
||||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceDecrease;
|
||||
import org.apache.hadoop.yarn.api.records.Resource;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerIdProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceDecreaseProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceDecreaseProtoOrBuilder;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ResourceProto;
|
||||
|
||||
public class ContainerResourceDecreasePBImpl extends ContainerResourceDecrease {
|
||||
ContainerResourceDecreaseProto proto = ContainerResourceDecreaseProto
|
||||
.getDefaultInstance();
|
||||
ContainerResourceDecreaseProto.Builder builder = null;
|
||||
boolean viaProto = false;
|
||||
|
||||
private ContainerId existingContainerId = null;
|
||||
private Resource targetCapability = null;
|
||||
|
||||
public ContainerResourceDecreasePBImpl() {
|
||||
builder = ContainerResourceDecreaseProto.newBuilder();
|
||||
}
|
||||
|
||||
public ContainerResourceDecreasePBImpl(ContainerResourceDecreaseProto proto) {
|
||||
this.proto = proto;
|
||||
viaProto = true;
|
||||
}
|
||||
|
||||
public ContainerResourceDecreaseProto getProto() {
|
||||
mergeLocalToProto();
|
||||
proto = viaProto ? proto : builder.build();
|
||||
viaProto = true;
|
||||
return proto;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ContainerId getContainerId() {
|
||||
ContainerResourceDecreaseProtoOrBuilder p = viaProto ? proto : builder;
|
||||
if (this.existingContainerId != null) {
|
||||
return this.existingContainerId;
|
||||
}
|
||||
if (p.hasContainerId()) {
|
||||
this.existingContainerId = convertFromProtoFormat(p.getContainerId());
|
||||
}
|
||||
return this.existingContainerId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setContainerId(ContainerId existingContainerId) {
|
||||
maybeInitBuilder();
|
||||
if (existingContainerId == null) {
|
||||
builder.clearContainerId();
|
||||
}
|
||||
this.existingContainerId = existingContainerId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Resource getCapability() {
|
||||
ContainerResourceDecreaseProtoOrBuilder p = viaProto ? proto : builder;
|
||||
if (this.targetCapability != null) {
|
||||
return this.targetCapability;
|
||||
}
|
||||
if (p.hasCapability()) {
|
||||
this.targetCapability = convertFromProtoFormat(p.getCapability());
|
||||
}
|
||||
return this.targetCapability;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setCapability(Resource targetCapability) {
|
||||
maybeInitBuilder();
|
||||
if (targetCapability == null) {
|
||||
builder.clearCapability();
|
||||
}
|
||||
this.targetCapability = targetCapability;
|
||||
}
|
||||
|
||||
private ContainerIdPBImpl convertFromProtoFormat(ContainerIdProto p) {
|
||||
return new ContainerIdPBImpl(p);
|
||||
}
|
||||
|
||||
private ContainerIdProto convertToProtoFormat(ContainerId t) {
|
||||
return ((ContainerIdPBImpl) t).getProto();
|
||||
}
|
||||
|
||||
private Resource convertFromProtoFormat(ResourceProto p) {
|
||||
return new ResourcePBImpl(p);
|
||||
}
|
||||
|
||||
private ResourceProto convertToProtoFormat(Resource t) {
|
||||
return ((ResourcePBImpl) t).getProto();
|
||||
}
|
||||
|
||||
private void mergeLocalToProto() {
|
||||
if (viaProto) {
|
||||
maybeInitBuilder();
|
||||
}
|
||||
mergeLocalToBuilder();
|
||||
proto = builder.build();
|
||||
viaProto = true;
|
||||
}
|
||||
|
||||
private void maybeInitBuilder() {
|
||||
if (viaProto || builder == null) {
|
||||
builder = ContainerResourceDecreaseProto.newBuilder(proto);
|
||||
}
|
||||
viaProto = false;
|
||||
}
|
||||
|
||||
private void mergeLocalToBuilder() {
|
||||
if (this.existingContainerId != null) {
|
||||
builder.setContainerId(convertToProtoFormat(this.existingContainerId));
|
||||
}
|
||||
if (this.targetCapability != null) {
|
||||
builder.setCapability(convertToProtoFormat(this.targetCapability));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,171 +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.api.records.impl.pb;
|
||||
|
||||
import org.apache.hadoop.security.proto.SecurityProtos.TokenProto;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceIncrease;
|
||||
import org.apache.hadoop.yarn.api.records.Resource;
|
||||
import org.apache.hadoop.yarn.api.records.Token;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerIdProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceIncreaseProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceIncreaseProtoOrBuilder;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ResourceProto;
|
||||
|
||||
public class ContainerResourceIncreasePBImpl extends ContainerResourceIncrease {
|
||||
ContainerResourceIncreaseProto proto = ContainerResourceIncreaseProto
|
||||
.getDefaultInstance();
|
||||
ContainerResourceIncreaseProto.Builder builder = null;
|
||||
boolean viaProto = false;
|
||||
|
||||
private ContainerId existingContainerId = null;
|
||||
private Resource targetCapability = null;
|
||||
private Token token = null;
|
||||
|
||||
public ContainerResourceIncreasePBImpl() {
|
||||
builder = ContainerResourceIncreaseProto.newBuilder();
|
||||
}
|
||||
|
||||
public ContainerResourceIncreasePBImpl(ContainerResourceIncreaseProto proto) {
|
||||
this.proto = proto;
|
||||
viaProto = true;
|
||||
}
|
||||
|
||||
public ContainerResourceIncreaseProto getProto() {
|
||||
mergeLocalToProto();
|
||||
proto = viaProto ? proto : builder.build();
|
||||
viaProto = true;
|
||||
return proto;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ContainerId getContainerId() {
|
||||
ContainerResourceIncreaseProtoOrBuilder p = viaProto ? proto : builder;
|
||||
if (this.existingContainerId != null) {
|
||||
return this.existingContainerId;
|
||||
}
|
||||
if (p.hasContainerId()) {
|
||||
this.existingContainerId = convertFromProtoFormat(p.getContainerId());
|
||||
}
|
||||
return this.existingContainerId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setContainerId(ContainerId existingContainerId) {
|
||||
maybeInitBuilder();
|
||||
if (existingContainerId == null) {
|
||||
builder.clearContainerId();
|
||||
}
|
||||
this.existingContainerId = existingContainerId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Resource getCapability() {
|
||||
ContainerResourceIncreaseProtoOrBuilder p = viaProto ? proto : builder;
|
||||
if (this.targetCapability != null) {
|
||||
return this.targetCapability;
|
||||
}
|
||||
if (p.hasCapability()) {
|
||||
this.targetCapability = convertFromProtoFormat(p.getCapability());
|
||||
}
|
||||
return this.targetCapability;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setCapability(Resource targetCapability) {
|
||||
maybeInitBuilder();
|
||||
if (targetCapability == null) {
|
||||
builder.clearCapability();
|
||||
}
|
||||
this.targetCapability = targetCapability;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Token getContainerToken() {
|
||||
ContainerResourceIncreaseProtoOrBuilder p = viaProto ? proto : builder;
|
||||
if (this.token != null) {
|
||||
return this.token;
|
||||
}
|
||||
if (p.hasContainerToken()) {
|
||||
this.token = convertFromProtoFormat(p.getContainerToken());
|
||||
}
|
||||
return this.token;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setContainerToken(Token token) {
|
||||
maybeInitBuilder();
|
||||
if (token == null) {
|
||||
builder.clearContainerToken();
|
||||
}
|
||||
this.token = token;
|
||||
}
|
||||
|
||||
private ContainerIdPBImpl convertFromProtoFormat(ContainerIdProto p) {
|
||||
return new ContainerIdPBImpl(p);
|
||||
}
|
||||
|
||||
private ContainerIdProto convertToProtoFormat(ContainerId t) {
|
||||
return ((ContainerIdPBImpl) t).getProto();
|
||||
}
|
||||
|
||||
private Resource convertFromProtoFormat(ResourceProto p) {
|
||||
return new ResourcePBImpl(p);
|
||||
}
|
||||
|
||||
private ResourceProto convertToProtoFormat(Resource t) {
|
||||
return ((ResourcePBImpl) t).getProto();
|
||||
}
|
||||
|
||||
private Token convertFromProtoFormat(TokenProto p) {
|
||||
return new TokenPBImpl(p);
|
||||
}
|
||||
|
||||
private TokenProto convertToProtoFormat(Token t) {
|
||||
return ((TokenPBImpl) t).getProto();
|
||||
}
|
||||
|
||||
private void mergeLocalToProto() {
|
||||
if (viaProto) {
|
||||
maybeInitBuilder();
|
||||
}
|
||||
mergeLocalToBuilder();
|
||||
proto = builder.build();
|
||||
viaProto = true;
|
||||
}
|
||||
|
||||
private void maybeInitBuilder() {
|
||||
if (viaProto || builder == null) {
|
||||
builder = ContainerResourceIncreaseProto.newBuilder(proto);
|
||||
}
|
||||
viaProto = false;
|
||||
}
|
||||
|
||||
private void mergeLocalToBuilder() {
|
||||
if (this.existingContainerId != null) {
|
||||
builder.setContainerId(convertToProtoFormat(this.existingContainerId));
|
||||
}
|
||||
if (this.targetCapability != null) {
|
||||
builder.setCapability(convertToProtoFormat(this.targetCapability));
|
||||
}
|
||||
if (this.token != null) {
|
||||
builder.setContainerToken(convertToProtoFormat(this.token));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -24,6 +24,8 @@ import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
|||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerState;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerStatus;
|
||||
import org.apache.hadoop.yarn.api.records.Resource;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ResourceProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerIdProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerStateProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerStatusProto;
|
||||
|
|
@ -78,6 +80,7 @@ public class ContainerStatusPBImpl extends ContainerStatus {
|
|||
sb.append("ContainerStatus: [");
|
||||
sb.append("ContainerId: ").append(getContainerId()).append(", ");
|
||||
sb.append("State: ").append(getState()).append(", ");
|
||||
sb.append("Capability: ").append(getCapability()).append(", ");
|
||||
sb.append("Diagnostics: ").append(getDiagnostics()).append(", ");
|
||||
sb.append("ExitStatus: ").append(getExitStatus()).append(", ");
|
||||
sb.append("]");
|
||||
|
|
@ -168,6 +171,25 @@ public class ContainerStatusPBImpl extends ContainerStatus {
|
|||
builder.setDiagnostics(diagnostics);
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized Resource getCapability() {
|
||||
ContainerStatusProtoOrBuilder p = viaProto ? proto : builder;
|
||||
if (!p.hasCapability()) {
|
||||
return null;
|
||||
}
|
||||
return convertFromProtoFormat(p.getCapability());
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void setCapability(Resource capability) {
|
||||
maybeInitBuilder();
|
||||
if (capability == null) {
|
||||
builder.clearCapability();
|
||||
return;
|
||||
}
|
||||
builder.setCapability(convertToProtoFormat(capability));
|
||||
}
|
||||
|
||||
private ContainerStateProto convertToProtoFormat(ContainerState e) {
|
||||
return ProtoUtils.convertToProtoFormat(e);
|
||||
}
|
||||
|
|
@ -184,6 +206,11 @@ public class ContainerStatusPBImpl extends ContainerStatus {
|
|||
return ((ContainerIdPBImpl)t).getProto();
|
||||
}
|
||||
|
||||
|
||||
|
||||
private ResourceProto convertToProtoFormat(Resource e) {
|
||||
return ((ResourcePBImpl)e).getProto();
|
||||
}
|
||||
|
||||
private ResourcePBImpl convertFromProtoFormat(ResourceProto p) {
|
||||
return new ResourcePBImpl(p);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,73 +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.api;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import org.junit.Assert;
|
||||
|
||||
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.Resource;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceIncreaseRequest;
|
||||
import org.apache.hadoop.yarn.proto.YarnServiceProtos.AllocateRequestProto;
|
||||
import org.junit.Test;
|
||||
|
||||
public class TestAllocateRequest {
|
||||
@Test
|
||||
public void testAllcoateRequestWithIncrease() {
|
||||
List<ContainerResourceIncreaseRequest> incRequests =
|
||||
new ArrayList<ContainerResourceIncreaseRequest>();
|
||||
for (int i = 0; i < 3; i++) {
|
||||
incRequests.add(ContainerResourceIncreaseRequest.newInstance(null,
|
||||
Resource.newInstance(0, i)));
|
||||
}
|
||||
AllocateRequest r =
|
||||
AllocateRequest.newInstance(123, 0f, null, null, null, incRequests);
|
||||
|
||||
// serde
|
||||
AllocateRequestProto p = ((AllocateRequestPBImpl) r).getProto();
|
||||
r = new AllocateRequestPBImpl(p);
|
||||
|
||||
// check value
|
||||
Assert.assertEquals(123, r.getResponseId());
|
||||
Assert.assertEquals(incRequests.size(), r.getIncreaseRequests().size());
|
||||
|
||||
for (int i = 0; i < incRequests.size(); i++) {
|
||||
Assert.assertEquals(r.getIncreaseRequests().get(i).getCapability()
|
||||
.getVirtualCores(), incRequests.get(i).getCapability()
|
||||
.getVirtualCores());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAllcoateRequestWithoutIncrease() {
|
||||
AllocateRequest r =
|
||||
AllocateRequest.newInstance(123, 0f, null, null, null, null);
|
||||
|
||||
// serde
|
||||
AllocateRequestProto p = ((AllocateRequestPBImpl) r).getProto();
|
||||
r = new AllocateRequestPBImpl(p);
|
||||
|
||||
// check value
|
||||
Assert.assertEquals(123, r.getResponseId());
|
||||
Assert.assertEquals(0, r.getIncreaseRequests().size());
|
||||
}
|
||||
}
|
||||
|
|
@ -1,114 +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.api;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
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.AMCommand;
|
||||
import org.apache.hadoop.yarn.api.records.Container;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceDecrease;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceIncrease;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerStatus;
|
||||
import org.apache.hadoop.yarn.api.records.NMToken;
|
||||
import org.apache.hadoop.yarn.api.records.NodeReport;
|
||||
import org.apache.hadoop.yarn.api.records.Resource;
|
||||
import org.apache.hadoop.yarn.proto.YarnServiceProtos.AllocateResponseProto;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
public class TestAllocateResponse {
|
||||
@SuppressWarnings("deprecation")
|
||||
@Test
|
||||
public void testAllocateResponseWithIncDecContainers() {
|
||||
List<ContainerResourceIncrease> incContainers =
|
||||
new ArrayList<ContainerResourceIncrease>();
|
||||
List<ContainerResourceDecrease> decContainers =
|
||||
new ArrayList<ContainerResourceDecrease>();
|
||||
for (int i = 0; i < 3; i++) {
|
||||
incContainers.add(ContainerResourceIncrease.newInstance(null,
|
||||
Resource.newInstance(1024, i), null));
|
||||
}
|
||||
for (int i = 0; i < 5; i++) {
|
||||
decContainers.add(ContainerResourceDecrease.newInstance(null,
|
||||
Resource.newInstance(1024, i)));
|
||||
}
|
||||
|
||||
AllocateResponse r =
|
||||
AllocateResponse.newInstance(3, new ArrayList<ContainerStatus>(),
|
||||
new ArrayList<Container>(), new ArrayList<NodeReport>(), null,
|
||||
AMCommand.AM_RESYNC, 3, null, new ArrayList<NMToken>(),
|
||||
incContainers, decContainers);
|
||||
|
||||
// serde
|
||||
AllocateResponseProto p = ((AllocateResponsePBImpl) r).getProto();
|
||||
r = new AllocateResponsePBImpl(p);
|
||||
|
||||
// check value
|
||||
Assert
|
||||
.assertEquals(incContainers.size(), r.getIncreasedContainers().size());
|
||||
Assert
|
||||
.assertEquals(decContainers.size(), r.getDecreasedContainers().size());
|
||||
|
||||
for (int i = 0; i < incContainers.size(); i++) {
|
||||
Assert.assertEquals(i, r.getIncreasedContainers().get(i).getCapability()
|
||||
.getVirtualCores());
|
||||
}
|
||||
|
||||
for (int i = 0; i < decContainers.size(); i++) {
|
||||
Assert.assertEquals(i, r.getDecreasedContainers().get(i).getCapability()
|
||||
.getVirtualCores());
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("deprecation")
|
||||
@Test
|
||||
public void testAllocateResponseWithoutIncDecContainers() {
|
||||
AllocateResponse r =
|
||||
AllocateResponse.newInstance(3, new ArrayList<ContainerStatus>(),
|
||||
new ArrayList<Container>(), new ArrayList<NodeReport>(), null,
|
||||
AMCommand.AM_RESYNC, 3, null, new ArrayList<NMToken>(), null, null);
|
||||
|
||||
// serde
|
||||
AllocateResponseProto p = ((AllocateResponsePBImpl) r).getProto();
|
||||
r = new AllocateResponsePBImpl(p);
|
||||
|
||||
// check value
|
||||
Assert.assertEquals(0, r.getIncreasedContainers().size());
|
||||
Assert.assertEquals(0, r.getDecreasedContainers().size());
|
||||
}
|
||||
}
|
||||
|
|
@ -1,66 +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.api;
|
||||
|
||||
import org.junit.Assert;
|
||||
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceDecrease;
|
||||
import org.apache.hadoop.yarn.api.records.Resource;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerResourceDecreasePBImpl;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceDecreaseProto;
|
||||
import org.junit.Test;
|
||||
|
||||
public class TestContainerResourceDecrease {
|
||||
@Test
|
||||
public void testResourceDecreaseContext() {
|
||||
ContainerId containerId = ContainerId
|
||||
.newContainerId(ApplicationAttemptId.newInstance(
|
||||
ApplicationId.newInstance(1234, 3), 3), 7);
|
||||
Resource resource = Resource.newInstance(1023, 3);
|
||||
ContainerResourceDecrease ctx = ContainerResourceDecrease.newInstance(
|
||||
containerId, resource);
|
||||
|
||||
// get proto and recover to ctx
|
||||
ContainerResourceDecreaseProto proto =
|
||||
((ContainerResourceDecreasePBImpl) ctx).getProto();
|
||||
ctx = new ContainerResourceDecreasePBImpl(proto);
|
||||
|
||||
// check values
|
||||
Assert.assertEquals(ctx.getCapability(), resource);
|
||||
Assert.assertEquals(ctx.getContainerId(), containerId);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testResourceDecreaseContextWithNull() {
|
||||
ContainerResourceDecrease ctx = ContainerResourceDecrease.newInstance(null,
|
||||
null);
|
||||
|
||||
// get proto and recover to ctx;
|
||||
ContainerResourceDecreaseProto proto =
|
||||
((ContainerResourceDecreasePBImpl) ctx).getProto();
|
||||
ctx = new ContainerResourceDecreasePBImpl(proto);
|
||||
|
||||
// check values
|
||||
Assert.assertNull(ctx.getCapability());
|
||||
Assert.assertNull(ctx.getContainerId());
|
||||
}
|
||||
}
|
||||
|
|
@ -1,74 +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.api;
|
||||
|
||||
import java.util.Arrays;
|
||||
|
||||
import org.junit.Assert;
|
||||
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceIncrease;
|
||||
import org.apache.hadoop.yarn.api.records.Resource;
|
||||
import org.apache.hadoop.yarn.api.records.Token;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerResourceIncreasePBImpl;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceIncreaseProto;
|
||||
import org.junit.Test;
|
||||
|
||||
public class TestContainerResourceIncrease {
|
||||
@Test
|
||||
public void testResourceIncreaseContext() {
|
||||
byte[] identifier = new byte[] { 1, 2, 3, 4 };
|
||||
Token token = Token.newInstance(identifier, "", "".getBytes(), "");
|
||||
ContainerId containerId = ContainerId
|
||||
.newContainerId(ApplicationAttemptId.newInstance(
|
||||
ApplicationId.newInstance(1234, 3), 3), 7);
|
||||
Resource resource = Resource.newInstance(1023, 3);
|
||||
ContainerResourceIncrease ctx = ContainerResourceIncrease.newInstance(
|
||||
containerId, resource, token);
|
||||
|
||||
// get proto and recover to ctx
|
||||
ContainerResourceIncreaseProto proto =
|
||||
((ContainerResourceIncreasePBImpl) ctx).getProto();
|
||||
ctx = new ContainerResourceIncreasePBImpl(proto);
|
||||
|
||||
// check values
|
||||
Assert.assertEquals(ctx.getCapability(), resource);
|
||||
Assert.assertEquals(ctx.getContainerId(), containerId);
|
||||
Assert.assertTrue(Arrays.equals(ctx.getContainerToken().getIdentifier()
|
||||
.array(), identifier));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testResourceIncreaseContextWithNull() {
|
||||
ContainerResourceIncrease ctx = ContainerResourceIncrease.newInstance(null,
|
||||
null, null);
|
||||
|
||||
// get proto and recover to ctx;
|
||||
ContainerResourceIncreaseProto proto =
|
||||
((ContainerResourceIncreasePBImpl) ctx).getProto();
|
||||
ctx = new ContainerResourceIncreasePBImpl(proto);
|
||||
|
||||
// check values
|
||||
Assert.assertNull(ctx.getContainerToken());
|
||||
Assert.assertNull(ctx.getCapability());
|
||||
Assert.assertNull(ctx.getContainerId());
|
||||
}
|
||||
}
|
||||
|
|
@ -1,68 +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.api;
|
||||
|
||||
import org.junit.Assert;
|
||||
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
|
||||
import org.apache.hadoop.yarn.api.records.ApplicationId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceIncreaseRequest;
|
||||
import org.apache.hadoop.yarn.api.records.Resource;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerResourceIncreaseRequestPBImpl;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceIncreaseRequestProto;
|
||||
import org.junit.Test;
|
||||
|
||||
public class TestContainerResourceIncreaseRequest {
|
||||
@Test
|
||||
public void ContainerResourceIncreaseRequest() {
|
||||
ContainerId containerId = ContainerId
|
||||
.newContainerId(ApplicationAttemptId.newInstance(
|
||||
ApplicationId.newInstance(1234, 3), 3), 7);
|
||||
Resource resource = Resource.newInstance(1023, 3);
|
||||
ContainerResourceIncreaseRequest context = ContainerResourceIncreaseRequest
|
||||
.newInstance(containerId, resource);
|
||||
|
||||
// to proto and get it back
|
||||
ContainerResourceIncreaseRequestProto proto =
|
||||
((ContainerResourceIncreaseRequestPBImpl) context).getProto();
|
||||
ContainerResourceIncreaseRequest contextRecover =
|
||||
new ContainerResourceIncreaseRequestPBImpl(proto);
|
||||
|
||||
// check value
|
||||
Assert.assertEquals(contextRecover.getContainerId(), containerId);
|
||||
Assert.assertEquals(contextRecover.getCapability(), resource);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testResourceChangeContextWithNullField() {
|
||||
ContainerResourceIncreaseRequest context = ContainerResourceIncreaseRequest
|
||||
.newInstance(null, null);
|
||||
|
||||
// to proto and get it back
|
||||
ContainerResourceIncreaseRequestProto proto =
|
||||
((ContainerResourceIncreaseRequestPBImpl) context).getProto();
|
||||
ContainerResourceIncreaseRequest contextRecover =
|
||||
new ContainerResourceIncreaseRequestPBImpl(proto);
|
||||
|
||||
// check value
|
||||
Assert.assertNull(contextRecover.getContainerId());
|
||||
Assert.assertNull(contextRecover.getCapability());
|
||||
}
|
||||
}
|
||||
|
|
@ -113,9 +113,7 @@ import org.apache.hadoop.yarn.api.records.Container;
|
|||
import org.apache.hadoop.yarn.api.records.ContainerId;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerReport;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceDecrease;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceIncrease;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceIncreaseRequest;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerResourceChangeRequest;
|
||||
import org.apache.hadoop.yarn.api.records.ContainerStatus;
|
||||
import org.apache.hadoop.yarn.api.records.LocalResource;
|
||||
import org.apache.hadoop.yarn.api.records.LogAggregationContext;
|
||||
|
|
@ -155,9 +153,7 @@ import org.apache.hadoop.yarn.api.records.impl.pb.ContainerIdPBImpl;
|
|||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerLaunchContextPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerReportPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerResourceDecreasePBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerResourceIncreasePBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerResourceIncreaseRequestPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerResourceChangeRequestPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerStatusPBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.LocalResourcePBImpl;
|
||||
import org.apache.hadoop.yarn.api.records.impl.pb.NMTokenPBImpl;
|
||||
|
|
@ -190,9 +186,7 @@ import org.apache.hadoop.yarn.proto.YarnProtos.ContainerIdProto;
|
|||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerLaunchContextProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerReportProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceDecreaseProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceIncreaseProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceIncreaseRequestProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerResourceChangeRequestProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.ContainerStatusProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.LocalResourceProto;
|
||||
import org.apache.hadoop.yarn.proto.YarnProtos.NodeIdProto;
|
||||
|
|
@ -471,9 +465,7 @@ public class TestPBImplRecords {
|
|||
generateByNewInstance(ContainerLaunchContext.class);
|
||||
generateByNewInstance(ApplicationSubmissionContext.class);
|
||||
generateByNewInstance(ContainerReport.class);
|
||||
generateByNewInstance(ContainerResourceDecrease.class);
|
||||
generateByNewInstance(ContainerResourceIncrease.class);
|
||||
generateByNewInstance(ContainerResourceIncreaseRequest.class);
|
||||
generateByNewInstance(ContainerResourceChangeRequest.class);
|
||||
generateByNewInstance(ContainerStatus.class);
|
||||
generateByNewInstance(PreemptionContainer.class);
|
||||
generateByNewInstance(PreemptionResourceRequest.class);
|
||||
|
|
@ -959,21 +951,9 @@ public class TestPBImplRecords {
|
|||
}
|
||||
|
||||
@Test
|
||||
public void testContainerResourceDecreasePBImpl() throws Exception {
|
||||
validatePBImplRecord(ContainerResourceDecreasePBImpl.class,
|
||||
ContainerResourceDecreaseProto.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testContainerResourceIncreasePBImpl() throws Exception {
|
||||
validatePBImplRecord(ContainerResourceIncreasePBImpl.class,
|
||||
ContainerResourceIncreaseProto.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testContainerResourceIncreaseRequestPBImpl() throws Exception {
|
||||
validatePBImplRecord(ContainerResourceIncreaseRequestPBImpl.class,
|
||||
ContainerResourceIncreaseRequestProto.class);
|
||||
public void testContainerResourceChangeRequestPBImpl() throws Exception {
|
||||
validatePBImplRecord(ContainerResourceChangeRequestPBImpl.class,
|
||||
ContainerResourceChangeRequestProto.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
|
|||
Loading…
Reference in New Issue