From 1075dc81d42c9ad9b9300c62af5880287cf91099 Mon Sep 17 00:00:00 2001 From: Rohith Sharma K S Date: Wed, 21 Sep 2016 16:51:17 +0530 Subject: [PATCH] YARN-5611 --- .../apache/hadoop/mapred/TestClientRedirect.java | 9 ++ .../hadoop/yarn/api/ApplicationClientProtocol.java | 20 +++ .../UpdateApplicationTimeoutsRequest.java | 72 +++++++++ .../UpdateApplicationTimeoutsResponse.java | 34 ++++ .../main/proto/applicationclient_protocol.proto | 1 + .../src/main/proto/yarn_service_protos.proto | 8 + .../ApplicationClientProtocolPBClientImpl.java | 21 ++- .../ApplicationClientProtocolPBServiceImpl.java | 22 +++ .../pb/UpdateApplicationTimeoutsRequestPBImpl.java | 173 +++++++++++++++++++++ .../UpdateApplicationTimeoutsResponsePBImpl.java | 73 +++++++++ .../amrmproxy/MockResourceManagerFacade.java | 9 ++ .../server/resourcemanager/ClientRMService.java | 138 ++++++++++++---- .../yarn/server/resourcemanager/RMAuditLogger.java | 4 +- .../rmapp/RMAppLifetimeMonitor.java | 44 ++++-- .../rmapp/TestApplicationLifetimeMonitor.java | 32 +++- 15 files changed, 610 insertions(+), 50 deletions(-) create mode 100644 hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/UpdateApplicationTimeoutsRequest.java create mode 100644 hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/UpdateApplicationTimeoutsResponse.java create mode 100644 hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/UpdateApplicationTimeoutsRequestPBImpl.java create mode 100644 hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/UpdateApplicationTimeoutsResponsePBImpl.java diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestClientRedirect.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestClientRedirect.java index 255f998..65eac65 100644 --- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestClientRedirect.java +++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapred/TestClientRedirect.java @@ -124,6 +124,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationResponse; import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationPriorityRequest; import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationPriorityResponse; +import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationTimeoutsRequest; +import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationTimeoutsResponse; import org.apache.hadoop.yarn.api.records.ApplicationId; import org.apache.hadoop.yarn.api.records.ApplicationReport; import org.apache.hadoop.yarn.api.records.FinalApplicationStatus; @@ -485,6 +487,13 @@ public SignalContainerResponse signalToContainer( SignalContainerRequest request) throws IOException { return null; } + + @Override + public UpdateApplicationTimeoutsResponse updateApplicationTimeouts( + UpdateApplicationTimeoutsRequest request) + throws YarnException, IOException { + return null; + } } class HistoryService extends AMService implements HSClientProtocol { diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/ApplicationClientProtocol.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/ApplicationClientProtocol.java index 8ee43fb..883aed1 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/ApplicationClientProtocol.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/ApplicationClientProtocol.java @@ -59,6 +59,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.ReservationUpdateResponse; import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationPriorityRequest; import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationPriorityResponse; +import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationTimeoutsRequest; +import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationTimeoutsResponse; import org.apache.hadoop.yarn.api.protocolrecords.SignalContainerRequest; import org.apache.hadoop.yarn.api.protocolrecords.SignalContainerResponse; import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationRequest; @@ -566,4 +568,22 @@ public UpdateApplicationPriorityResponse updateApplicationPriority( SignalContainerResponse signalToContainer( SignalContainerRequest request) throws YarnException, IOException; + + /** + *

+ * The interface used by client to set ApplicationTimeouts of an application. + *

+ * Note : It is a blocking call + * + * @param request to set ApplicationTimeouts of an application + * @return an empty response + * @throws YarnException + * @throws IOException + */ + @Public + @Unstable + @Idempotent + public UpdateApplicationTimeoutsResponse updateApplicationTimeouts( + UpdateApplicationTimeoutsRequest request) + throws YarnException, IOException; } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/UpdateApplicationTimeoutsRequest.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/UpdateApplicationTimeoutsRequest.java new file mode 100644 index 0000000..52faf5d --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/UpdateApplicationTimeoutsRequest.java @@ -0,0 +1,72 @@ +/** + * 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.protocolrecords; + +import org.apache.hadoop.classification.InterfaceAudience.Public; +import org.apache.hadoop.classification.InterfaceStability.Unstable; +import org.apache.hadoop.yarn.api.records.ApplicationId; +import org.apache.hadoop.yarn.api.records.ApplicationTimeouts; +import org.apache.hadoop.yarn.util.Records; + +@Public +@Unstable +public abstract class UpdateApplicationTimeoutsRequest { + public static UpdateApplicationTimeoutsRequest newInstance( + ApplicationId applicationId, + ApplicationTimeouts applicationTimeouts) { + UpdateApplicationTimeoutsRequest request = + Records.newRecord(UpdateApplicationTimeoutsRequest.class); + request.setApplicationId(applicationId); + request.setApplicationTimeouts(applicationTimeouts); + return request; + } + + /** + * Get the ApplicationId of the application. + * + * @return ApplicationId of the application + */ + public abstract ApplicationId getApplicationId(); + + /** + * Set the ApplicationId of the application. + * + * @param applicationId ApplicationId of the application + */ + public abstract void setApplicationId(ApplicationId applicationId); + + /** + * Get ApplicationTimeouts of the application. + * + * @return ApplicationTimeouts of the application. + */ + @Public + @Unstable + public abstract ApplicationTimeouts getApplicationTimeouts(); + + /** + * Get ApplicationTimeouts of the application. + * + * @param applicationTimeouts for the application. + */ + @Public + @Unstable + public abstract void setApplicationTimeouts( + ApplicationTimeouts applicationTimeouts); +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/UpdateApplicationTimeoutsResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/UpdateApplicationTimeoutsResponse.java new file mode 100644 index 0000000..3217569 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/UpdateApplicationTimeoutsResponse.java @@ -0,0 +1,34 @@ +/** + * 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.protocolrecords; + +import org.apache.hadoop.classification.InterfaceAudience.Public; +import org.apache.hadoop.classification.InterfaceStability.Unstable; +import org.apache.hadoop.yarn.util.Records; + +@Public +@Unstable +public abstract class UpdateApplicationTimeoutsResponse { + + public static UpdateApplicationTimeoutsResponse newInstance() { + UpdateApplicationTimeoutsResponse response = + Records.newRecord(UpdateApplicationTimeoutsResponse.class); + return response; + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/applicationclient_protocol.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/applicationclient_protocol.proto index f1c3839..ba79db0 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/applicationclient_protocol.proto +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/applicationclient_protocol.proto @@ -60,4 +60,5 @@ service ApplicationClientProtocolService { rpc getClusterNodeLabels (GetClusterNodeLabelsRequestProto) returns (GetClusterNodeLabelsResponseProto); rpc updateApplicationPriority (UpdateApplicationPriorityRequestProto) returns (UpdateApplicationPriorityResponseProto); rpc signalToContainer(SignalContainerRequestProto) returns (SignalContainerResponseProto); + rpc updateApplicationTimeouts (UpdateApplicationTimeoutsRequestProto) returns (UpdateApplicationTimeoutsResponseProto); } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto index 1385ea4..8bf27d8 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto @@ -267,6 +267,14 @@ message SignalContainerRequestProto { message SignalContainerResponseProto { } +message UpdateApplicationTimeoutsRequestProto { + required ApplicationIdProto applicationId = 1; + required ApplicationTimeoutsProto application_timeouts = 2; +} + +message UpdateApplicationTimeoutsResponseProto { +} + ////////////////////////////////////////////////////// /////// client_NM_Protocol /////////////////////////// ////////////////////////////////////////////////////// diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/ApplicationClientProtocolPBClientImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/ApplicationClientProtocolPBClientImpl.java index 2d755a2..ad7cb29 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/ApplicationClientProtocolPBClientImpl.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/ApplicationClientProtocolPBClientImpl.java @@ -83,6 +83,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.ReservationUpdateResponse; import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationPriorityRequest; import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationPriorityResponse; +import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationTimeoutsRequest; +import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationTimeoutsResponse; import org.apache.hadoop.yarn.api.protocolrecords.SignalContainerRequest; import org.apache.hadoop.yarn.api.protocolrecords.SignalContainerResponse; import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationRequest; @@ -139,6 +141,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.ReservationUpdateResponsePBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.UpdateApplicationPriorityRequestPBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.UpdateApplicationPriorityResponsePBImpl; +import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.UpdateApplicationTimeoutsRequestPBImpl; +import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.UpdateApplicationTimeoutsResponsePBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.SignalContainerRequestPBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.SignalContainerResponsePBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.SubmitApplicationRequestPBImpl; @@ -165,7 +169,7 @@ import org.apache.hadoop.yarn.proto.YarnServiceProtos.ReservationSubmissionRequestProto; import org.apache.hadoop.yarn.proto.YarnServiceProtos.ReservationUpdateRequestProto; import org.apache.hadoop.yarn.proto.YarnServiceProtos.UpdateApplicationPriorityRequestProto; -import org.apache.hadoop.yarn.proto.YarnServiceProtos.UpdateApplicationPriorityResponseProto; +import org.apache.hadoop.yarn.proto.YarnServiceProtos.UpdateApplicationTimeoutsRequestProto; import org.apache.hadoop.yarn.proto.YarnServiceProtos.SubmitApplicationRequestProto; import com.google.protobuf.ServiceException; @@ -600,4 +604,19 @@ public SignalContainerResponse signalToContainer( return null; } } + + @Override + public UpdateApplicationTimeoutsResponse updateApplicationTimeouts( + UpdateApplicationTimeoutsRequest request) + throws YarnException, IOException { + UpdateApplicationTimeoutsRequestProto requestProto = + ((UpdateApplicationTimeoutsRequestPBImpl) request).getProto(); + try { + return new UpdateApplicationTimeoutsResponsePBImpl( + proxy.updateApplicationTimeouts(null, requestProto)); + } catch (ServiceException e) { + RPCUtil.unwrapAndThrowException(e); + return null; + } + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/service/ApplicationClientProtocolPBServiceImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/service/ApplicationClientProtocolPBServiceImpl.java index 300ef57..93ce6a3 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/service/ApplicationClientProtocolPBServiceImpl.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/service/ApplicationClientProtocolPBServiceImpl.java @@ -55,6 +55,7 @@ import org.apache.hadoop.yarn.api.protocolrecords.ReservationSubmissionResponse; import org.apache.hadoop.yarn.api.protocolrecords.ReservationUpdateResponse; import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationPriorityResponse; +import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationTimeoutsResponse; import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationResponse; import org.apache.hadoop.yarn.api.protocolrecords.SignalContainerResponse; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.CancelDelegationTokenRequestPBImpl; @@ -111,6 +112,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.SignalContainerResponsePBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.UpdateApplicationPriorityRequestPBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.UpdateApplicationPriorityResponsePBImpl; +import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.UpdateApplicationTimeoutsRequestPBImpl; +import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.UpdateApplicationTimeoutsResponsePBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.SubmitApplicationRequestPBImpl; import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.SubmitApplicationResponsePBImpl; import org.apache.hadoop.yarn.exceptions.YarnException; @@ -162,6 +165,8 @@ import org.apache.hadoop.yarn.proto.YarnServiceProtos.SignalContainerResponseProto; import org.apache.hadoop.yarn.proto.YarnServiceProtos.UpdateApplicationPriorityRequestProto; import org.apache.hadoop.yarn.proto.YarnServiceProtos.UpdateApplicationPriorityResponseProto; +import org.apache.hadoop.yarn.proto.YarnServiceProtos.UpdateApplicationTimeoutsRequestProto; +import org.apache.hadoop.yarn.proto.YarnServiceProtos.UpdateApplicationTimeoutsResponseProto; import org.apache.hadoop.yarn.proto.YarnServiceProtos.SubmitApplicationRequestProto; import org.apache.hadoop.yarn.proto.YarnServiceProtos.SubmitApplicationResponseProto; @@ -609,4 +614,21 @@ public SignalContainerResponseProto signalToContainer( throw new ServiceException(e); } } + + @Override + public UpdateApplicationTimeoutsResponseProto updateApplicationTimeouts( + RpcController controller, UpdateApplicationTimeoutsRequestProto proto) + throws ServiceException { + UpdateApplicationTimeoutsRequestPBImpl request = + new UpdateApplicationTimeoutsRequestPBImpl(proto); + try { + UpdateApplicationTimeoutsResponse response = + real.updateApplicationTimeouts(request); + return ((UpdateApplicationTimeoutsResponsePBImpl) response).getProto(); + } catch (YarnException e) { + throw new ServiceException(e); + } catch (IOException e) { + throw new ServiceException(e); + } + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/UpdateApplicationTimeoutsRequestPBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/UpdateApplicationTimeoutsRequestPBImpl.java new file mode 100644 index 0000000..a4c0c86 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/UpdateApplicationTimeoutsRequestPBImpl.java @@ -0,0 +1,173 @@ +/** + * 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.protocolrecords.impl.pb; + +import org.apache.hadoop.classification.InterfaceAudience.Private; +import org.apache.hadoop.classification.InterfaceStability.Unstable; +import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationTimeoutsRequest; +import org.apache.hadoop.yarn.api.records.ApplicationId; +import org.apache.hadoop.yarn.api.records.ApplicationTimeouts; +import org.apache.hadoop.yarn.api.records.impl.pb.ApplicationIdPBImpl; +import org.apache.hadoop.yarn.api.records.impl.pb.ApplicationTimeoutsPBImpl; +import org.apache.hadoop.yarn.proto.YarnProtos.ApplicationIdProto; +import org.apache.hadoop.yarn.proto.YarnProtos.ApplicationTimeoutsProto; +import org.apache.hadoop.yarn.proto.YarnServiceProtos.UpdateApplicationTimeoutsRequestProto; +import org.apache.hadoop.yarn.proto.YarnServiceProtos.UpdateApplicationTimeoutsRequestProtoOrBuilder; + +import com.google.protobuf.TextFormat; + +@Private +@Unstable +public class UpdateApplicationTimeoutsRequestPBImpl + extends UpdateApplicationTimeoutsRequest { + + UpdateApplicationTimeoutsRequestProto proto = + UpdateApplicationTimeoutsRequestProto.getDefaultInstance(); + UpdateApplicationTimeoutsRequestProto.Builder builder = null; + boolean viaProto = false; + + private ApplicationId applicationId = null; + private ApplicationTimeouts applicationTimeouts = null; + + public UpdateApplicationTimeoutsRequestPBImpl() { + builder = UpdateApplicationTimeoutsRequestProto.newBuilder(); + } + + public UpdateApplicationTimeoutsRequestPBImpl( + UpdateApplicationTimeoutsRequestProto proto) { + this.proto = proto; + viaProto = true; + } + + public UpdateApplicationTimeoutsRequestProto getProto() { + mergeLocalToProto(); + proto = viaProto ? proto : builder.build(); + viaProto = true; + return proto; + } + + private void mergeLocalToProto() { + if (viaProto) + maybeInitBuilder(); + mergeLocalToBuilder(); + proto = builder.build(); + viaProto = true; + } + + private void maybeInitBuilder() { + if (viaProto || builder == null) { + builder = UpdateApplicationTimeoutsRequestProto.newBuilder(proto); + } + viaProto = false; + } + + private void mergeLocalToBuilder() { + if (this.applicationId != null) { + builder.setApplicationId(convertToProtoFormat(this.applicationId)); + } + if (this.applicationTimeouts != null) { + builder.setApplicationTimeouts( + convertToProtoFormat(this.applicationTimeouts)); + } + } + + @Override + public ApplicationId getApplicationId() { + UpdateApplicationTimeoutsRequestProtoOrBuilder p = + viaProto ? proto : builder; + if (this.applicationId != null) { + return applicationId; + } // Else via proto + if (!p.hasApplicationId()) { + return null; + } + applicationId = convertFromProtoFormat(p.getApplicationId()); + return applicationId; + } + + @Override + public void setApplicationId(ApplicationId applicationId) { + maybeInitBuilder(); + if (applicationId == null) { + builder.clearApplicationId(); + } + this.applicationId = applicationId; + } + + private ApplicationIdPBImpl convertFromProtoFormat(ApplicationIdProto p) { + return new ApplicationIdPBImpl(p); + } + + private ApplicationIdProto convertToProtoFormat(ApplicationId t) { + return ((ApplicationIdPBImpl) t).getProto(); + } + + @Override + public ApplicationTimeouts getApplicationTimeouts() { + UpdateApplicationTimeoutsRequestProtoOrBuilder p = + viaProto ? proto : builder; + + if (this.applicationTimeouts != null) { + return this.applicationTimeouts; + } // Else via proto + if (!p.hasApplicationTimeouts()) { + return null; + } + applicationTimeouts = convertFromProtoFormat(p.getApplicationTimeouts()); + return applicationTimeouts; + } + + @Override + public void setApplicationTimeouts(ApplicationTimeouts appTimeouts) { + maybeInitBuilder(); + if (applicationTimeouts == null) { + builder.clearApplicationTimeouts(); + } + this.applicationTimeouts = appTimeouts; + } + + private ApplicationTimeoutsPBImpl convertFromProtoFormat( + ApplicationTimeoutsProto p) { + return new ApplicationTimeoutsPBImpl(p); + } + + private ApplicationTimeoutsProto convertToProtoFormat(ApplicationTimeouts t) { + return ((ApplicationTimeoutsPBImpl) t).getProto(); + } + + @Override + public int hashCode() { + return getProto().hashCode(); + } + + @Override + public boolean equals(Object other) { + if (other == null) + return false; + if (other.getClass().isAssignableFrom(this.getClass())) { + return this.getProto().equals(this.getClass().cast(other).getProto()); + } + return false; + } + + @Override + public String toString() { + return TextFormat.shortDebugString(getProto()); + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/UpdateApplicationTimeoutsResponsePBImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/UpdateApplicationTimeoutsResponsePBImpl.java new file mode 100644 index 0000000..9c2b157 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/UpdateApplicationTimeoutsResponsePBImpl.java @@ -0,0 +1,73 @@ +/** + * 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.protocolrecords.impl.pb; + + +import org.apache.hadoop.classification.InterfaceAudience.Private; +import org.apache.hadoop.classification.InterfaceStability.Unstable; +import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationTimeoutsResponse; +import org.apache.hadoop.yarn.proto.YarnServiceProtos.UpdateApplicationTimeoutsResponseProto; + +import com.google.protobuf.TextFormat; + +@Private +@Unstable +public class UpdateApplicationTimeoutsResponsePBImpl + extends UpdateApplicationTimeoutsResponse { + UpdateApplicationTimeoutsResponseProto proto = + UpdateApplicationTimeoutsResponseProto.getDefaultInstance(); + UpdateApplicationTimeoutsResponseProto.Builder builder = null; + boolean viaProto = false; + + public UpdateApplicationTimeoutsResponsePBImpl() { + builder = UpdateApplicationTimeoutsResponseProto.newBuilder(); + } + + public UpdateApplicationTimeoutsResponsePBImpl( + UpdateApplicationTimeoutsResponseProto proto) { + this.proto = proto; + viaProto = true; + } + + public UpdateApplicationTimeoutsResponseProto getProto() { + proto = viaProto ? proto : builder.build(); + viaProto = true; + return proto; + } + + @Override + public int hashCode() { + return getProto().hashCode(); + } + + @Override + public boolean equals(Object other) { + if (other == null) + return false; + if (other.getClass().isAssignableFrom(this.getClass())) { + return this.getProto().equals(this.getClass().cast(other).getProto()); + } + return false; + } + + @Override + public String toString() { + return TextFormat.shortDebugString(getProto()); + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/MockResourceManagerFacade.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/MockResourceManagerFacade.java index 2ccf827..b530fb4 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/MockResourceManagerFacade.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/amrmproxy/MockResourceManagerFacade.java @@ -91,6 +91,8 @@ import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationResponse; import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationPriorityRequest; import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationPriorityResponse; +import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationTimeoutsRequest; +import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationTimeoutsResponse; import org.apache.hadoop.yarn.api.ApplicationClientProtocol; import org.apache.hadoop.yarn.api.ApplicationMasterProtocol; import org.apache.hadoop.yarn.api.records.AMCommand; @@ -496,4 +498,11 @@ public FailApplicationAttemptResponse failApplicationAttempt( FailApplicationAttemptRequest request) throws YarnException, IOException { throw new NotImplementedException(); } + + @Override + public UpdateApplicationTimeoutsResponse updateApplicationTimeouts( + UpdateApplicationTimeoutsRequest request) + throws YarnException, IOException { + throw new NotImplementedException(); + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java index e9bd230..1068761 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java @@ -108,12 +108,15 @@ import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationResponse; import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationPriorityRequest; import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationPriorityResponse; +import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationTimeoutsRequest; +import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationTimeoutsResponse; import org.apache.hadoop.yarn.api.records.ApplicationAccessType; import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; import org.apache.hadoop.yarn.api.records.ApplicationAttemptReport; import org.apache.hadoop.yarn.api.records.ApplicationId; import org.apache.hadoop.yarn.api.records.ApplicationReport; import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext; +import org.apache.hadoop.yarn.api.records.ApplicationTimeouts; import org.apache.hadoop.yarn.api.records.ContainerId; import org.apache.hadoop.yarn.api.records.ContainerReport; import org.apache.hadoop.yarn.api.records.NodeReport; @@ -140,6 +143,7 @@ import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier; import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstants; import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; +import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData; import org.apache.hadoop.yarn.server.resourcemanager.reservation.Plan; import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationAllocation; import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationInputValidator; @@ -1589,37 +1593,11 @@ public UpdateApplicationPriorityResponse updateApplicationPriority( ApplicationId applicationId = request.getApplicationId(); Priority newAppPriority = request.getApplicationPriority(); - UserGroupInformation callerUGI; - try { - callerUGI = UserGroupInformation.getCurrentUser(); - } catch (IOException ie) { - LOG.info("Error getting UGI ", ie); - RMAuditLogger.logFailure("UNKNOWN", AuditConstants.UPDATE_APP_PRIORITY, - "UNKNOWN", "ClientRMService", "Error getting UGI", applicationId); - throw RPCUtil.getRemoteException(ie); - } - RMApp application = this.rmContext.getRMApps().get(applicationId); - if (application == null) { - RMAuditLogger.logFailure(callerUGI.getUserName(), - AuditConstants.UPDATE_APP_PRIORITY, "UNKNOWN", "ClientRMService", - "Trying to update priority of an absent application", applicationId); - throw new ApplicationNotFoundException( - "Trying to update priority of an absent application " - + applicationId); - } - - if (!checkAccess(callerUGI, application.getUser(), - ApplicationAccessType.MODIFY_APP, application)) { - RMAuditLogger.logFailure(callerUGI.getShortUserName(), - AuditConstants.UPDATE_APP_PRIORITY, - "User doesn't have permissions to " - + ApplicationAccessType.MODIFY_APP.toString(), "ClientRMService", - AuditConstants.UNAUTHORIZED_USER, applicationId); - throw RPCUtil.getRemoteException(new AccessControlException("User " - + callerUGI.getShortUserName() + " cannot perform operation " - + ApplicationAccessType.MODIFY_APP.name() + " on " + applicationId)); - } + UserGroupInformation callerUGI = + getCallerUgi(applicationId, AuditConstants.UPDATE_APP_PRIORITY); + RMApp application = verifyUserAccessForRMApp(applicationId, callerUGI, + AuditConstants.UPDATE_APP_PRIORITY); UpdateApplicationPriorityResponse response = recordFactory .newRecordInstance(UpdateApplicationPriorityResponse.class); @@ -1724,4 +1702,104 @@ public SignalContainerResponse signalToContainer( .newRecordInstance(SignalContainerResponse.class); } + @Override + public UpdateApplicationTimeoutsResponse updateApplicationTimeouts( + UpdateApplicationTimeoutsRequest request) + throws YarnException, IOException { + + ApplicationId applicationId = request.getApplicationId(); + ApplicationTimeouts newAppTimeouts = request.getApplicationTimeouts(); + + UserGroupInformation callerUGI = + getCallerUgi(applicationId, AuditConstants.UPDATE_APP_TIMEOUTS); + RMApp application = verifyUserAccessForRMApp(applicationId, callerUGI, + AuditConstants.UPDATE_APP_TIMEOUTS); + + UpdateApplicationTimeoutsResponse response = recordFactory + .newRecordInstance(UpdateApplicationTimeoutsResponse.class); + + if (COMPLETED_APP_STATES.contains(application.getState())) { + RMAuditLogger.logSuccess(callerUGI.getShortUserName(), + AuditConstants.UPDATE_APP_TIMEOUTS, "ClientRMService", applicationId); + return response; + } + + long newLifetime = newAppTimeouts.getLifetime(); + long oldLifetime = 0; + + if (newLifetime < 0) { + RMAuditLogger.logFailure("UNKNOWN", AuditConstants.UPDATE_APP_TIMEOUTS, + "UNKNOWN", "ClientRMService", "Invalid timeout value", applicationId); + throw RPCUtil.getRemoteException( + "Invalid timeout value " + newLifetime + " has been provided"); + } + + // update in monitor first to reflect immediately + rmContext.getRMAppLifetimeMonitor().updateApplicationLifetime(applicationId, + newLifetime * 1000); + + ApplicationTimeouts appTimeouts = + application.getApplicationSubmissionContext().getApplicationTimeouts(); + if (appTimeouts != null) { + oldLifetime = + appTimeouts.getLifetime() > 0 ? appTimeouts.getLifetime() : 0; + } + appTimeouts.setLifetime(oldLifetime + newLifetime); + + // update submission context + application.getApplicationSubmissionContext() + .setApplicationTimeouts(appTimeouts); + + // Update to state store + ApplicationStateData appState = ApplicationStateData.newInstance( + application.getSubmitTime(), application.getStartTime(), + application.getApplicationSubmissionContext(), application.getUser(), + application.getCallerContext()); + rmContext.getStateStore().updateApplicationStateSynchronously(appState, + false); + + RMAuditLogger.logSuccess(callerUGI.getShortUserName(), + AuditConstants.UPDATE_APP_TIMEOUTS, "ClientRMService", applicationId); + return response; + } + + private UserGroupInformation getCallerUgi(ApplicationId applicationId, + String operation) throws YarnException { + UserGroupInformation callerUGI; + try { + callerUGI = UserGroupInformation.getCurrentUser(); + } catch (IOException ie) { + LOG.info("Error getting UGI ", ie); + RMAuditLogger.logFailure("UNKNOWN", operation, "UNKNOWN", + "ClientRMService", "Error getting UGI", applicationId); + throw RPCUtil.getRemoteException(ie); + } + return callerUGI; + } + + private RMApp verifyUserAccessForRMApp(ApplicationId applicationId, + UserGroupInformation callerUGI, String operation) throws YarnException { + RMApp application = this.rmContext.getRMApps().get(applicationId); + if (application == null) { + RMAuditLogger.logFailure(callerUGI.getUserName(), operation, "UNKNOWN", + "ClientRMService", + "Trying to " + operation + " of an absent application", + applicationId); + throw new ApplicationNotFoundException("Trying to " + operation + + " of an absent application " + applicationId); + } + + if (!checkAccess(callerUGI, application.getUser(), + ApplicationAccessType.MODIFY_APP, application)) { + RMAuditLogger.logFailure(callerUGI.getShortUserName(), operation, + "User doesn't have permissions to " + + ApplicationAccessType.MODIFY_APP.toString(), + "ClientRMService", AuditConstants.UNAUTHORIZED_USER, applicationId); + throw RPCUtil.getRemoteException(new AccessControlException("User " + + callerUGI.getShortUserName() + " cannot perform operation " + + ApplicationAccessType.MODIFY_APP.name() + " on " + applicationId)); + } + return application; + } + } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAuditLogger.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAuditLogger.java index 84c0390..bba2a90 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAuditLogger.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMAuditLogger.java @@ -60,7 +60,9 @@ public static final String ALLOC_CONTAINER = "AM Allocated Container"; public static final String RELEASE_CONTAINER = "AM Released Container"; public static final String UPDATE_APP_PRIORITY = - "Update Application Priority Request"; + "Update Application Priority"; + public static final String UPDATE_APP_TIMEOUTS = + "Update Application Timeouts"; public static final String CHANGE_CONTAINER_RESOURCE = "AM Changed Container Resource"; public static final String SIGNAL_CONTAINER = "Signal Container Request"; diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppLifetimeMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppLifetimeMonitor.java index cb5b9c2..e8a2aad 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppLifetimeMonitor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/RMAppLifetimeMonitor.java @@ -19,8 +19,8 @@ package org.apache.hadoop.yarn.server.resourcemanager.rmapp; import java.util.EnumSet; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; +import java.util.HashMap; +import java.util.Map; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -41,8 +41,8 @@ private static final Log LOG = LogFactory.getLog(RMAppLifetimeMonitor.class); private RMContext rmContext; - private ConcurrentMap monitoredApps = - new ConcurrentHashMap(); + private Map monitoredApps = + new HashMap(); private static final EnumSet COMPLETED_APP_STATES = EnumSet.of(RMAppState.FINISHED, RMAppState.FINISHING, RMAppState.FAILED, @@ -70,7 +70,7 @@ protected void serviceInit(Configuration conf) throws Exception { @SuppressWarnings("unchecked") @Override - protected void expire(ApplicationId appId) { + protected synchronized void expire(ApplicationId appId) { Long remove = monitoredApps.remove(appId); RMApp app = rmContext.getRMApps().get(appId); if (app == null) { @@ -78,36 +78,48 @@ protected void expire(ApplicationId appId) { } // Don't trigger a KILL event if application is in completed states if (!COMPLETED_APP_STATES.contains(app.getState())) { - String diagnosis = + String diagnostics = "Application killed due to exceeding its lifetime period " + remove + " milliseconds"; rmContext.getDispatcher().getEventHandler() - .handle(new RMAppEvent(appId, RMAppEventType.KILL, diagnosis)); + .handle(new RMAppEvent(appId, RMAppEventType.KILL, diagnostics)); } else { LOG.info("Application " + appId + " is about to complete. So not killing the application."); } } - public void registerApp(ApplicationId appId, long monitorStartTime, - long applicationLifetime) { + public synchronized void registerApp(ApplicationId appId, + long monitorStartTime, long applicationLifetime) { register(appId, monitorStartTime); - monitoredApps.putIfAbsent(appId, applicationLifetime); + monitoredApps.put(appId, applicationLifetime); } @Override - protected long getExpireInterval(ApplicationId appId) { + protected synchronized long getExpireInterval(ApplicationId appId) { return monitoredApps.get(appId); } - public void unregisterApp(ApplicationId appId) { + public synchronized void unregisterApp(ApplicationId appId) { unregister(appId); monitoredApps.remove(appId); } - public void updateApplicationLifetime(ApplicationId appId, - long newAppLifetime) { - // only update for the registered objects - monitoredApps.replace(appId, newAppLifetime); + public synchronized void updateApplicationLifetime(ApplicationId appId, + long newLifetime) { + if (!monitoredApps.containsKey(appId)) { + // If application is not monitored earlier then start monitoring from now + register(appId); + monitoredApps.put(appId, newLifetime); + LOG.info("Application " + appId + + " is registered with Application lifetime monitor. " + + "The lifetime configured is " + newLifetime / 1000 + " seconds"); + } else { + Long oldLifetime = monitoredApps.get(appId); + monitoredApps.put(appId, oldLifetime + newLifetime); + LOG.info("Application " + appId + " is updated lifetime value. " + + "New lifetime value is " + (oldLifetime + newLifetime) / 1000 + + " seconds"); + } } } \ No newline at end of file diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/TestApplicationLifetimeMonitor.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/TestApplicationLifetimeMonitor.java index 18fd151..7bc42d8 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/TestApplicationLifetimeMonitor.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/TestApplicationLifetimeMonitor.java @@ -25,6 +25,8 @@ import java.util.Set; import org.apache.hadoop.security.UserGroupInformation; +import org.apache.hadoop.yarn.api.protocolrecords.UpdateApplicationTimeoutsRequest; +import org.apache.hadoop.yarn.api.records.ApplicationTimeouts; import org.apache.hadoop.yarn.api.records.ContainerId; import org.apache.hadoop.yarn.api.records.ContainerState; import org.apache.hadoop.yarn.api.records.Priority; @@ -74,14 +76,40 @@ public void testApplicationLifetimeMonitor() throws Exception { Priority appPriority = Priority.newInstance(0); MockNM nm1 = rm.registerNode("127.0.0.1:1234", 16 * 1024); RMApp app1 = rm.submitApp(1024, appPriority, 10); + RMApp app2 = rm.submitApp(1024, appPriority, 20); nm1.nodeHeartbeat(true); // Send launch Event MockAM am1 = rm.sendAMLaunched(app1.getCurrentAppAttempt().getAppAttemptId()); am1.registerAppAttempt(); rm.waitForState(app1.getApplicationId(), RMAppState.KILLED); - Assert.assertTrue("Applicaiton killed before lifetime value", + Assert.assertTrue("Application killed before lifetime value", (System.currentTimeMillis() - app1.getSubmitTime()) > 10000); + + long newLifetime = 10; + ApplicationTimeouts timeouts = + ApplicationTimeouts.newInstance(newLifetime); + UpdateApplicationTimeoutsRequest request = + UpdateApplicationTimeoutsRequest.newInstance(app2.getApplicationId(), + timeouts); + + long beforeUpdate = app2.getApplicationSubmissionContext() + .getApplicationTimeouts().getLifetime(); + + // update second app1 lifetime + rm.getRMContext().getClientRMService().updateApplicationTimeouts(request); + + long afterUpdate = app2.getApplicationSubmissionContext() + .getApplicationTimeouts().getLifetime(); + + Assert.assertEquals("Application lifetime value not updated", + beforeUpdate + newLifetime, afterUpdate); + + rm.waitForState(app2.getApplicationId(), RMAppState.KILLED); + // verify for app killed with updated lifetime + Assert.assertTrue("Application killed before lifetime value", + (System.currentTimeMillis() - app2.getSubmitTime()) > afterUpdate + * 1000); } finally { stopRM(rm); } @@ -142,7 +170,7 @@ public void testApplicationLifetimeOnRMRestart() throws Exception { // wait for app life time and application to be in killed state. rm2.waitForState(recoveredApp1.getApplicationId(), RMAppState.KILLED); - Assert.assertTrue("Applicaiton killed before lifetime value", + Assert.assertTrue("Application killed before lifetime value", (System.currentTimeMillis() - recoveredApp1.getSubmitTime()) > appLifetime); } -- 2.7.4 (Apple Git-66)