From 70c03a3d1d06cf4e434a9cbf9b76bd5492dff543 Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Fri, 12 Aug 2022 08:49:30 -0700 Subject: [PATCH 01/13] YARN-11250. Capture the Performance Metrics of ZookeeperFederationStateStore. --- .../ZKFederationStateStoreOpDurations.java | 155 ++++++++++++++++++ .../impl/ZookeeperFederationStateStore.java | 54 ++++-- .../TestZookeeperFederationStateStore.java | 41 +++++ 3 files changed, 238 insertions(+), 12 deletions(-) create mode 100644 hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java new file mode 100644 index 0000000000000..4021b04677c47 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java @@ -0,0 +1,155 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with this + * work for additional information regarding copyright ownership. The ASF + * licenses this file to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations under + * the License. + */ +package org.apache.hadoop.yarn.server.federation.store.impl; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; +import org.apache.hadoop.metrics2.MetricsCollector; +import org.apache.hadoop.metrics2.MetricsInfo; +import org.apache.hadoop.metrics2.MetricsSource; +import org.apache.hadoop.metrics2.MetricsSystem; +import org.apache.hadoop.metrics2.annotation.Metric; +import org.apache.hadoop.metrics2.annotation.Metrics; +import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem; +import org.apache.hadoop.metrics2.lib.MetricsRegistry; +import org.apache.hadoop.metrics2.lib.MutableRate; + +import static org.apache.hadoop.metrics2.lib.Interns.info; + +@InterfaceAudience.Private +@InterfaceStability.Unstable +@Metrics(context="ZKFederationStateStore-op-durations") +public final class ZKFederationStateStoreOpDurations implements MetricsSource { + + @Metric("Duration for a add application homeSubcluster call") + private MutableRate addAppHomeSubClusterCall; + + @Metric("Duration for a update application homeSubcluster call") + private MutableRate updateAppHomeSubClusterCall; + + @Metric("Duration for a get application homeSubcluster call") + private MutableRate getAppHomeSubClusterCall; + + @Metric("Duration for a get applications homeSubcluster call") + private MutableRate getAppsHomeSubClusterCall; + + @Metric("Duration for a delete applications homeSubcluster call") + private MutableRate deleteAppHomeSubClusterCall; + + @Metric("Duration for a register subCluster call") + private MutableRate registerSubClusterCall; + + @Metric("Duration for a deregister subCluster call") + private MutableRate deregisterSubCluster; + + @Metric("Duration for a subCluster Heartbeat call") + private MutableRate subClusterHeartbeat; + + @Metric("Duration for a get SubCluster call") + private MutableRate getSubCluster; + + @Metric("Duration for a get SubClusters call") + private MutableRate getSubClusters; + + @Metric("Duration for a get PolicyConfiguration call") + private MutableRate getPolicyConfiguration; + + @Metric("Duration for a set PolicyConfiguration call") + private MutableRate setPolicyConfiguration; + + @Metric("Duration for a get PolicyConfigurations call") + private MutableRate getPoliciesConfigurations; + + protected static final MetricsInfo RECORD_INFO = + info("ZKFederationStateStoreOpDurations", "Durations of ZKFederationStateStore calls"); + + private final MetricsRegistry registry; + + private static final ZKFederationStateStoreOpDurations INSTANCE = + new ZKFederationStateStoreOpDurations(); + + public static ZKFederationStateStoreOpDurations getInstance() { + return INSTANCE; + } + + private ZKFederationStateStoreOpDurations() { + registry = new MetricsRegistry(RECORD_INFO); + registry.tag(RECORD_INFO, "ZKFederationStateStoreOpDurations"); + + MetricsSystem ms = DefaultMetricsSystem.instance(); + if (ms != null) { + ms.register(RECORD_INFO.name(), RECORD_INFO.description(), this); + } + } + + @Override + public synchronized void getMetrics(MetricsCollector collector, boolean all) { + registry.snapshot(collector.addRecord(registry.info()), all); + } + + public void addAppHomeSubClusterCallDuration(long value) { + addAppHomeSubClusterCall.add(value); + } + + public void addUpdateAppHomeSubClusterCallDuration(long value) { + updateAppHomeSubClusterCall.add(value); + } + + public void addGetAppHomeSubClusterCallDuration(long value) { + getAppHomeSubClusterCall.add(value); + } + + public void addGetAppsHomeSubClusterCallDuration(long value) { + getAppsHomeSubClusterCall.add(value); + } + + public void addDeleteAppHomeSubClusterCallDuration(long value) { + deleteAppHomeSubClusterCall.add(value); + } + + public void addRegisterSubClusterCallDuration(long value) { + registerSubClusterCall.add(value); + } + + public void addDeregisterSubClusterCallDuration(long value) { + deregisterSubCluster.add(value); + } + + public void addSubClusterHeartbeatCallDuration(long value) { + subClusterHeartbeat.add(value); + } + + public void addGetSubClusterCallDuration(long value) { + getSubCluster.add(value); + } + + public void addGetSubClustersCallDuration(long value) { + getSubClusters.add(value); + } + + public void addGetPolicyConfigurationDuration(long value) { + getPolicyConfiguration.add(value); + } + + public void addSetPolicyConfigurationDuration(long value) { + setPolicyConfiguration.add(value); + } + + public void addGetPoliciesConfigurationsDuration(long value) { + getPoliciesConfigurations.add(value); + } +} diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java index c9b5849ad68d2..6d58cf3f6f7bb 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java @@ -25,6 +25,7 @@ import java.util.List; import java.util.TimeZone; +import org.apache.hadoop.classification.VisibleForTesting; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.util.curator.ZKCuratorManager; import org.apache.hadoop.yarn.api.records.ApplicationId; @@ -73,6 +74,8 @@ import org.apache.hadoop.yarn.server.federation.store.utils.FederationPolicyStoreInputValidator; import org.apache.hadoop.yarn.server.federation.store.utils.FederationStateStoreUtils; import org.apache.hadoop.yarn.server.records.Version; +import org.apache.hadoop.yarn.util.Clock; +import org.apache.hadoop.yarn.util.SystemClock; import org.apache.zookeeper.data.ACL; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -113,6 +116,12 @@ public class ZookeeperFederationStateStore implements FederationStateStore { private String membershipZNode; private String policiesZNode; + private volatile Clock clock = SystemClock.getInstance(); + + @VisibleForTesting + private ZKFederationStateStoreOpDurations opDurations = + ZKFederationStateStoreOpDurations.getInstance(); + @Override public void init(Configuration conf) throws YarnException { LOG.info("Initializing ZooKeeper connection"); @@ -142,7 +151,6 @@ public void init(Configuration conf) throws YarnException { String errMsg = "Cannot create base directories: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - } @Override @@ -156,6 +164,7 @@ public void close() throws Exception { public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( AddApplicationHomeSubClusterRequest request) throws YarnException { + long start = clock.getTime(); FederationApplicationHomeSubClusterStoreInputValidator.validate(request); ApplicationHomeSubCluster app = request.getApplicationHomeSubCluster(); ApplicationId appId = app.getApplicationId(); @@ -177,8 +186,8 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - return AddApplicationHomeSubClusterResponse - .newInstance(homeSubCluster); + opDurations.addAppHomeSubClusterCallDuration(clock.getTime() - start); + return AddApplicationHomeSubClusterResponse.newInstance(homeSubCluster); } @Override @@ -187,6 +196,7 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( UpdateApplicationHomeSubClusterRequest request) throws YarnException { + long start = clock.getTime(); FederationApplicationHomeSubClusterStoreInputValidator.validate(request); ApplicationHomeSubCluster app = request.getApplicationHomeSubCluster(); ApplicationId appId = app.getApplicationId(); @@ -198,6 +208,8 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( SubClusterId newSubClusterId = request.getApplicationHomeSubCluster().getHomeSubCluster(); putApp(appId, newSubClusterId, true); + + opDurations.addUpdateAppHomeSubClusterCallDuration(clock.getTime() - start); return UpdateApplicationHomeSubClusterResponse.newInstance(); } @@ -205,6 +217,7 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( public GetApplicationHomeSubClusterResponse getApplicationHomeSubCluster( GetApplicationHomeSubClusterRequest request) throws YarnException { + long start = clock.getTime(); FederationApplicationHomeSubClusterStoreInputValidator.validate(request); ApplicationId appId = request.getApplicationId(); SubClusterId homeSubCluster = getApp(appId); @@ -212,6 +225,7 @@ public GetApplicationHomeSubClusterResponse getApplicationHomeSubCluster( String errMsg = "Application " + appId + " does not exist"; FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } + opDurations.addGetAppHomeSubClusterCallDuration(clock.getTime() - start); return GetApplicationHomeSubClusterResponse.newInstance( ApplicationHomeSubCluster.newInstance(appId, homeSubCluster)); } @@ -219,6 +233,7 @@ public GetApplicationHomeSubClusterResponse getApplicationHomeSubCluster( @Override public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( GetApplicationsHomeSubClusterRequest request) throws YarnException { + long start = clock.getTime(); List result = new ArrayList<>(); try { @@ -233,7 +248,7 @@ public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( String errMsg = "Cannot get apps: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - + opDurations.addGetAppsHomeSubClusterCallDuration(clock.getTime() - start); return GetApplicationsHomeSubClusterResponse.newInstance(result); } @@ -242,7 +257,7 @@ public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( deleteApplicationHomeSubCluster( DeleteApplicationHomeSubClusterRequest request) throws YarnException { - + long start = clock.getTime(); FederationApplicationHomeSubClusterStoreInputValidator.validate(request); ApplicationId appId = request.getApplicationId(); String appZNode = getNodePath(appsZNode, appId.toString()); @@ -265,13 +280,14 @@ public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( String errMsg = "Cannot delete app: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - + opDurations.addDeleteAppHomeSubClusterCallDuration(clock.getTime() - start); return DeleteApplicationHomeSubClusterResponse.newInstance(); } @Override public SubClusterRegisterResponse registerSubCluster( SubClusterRegisterRequest request) throws YarnException { + long start = clock.getTime(); FederationMembershipStateStoreInputValidator.validate(request); SubClusterInfo subClusterInfo = request.getSubClusterInfo(); SubClusterId subclusterId = subClusterInfo.getSubClusterId(); @@ -286,12 +302,14 @@ public SubClusterRegisterResponse registerSubCluster( String errMsg = "Cannot register subcluster: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } + opDurations.addRegisterSubClusterCallDuration(clock.getTime() - start); return SubClusterRegisterResponse.newInstance(); } @Override public SubClusterDeregisterResponse deregisterSubCluster( SubClusterDeregisterRequest request) throws YarnException { + long start = clock.getTime(); FederationMembershipStateStoreInputValidator.validate(request); SubClusterId subClusterId = request.getSubClusterId(); SubClusterState state = request.getState(); @@ -305,14 +323,14 @@ public SubClusterDeregisterResponse deregisterSubCluster( subClusterInfo.setState(state); putSubclusterInfo(subClusterId, subClusterInfo, true); } - + opDurations.addDeregisterSubClusterCallDuration(clock.getTime() - start); return SubClusterDeregisterResponse.newInstance(); } @Override public SubClusterHeartbeatResponse subClusterHeartbeat( SubClusterHeartbeatRequest request) throws YarnException { - + long start = clock.getTime(); FederationMembershipStateStoreInputValidator.validate(request); SubClusterId subClusterId = request.getSubClusterId(); @@ -329,14 +347,14 @@ public SubClusterHeartbeatResponse subClusterHeartbeat( subClusterInfo.setCapability(request.getCapability()); putSubclusterInfo(subClusterId, subClusterInfo, true); - + opDurations.addSubClusterHeartbeatCallDuration(clock.getTime() - start); return SubClusterHeartbeatResponse.newInstance(); } @Override public GetSubClusterInfoResponse getSubCluster( GetSubClusterInfoRequest request) throws YarnException { - + long start = clock.getTime(); FederationMembershipStateStoreInputValidator.validate(request); SubClusterId subClusterId = request.getSubClusterId(); SubClusterInfo subClusterInfo = null; @@ -350,12 +368,14 @@ public GetSubClusterInfoResponse getSubCluster( String errMsg = "Cannot get subcluster: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } + opDurations.addGetSubClusterCallDuration(clock.getTime() - start); return GetSubClusterInfoResponse.newInstance(subClusterInfo); } @Override public GetSubClustersInfoResponse getSubClusters( GetSubClustersInfoRequest request) throws YarnException { + long start = clock.getTime(); List result = new ArrayList<>(); try { @@ -371,6 +391,7 @@ public GetSubClustersInfoResponse getSubClusters( String errMsg = "Cannot get subclusters: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } + opDurations.addGetSubClustersCallDuration(clock.getTime() - start); return GetSubClustersInfoResponse.newInstance(result); } @@ -378,7 +399,7 @@ public GetSubClustersInfoResponse getSubClusters( @Override public GetSubClusterPolicyConfigurationResponse getPolicyConfiguration( GetSubClusterPolicyConfigurationRequest request) throws YarnException { - + long start = clock.getTime(); FederationPolicyStoreInputValidator.validate(request); String queue = request.getQueue(); SubClusterPolicyConfiguration policy = null; @@ -393,6 +414,7 @@ public GetSubClusterPolicyConfigurationResponse getPolicyConfiguration( LOG.warn("Policy for queue: {} does not exist.", queue); return null; } + opDurations.addGetPolicyConfigurationDuration(clock.getTime() - start); return GetSubClusterPolicyConfigurationResponse .newInstance(policy); } @@ -400,7 +422,7 @@ public GetSubClusterPolicyConfigurationResponse getPolicyConfiguration( @Override public SetSubClusterPolicyConfigurationResponse setPolicyConfiguration( SetSubClusterPolicyConfigurationRequest request) throws YarnException { - + long start = clock.getTime(); FederationPolicyStoreInputValidator.validate(request); SubClusterPolicyConfiguration policy = request.getPolicyConfiguration(); @@ -411,12 +433,14 @@ public SetSubClusterPolicyConfigurationResponse setPolicyConfiguration( String errMsg = "Cannot set policy: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } + opDurations.addSetPolicyConfigurationDuration(clock.getTime() - start); return SetSubClusterPolicyConfigurationResponse.newInstance(); } @Override public GetSubClusterPoliciesConfigurationsResponse getPoliciesConfigurations( GetSubClusterPoliciesConfigurationsRequest request) throws YarnException { + long start = clock.getTime(); List result = new ArrayList<>(); try { @@ -432,6 +456,7 @@ public GetSubClusterPoliciesConfigurationsResponse getPoliciesConfigurations( String errMsg = "Cannot get policies: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } + opDurations.addGetPoliciesConfigurationsDuration(clock.getTime() - start); return GetSubClusterPoliciesConfigurationsResponse.newInstance(result); } @@ -637,4 +662,9 @@ private static long getCurrentTime() { Calendar cal = Calendar.getInstance(TimeZone.getTimeZone("UTC")); return cal.getTimeInMillis(); } + + @VisibleForTesting + public ZKFederationStateStoreOpDurations getOpDurations() { + return opDurations; + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java index fe28641eb2560..51fd9c89437a2 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java @@ -25,14 +25,20 @@ import org.apache.curator.test.TestingServer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.CommonConfigurationKeys; +import org.apache.hadoop.metrics2.MetricsRecord; +import org.apache.hadoop.metrics2.impl.MetricsCollectorImpl; +import org.apache.hadoop.metrics2.impl.MetricsRecords; import org.apache.hadoop.yarn.conf.YarnConfiguration; import org.apache.hadoop.yarn.exceptions.YarnException; import org.apache.hadoop.yarn.server.federation.store.FederationStateStore; import org.junit.After; import org.junit.Before; +import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static org.junit.Assert.assertEquals; + /** * Unit tests for ZookeeperFederationStateStore. */ @@ -84,4 +90,39 @@ protected FederationStateStore createStateStore() { super.setConf(getConf()); return new ZookeeperFederationStateStore(); } + + @Test + public void testMetricsInited() throws Exception { + ZookeeperFederationStateStore zkStateStore = (ZookeeperFederationStateStore) createStateStore(); + ZKFederationStateStoreOpDurations zkStateStoreOpDurations = zkStateStore.getOpDurations(); + MetricsCollectorImpl collector = new MetricsCollectorImpl(); + + long anyDuration = 10; + zkStateStoreOpDurations.addAppHomeSubClusterCallDuration(anyDuration); + zkStateStoreOpDurations.addUpdateAppHomeSubClusterCallDuration(anyDuration); + zkStateStoreOpDurations.addGetAppHomeSubClusterCallDuration(anyDuration); + zkStateStoreOpDurations.addGetAppsHomeSubClusterCallDuration(anyDuration); + zkStateStoreOpDurations.addDeleteAppHomeSubClusterCallDuration(anyDuration); + + zkStateStoreOpDurations.getMetrics(collector, true); + assertEquals("Incorrect number of perf metrics", 1, collector.getRecords().size()); + + MetricsRecord record = collector.getRecords().get(0); + MetricsRecords.assertTag(record, ZKFederationStateStoreOpDurations.RECORD_INFO.name(), + "ZKFederationStateStoreOpDurations"); + + double expectAvgTime = anyDuration; + MetricsRecords.assertMetric(record, "AddAppHomeSubClusterCallAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "UpdateAppHomeSubClusterCallAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "GetAppHomeSubClusterCallAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "GetAppsHomeSubClusterCallAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "DeleteAppHomeSubClusterCallAvgTime", expectAvgTime); + + long expectOps = 1; + MetricsRecords.assertMetric(record, "AddAppHomeSubClusterCallNumOps", expectOps); + MetricsRecords.assertMetric(record, "UpdateAppHomeSubClusterCallNumOps", expectOps); + MetricsRecords.assertMetric(record, "GetAppHomeSubClusterCallNumOps", expectOps); + MetricsRecords.assertMetric(record, "GetAppsHomeSubClusterCallNumOps", expectOps); + MetricsRecords.assertMetric(record, "DeleteAppHomeSubClusterCallNumOps", expectOps); + } } \ No newline at end of file From 4e98d5e457552d47cd5559c4afa42ecd4a7ed599 Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Fri, 12 Aug 2022 17:32:44 -0700 Subject: [PATCH 02/13] YARN-11250. Capture the Performance Metrics of ZookeeperFederationStateStore. --- .../ZKFederationStateStoreOpDurations.java | 28 +++++++++---------- .../TestZookeeperFederationStateStore.java | 24 ++++++++++++++++ 2 files changed, 38 insertions(+), 14 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java index 4021b04677c47..9b065de7100fb 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java @@ -54,25 +54,25 @@ public final class ZKFederationStateStoreOpDurations implements MetricsSource { private MutableRate registerSubClusterCall; @Metric("Duration for a deregister subCluster call") - private MutableRate deregisterSubCluster; + private MutableRate deregisterSubClusterCall; @Metric("Duration for a subCluster Heartbeat call") - private MutableRate subClusterHeartbeat; + private MutableRate subClusterHeartbeatCall; @Metric("Duration for a get SubCluster call") - private MutableRate getSubCluster; + private MutableRate getSubClusterCall; @Metric("Duration for a get SubClusters call") - private MutableRate getSubClusters; + private MutableRate getSubClustersCall; @Metric("Duration for a get PolicyConfiguration call") - private MutableRate getPolicyConfiguration; + private MutableRate getPolicyConfigurationCall; @Metric("Duration for a set PolicyConfiguration call") - private MutableRate setPolicyConfiguration; + private MutableRate setPolicyConfigurationCall; @Metric("Duration for a get PolicyConfigurations call") - private MutableRate getPoliciesConfigurations; + private MutableRate getPoliciesConfigurationsCall; protected static final MetricsInfo RECORD_INFO = info("ZKFederationStateStoreOpDurations", "Durations of ZKFederationStateStore calls"); @@ -126,30 +126,30 @@ public void addRegisterSubClusterCallDuration(long value) { } public void addDeregisterSubClusterCallDuration(long value) { - deregisterSubCluster.add(value); + deregisterSubClusterCall.add(value); } public void addSubClusterHeartbeatCallDuration(long value) { - subClusterHeartbeat.add(value); + subClusterHeartbeatCall.add(value); } public void addGetSubClusterCallDuration(long value) { - getSubCluster.add(value); + getSubClusterCall.add(value); } public void addGetSubClustersCallDuration(long value) { - getSubClusters.add(value); + getSubClustersCall.add(value); } public void addGetPolicyConfigurationDuration(long value) { - getPolicyConfiguration.add(value); + getPolicyConfigurationCall.add(value); } public void addSetPolicyConfigurationDuration(long value) { - setPolicyConfiguration.add(value); + setPolicyConfigurationCall.add(value); } public void addGetPoliciesConfigurationsDuration(long value) { - getPoliciesConfigurations.add(value); + getPoliciesConfigurationsCall.add(value); } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java index 51fd9c89437a2..bb117ea36ff6c 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java @@ -103,6 +103,14 @@ public void testMetricsInited() throws Exception { zkStateStoreOpDurations.addGetAppHomeSubClusterCallDuration(anyDuration); zkStateStoreOpDurations.addGetAppsHomeSubClusterCallDuration(anyDuration); zkStateStoreOpDurations.addDeleteAppHomeSubClusterCallDuration(anyDuration); + zkStateStoreOpDurations.addRegisterSubClusterCallDuration(anyDuration); + zkStateStoreOpDurations.addDeregisterSubClusterCallDuration(anyDuration); + zkStateStoreOpDurations.addSubClusterHeartbeatCallDuration(anyDuration); + zkStateStoreOpDurations.addGetSubClusterCallDuration(anyDuration); + zkStateStoreOpDurations.addGetSubClustersCallDuration(anyDuration); + zkStateStoreOpDurations.addGetPolicyConfigurationDuration(anyDuration); + zkStateStoreOpDurations.addSetPolicyConfigurationDuration(anyDuration); + zkStateStoreOpDurations.addGetPoliciesConfigurationsDuration(anyDuration); zkStateStoreOpDurations.getMetrics(collector, true); assertEquals("Incorrect number of perf metrics", 1, collector.getRecords().size()); @@ -117,6 +125,14 @@ public void testMetricsInited() throws Exception { MetricsRecords.assertMetric(record, "GetAppHomeSubClusterCallAvgTime", expectAvgTime); MetricsRecords.assertMetric(record, "GetAppsHomeSubClusterCallAvgTime", expectAvgTime); MetricsRecords.assertMetric(record, "DeleteAppHomeSubClusterCallAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "RegisterSubClusterCallAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "DeregisterSubClusterCallAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "SubClusterHeartbeatCallAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "GetSubClusterCallAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "GetSubClustersCallAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "GetPolicyConfigurationCallAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "SetPolicyConfigurationCallAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "GetPoliciesConfigurationsCallAvgTime", expectAvgTime); long expectOps = 1; MetricsRecords.assertMetric(record, "AddAppHomeSubClusterCallNumOps", expectOps); @@ -124,5 +140,13 @@ public void testMetricsInited() throws Exception { MetricsRecords.assertMetric(record, "GetAppHomeSubClusterCallNumOps", expectOps); MetricsRecords.assertMetric(record, "GetAppsHomeSubClusterCallNumOps", expectOps); MetricsRecords.assertMetric(record, "DeleteAppHomeSubClusterCallNumOps", expectOps); + MetricsRecords.assertMetric(record, "RegisterSubClusterCallNumOps", expectOps); + MetricsRecords.assertMetric(record, "DeregisterSubClusterCallNumOps", expectOps); + MetricsRecords.assertMetric(record, "SubClusterHeartbeatCallNumOps", expectOps); + MetricsRecords.assertMetric(record, "GetSubClusterCallNumOps", expectOps); + MetricsRecords.assertMetric(record, "GetSubClustersCallNumOps", expectOps); + MetricsRecords.assertMetric(record, "GetPolicyConfigurationCallNumOps", expectOps); + MetricsRecords.assertMetric(record, "SetPolicyConfigurationCallNumOps", expectOps); + MetricsRecords.assertMetric(record, "GetPoliciesConfigurationsCallNumOps", expectOps); } } \ No newline at end of file From 2a1454bc657f68da2f24b01eac5a52fd7ac2569c Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Tue, 16 Aug 2022 17:31:10 +0800 Subject: [PATCH 03/13] YARN-11250. Fix CheckStyle. --- .../ZKFederationStateStoreOpDurations.java | 72 +++++----- .../impl/ZookeeperFederationStateStore.java | 132 ++++++++++-------- .../TestZookeeperFederationStateStore.java | 72 +++++----- 3 files changed, 145 insertions(+), 131 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java index 9b065de7100fb..b429f4c6db580 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java @@ -36,43 +36,43 @@ public final class ZKFederationStateStoreOpDurations implements MetricsSource { @Metric("Duration for a add application homeSubcluster call") - private MutableRate addAppHomeSubClusterCall; + private MutableRate addAppHomeSubCluster; @Metric("Duration for a update application homeSubcluster call") - private MutableRate updateAppHomeSubClusterCall; + private MutableRate updateAppHomeSubCluster; @Metric("Duration for a get application homeSubcluster call") - private MutableRate getAppHomeSubClusterCall; + private MutableRate getAppHomeSubCluster; @Metric("Duration for a get applications homeSubcluster call") - private MutableRate getAppsHomeSubClusterCall; + private MutableRate getAppsHomeSubCluster; @Metric("Duration for a delete applications homeSubcluster call") - private MutableRate deleteAppHomeSubClusterCall; + private MutableRate deleteAppHomeSubCluster; @Metric("Duration for a register subCluster call") - private MutableRate registerSubClusterCall; + private MutableRate registerSubCluster; @Metric("Duration for a deregister subCluster call") - private MutableRate deregisterSubClusterCall; + private MutableRate deregisterSubCluster; @Metric("Duration for a subCluster Heartbeat call") - private MutableRate subClusterHeartbeatCall; + private MutableRate subClusterHeartbeat; @Metric("Duration for a get SubCluster call") - private MutableRate getSubClusterCall; + private MutableRate getSubCluster; @Metric("Duration for a get SubClusters call") - private MutableRate getSubClustersCall; + private MutableRate getSubClusters; @Metric("Duration for a get PolicyConfiguration call") - private MutableRate getPolicyConfigurationCall; + private MutableRate getPolicyConfiguration; @Metric("Duration for a set PolicyConfiguration call") - private MutableRate setPolicyConfigurationCall; + private MutableRate setPolicyConfiguration; @Metric("Duration for a get PolicyConfigurations call") - private MutableRate getPoliciesConfigurationsCall; + private MutableRate getPoliciesConfigurations; protected static final MetricsInfo RECORD_INFO = info("ZKFederationStateStoreOpDurations", "Durations of ZKFederationStateStore calls"); @@ -101,55 +101,55 @@ public synchronized void getMetrics(MetricsCollector collector, boolean all) { registry.snapshot(collector.addRecord(registry.info()), all); } - public void addAppHomeSubClusterCallDuration(long value) { - addAppHomeSubClusterCall.add(value); + public void addAppHomeSubClusterDuration(long value) { + addAppHomeSubCluster.add(value); } - public void addUpdateAppHomeSubClusterCallDuration(long value) { - updateAppHomeSubClusterCall.add(value); + public void addUpdateAppHomeSubClusterDuration(long value) { + updateAppHomeSubCluster.add(value); } - public void addGetAppHomeSubClusterCallDuration(long value) { - getAppHomeSubClusterCall.add(value); + public void addGetAppHomeSubClusterDuration(long value) { + getAppHomeSubCluster.add(value); } - public void addGetAppsHomeSubClusterCallDuration(long value) { - getAppsHomeSubClusterCall.add(value); + public void addGetAppsHomeSubClusterDuration(long value) { + getAppsHomeSubCluster.add(value); } - public void addDeleteAppHomeSubClusterCallDuration(long value) { - deleteAppHomeSubClusterCall.add(value); + public void addDeleteAppHomeSubClusterDuration(long value) { + deleteAppHomeSubCluster.add(value); } - public void addRegisterSubClusterCallDuration(long value) { - registerSubClusterCall.add(value); + public void addRegisterSubClusterDuration(long value) { + registerSubCluster.add(value); } - public void addDeregisterSubClusterCallDuration(long value) { - deregisterSubClusterCall.add(value); + public void addDeregisterSubClusterDuration(long value) { + deregisterSubCluster.add(value); } - public void addSubClusterHeartbeatCallDuration(long value) { - subClusterHeartbeatCall.add(value); + public void addSubClusterHeartbeatDuration(long value) { + subClusterHeartbeat.add(value); } - public void addGetSubClusterCallDuration(long value) { - getSubClusterCall.add(value); + public void addGetSubClusterDuration(long value) { + getSubCluster.add(value); } - public void addGetSubClustersCallDuration(long value) { - getSubClustersCall.add(value); + public void addGetSubClustersDuration(long value) { + getSubClusters.add(value); } public void addGetPolicyConfigurationDuration(long value) { - getPolicyConfigurationCall.add(value); + getPolicyConfiguration.add(value); } public void addSetPolicyConfigurationDuration(long value) { - setPolicyConfigurationCall.add(value); + setPolicyConfiguration.add(value); } public void addGetPoliciesConfigurationsDuration(long value) { - getPoliciesConfigurationsCall.add(value); + getPoliciesConfigurations.add(value); } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java index df5e74bd51f20..dd07943c0dee7 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java @@ -107,16 +107,20 @@ public class ZookeeperFederationStateStore implements FederationStateStore { private static final Logger LOG = - LoggerFactory.getLogger(ZookeeperFederationStateStore.class); + LoggerFactory.getLogger(ZookeeperFederationStateStore.class); private final static String ROOT_ZNODE_NAME_MEMBERSHIP = "memberships"; private final static String ROOT_ZNODE_NAME_APPLICATION = "applications"; private final static String ROOT_ZNODE_NAME_POLICY = "policies"; - /** Interface to Zookeeper. */ + /** + * Interface to Zookeeper. + */ private ZKCuratorManager zkManager; - /** Directory to store the state store data. */ + /** + * Directory to store the state store data. + */ private String baseZNode; private String appsZNode; @@ -127,15 +131,15 @@ public class ZookeeperFederationStateStore implements FederationStateStore { @VisibleForTesting private ZKFederationStateStoreOpDurations opDurations = - ZKFederationStateStoreOpDurations.getInstance(); + ZKFederationStateStoreOpDurations.getInstance(); @Override public void init(Configuration conf) throws YarnException { LOG.info("Initializing ZooKeeper connection"); baseZNode = conf.get( - YarnConfiguration.FEDERATION_STATESTORE_ZK_PARENT_PATH, - YarnConfiguration.DEFAULT_FEDERATION_STATESTORE_ZK_PARENT_PATH); + YarnConfiguration.FEDERATION_STATESTORE_ZK_PARENT_PATH, + YarnConfiguration.DEFAULT_FEDERATION_STATESTORE_ZK_PARENT_PATH); try { this.zkManager = new ZKCuratorManager(conf); this.zkManager.start(); @@ -169,7 +173,7 @@ public void close() throws Exception { @Override public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( - AddApplicationHomeSubClusterRequest request) throws YarnException { + AddApplicationHomeSubClusterRequest request) throws YarnException { long start = clock.getTime(); FederationApplicationHomeSubClusterStoreInputValidator.validate(request); @@ -193,15 +197,15 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - opDurations.addAppHomeSubClusterCallDuration(clock.getTime() - start); + opDurations.addAppHomeSubClusterDuration(clock.getTime() - start); return AddApplicationHomeSubClusterResponse.newInstance(homeSubCluster); } @Override public UpdateApplicationHomeSubClusterResponse - updateApplicationHomeSubCluster( + updateApplicationHomeSubCluster( UpdateApplicationHomeSubClusterRequest request) - throws YarnException { + throws YarnException { long start = clock.getTime(); FederationApplicationHomeSubClusterStoreInputValidator.validate(request); @@ -213,16 +217,16 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } SubClusterId newSubClusterId = - request.getApplicationHomeSubCluster().getHomeSubCluster(); + request.getApplicationHomeSubCluster().getHomeSubCluster(); putApp(appId, newSubClusterId, true); - opDurations.addUpdateAppHomeSubClusterCallDuration(clock.getTime() - start); + opDurations.addUpdateAppHomeSubClusterDuration(clock.getTime() - start); return UpdateApplicationHomeSubClusterResponse.newInstance(); } @Override public GetApplicationHomeSubClusterResponse getApplicationHomeSubCluster( - GetApplicationHomeSubClusterRequest request) throws YarnException { + GetApplicationHomeSubClusterRequest request) throws YarnException { long start = clock.getTime(); FederationApplicationHomeSubClusterStoreInputValidator.validate(request); @@ -232,14 +236,14 @@ public GetApplicationHomeSubClusterResponse getApplicationHomeSubCluster( String errMsg = "Application " + appId + " does not exist"; FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - opDurations.addGetAppHomeSubClusterCallDuration(clock.getTime() - start); + opDurations.addGetAppHomeSubClusterDuration(clock.getTime() - start); return GetApplicationHomeSubClusterResponse.newInstance( - ApplicationHomeSubCluster.newInstance(appId, homeSubCluster)); + ApplicationHomeSubCluster.newInstance(appId, homeSubCluster)); } @Override public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( - GetApplicationsHomeSubClusterRequest request) throws YarnException { + GetApplicationsHomeSubClusterRequest request) throws YarnException { long start = clock.getTime(); List result = new ArrayList<>(); @@ -248,22 +252,22 @@ public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( ApplicationId appId = ApplicationId.fromString(child); SubClusterId homeSubCluster = getApp(appId); ApplicationHomeSubCluster app = - ApplicationHomeSubCluster.newInstance(appId, homeSubCluster); + ApplicationHomeSubCluster.newInstance(appId, homeSubCluster); result.add(app); } } catch (Exception e) { String errMsg = "Cannot get apps: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - opDurations.addGetAppsHomeSubClusterCallDuration(clock.getTime() - start); + opDurations.addGetAppsHomeSubClusterDuration(clock.getTime() - start); return GetApplicationsHomeSubClusterResponse.newInstance(result); } @Override public DeleteApplicationHomeSubClusterResponse - deleteApplicationHomeSubCluster( + deleteApplicationHomeSubCluster( DeleteApplicationHomeSubClusterRequest request) - throws YarnException { + throws YarnException { long start = clock.getTime(); FederationApplicationHomeSubClusterStoreInputValidator.validate(request); ApplicationId appId = request.getApplicationId(); @@ -287,13 +291,13 @@ public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( String errMsg = "Cannot delete app: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - opDurations.addDeleteAppHomeSubClusterCallDuration(clock.getTime() - start); + opDurations.addDeleteAppHomeSubClusterDuration(clock.getTime() - start); return DeleteApplicationHomeSubClusterResponse.newInstance(); } @Override public SubClusterRegisterResponse registerSubCluster( - SubClusterRegisterRequest request) throws YarnException { + SubClusterRegisterRequest request) throws YarnException { long start = clock.getTime(); FederationMembershipStateStoreInputValidator.validate(request); SubClusterInfo subClusterInfo = request.getSubClusterInfo(); @@ -309,13 +313,13 @@ public SubClusterRegisterResponse registerSubCluster( String errMsg = "Cannot register subcluster: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - opDurations.addRegisterSubClusterCallDuration(clock.getTime() - start); + opDurations.addRegisterSubClusterDuration(clock.getTime() - start); return SubClusterRegisterResponse.newInstance(); } @Override public SubClusterDeregisterResponse deregisterSubCluster( - SubClusterDeregisterRequest request) throws YarnException { + SubClusterDeregisterRequest request) throws YarnException { long start = clock.getTime(); FederationMembershipStateStoreInputValidator.validate(request); SubClusterId subClusterId = request.getSubClusterId(); @@ -330,13 +334,13 @@ public SubClusterDeregisterResponse deregisterSubCluster( subClusterInfo.setState(state); putSubclusterInfo(subClusterId, subClusterInfo, true); } - opDurations.addDeregisterSubClusterCallDuration(clock.getTime() - start); + opDurations.addDeregisterSubClusterDuration(clock.getTime() - start); return SubClusterDeregisterResponse.newInstance(); } @Override public SubClusterHeartbeatResponse subClusterHeartbeat( - SubClusterHeartbeatRequest request) throws YarnException { + SubClusterHeartbeatRequest request) throws YarnException { long start = clock.getTime(); FederationMembershipStateStoreInputValidator.validate(request); SubClusterId subClusterId = request.getSubClusterId(); @@ -344,7 +348,7 @@ public SubClusterHeartbeatResponse subClusterHeartbeat( SubClusterInfo subClusterInfo = getSubclusterInfo(subClusterId); if (subClusterInfo == null) { String errMsg = "SubCluster " + subClusterId - + " does not exist; cannot heartbeat"; + + " does not exist; cannot heartbeat"; FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } @@ -354,13 +358,13 @@ public SubClusterHeartbeatResponse subClusterHeartbeat( subClusterInfo.setCapability(request.getCapability()); putSubclusterInfo(subClusterId, subClusterInfo, true); - opDurations.addSubClusterHeartbeatCallDuration(clock.getTime() - start); + opDurations.addSubClusterHeartbeatDuration(clock.getTime() - start); return SubClusterHeartbeatResponse.newInstance(); } @Override public GetSubClusterInfoResponse getSubCluster( - GetSubClusterInfoRequest request) throws YarnException { + GetSubClusterInfoRequest request) throws YarnException { long start = clock.getTime(); FederationMembershipStateStoreInputValidator.validate(request); SubClusterId subClusterId = request.getSubClusterId(); @@ -375,13 +379,13 @@ public GetSubClusterInfoResponse getSubCluster( String errMsg = "Cannot get subcluster: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - opDurations.addGetSubClusterCallDuration(clock.getTime() - start); + opDurations.addGetSubClusterDuration(clock.getTime() - start); return GetSubClusterInfoResponse.newInstance(subClusterInfo); } @Override public GetSubClustersInfoResponse getSubClusters( - GetSubClustersInfoRequest request) throws YarnException { + GetSubClustersInfoRequest request) throws YarnException { long start = clock.getTime(); List result = new ArrayList<>(); @@ -390,7 +394,7 @@ public GetSubClustersInfoResponse getSubClusters( SubClusterId subClusterId = SubClusterId.newInstance(child); SubClusterInfo info = getSubclusterInfo(subClusterId); if (!request.getFilterInactiveSubClusters() || - info.getState().isActive()) { + info.getState().isActive()) { result.add(info); } } @@ -398,14 +402,14 @@ public GetSubClustersInfoResponse getSubClusters( String errMsg = "Cannot get subclusters: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - opDurations.addGetSubClustersCallDuration(clock.getTime() - start); + opDurations.addGetSubClustersDuration(clock.getTime() - start); return GetSubClustersInfoResponse.newInstance(result); } @Override public GetSubClusterPolicyConfigurationResponse getPolicyConfiguration( - GetSubClusterPolicyConfigurationRequest request) throws YarnException { + GetSubClusterPolicyConfigurationRequest request) throws YarnException { long start = clock.getTime(); FederationPolicyStoreInputValidator.validate(request); String queue = request.getQueue(); @@ -423,16 +427,16 @@ public GetSubClusterPolicyConfigurationResponse getPolicyConfiguration( } opDurations.addGetPolicyConfigurationDuration(clock.getTime() - start); return GetSubClusterPolicyConfigurationResponse - .newInstance(policy); + .newInstance(policy); } @Override public SetSubClusterPolicyConfigurationResponse setPolicyConfiguration( - SetSubClusterPolicyConfigurationRequest request) throws YarnException { + SetSubClusterPolicyConfigurationRequest request) throws YarnException { long start = clock.getTime(); FederationPolicyStoreInputValidator.validate(request); SubClusterPolicyConfiguration policy = - request.getPolicyConfiguration(); + request.getPolicyConfiguration(); try { String queue = policy.getQueue(); putPolicy(queue, policy, true); @@ -446,7 +450,7 @@ public SetSubClusterPolicyConfigurationResponse setPolicyConfiguration( @Override public GetSubClusterPoliciesConfigurationsResponse getPoliciesConfigurations( - GetSubClusterPoliciesConfigurationsRequest request) throws YarnException { + GetSubClusterPoliciesConfigurationsRequest request) throws YarnException { long start = clock.getTime(); List result = new ArrayList<>(); @@ -479,6 +483,7 @@ public Version loadVersion() { /** * Get the subcluster for an application. + * * @param appId Application identifier. * @return Subcluster identifier. * @throws Exception If it cannot contact ZooKeeper. @@ -491,7 +496,7 @@ private SubClusterId getApp(final ApplicationId appId) throws YarnException { if (data != null) { try { subClusterId = new SubClusterIdPBImpl( - SubClusterIdProto.parseFrom(data)); + SubClusterIdProto.parseFrom(data)); } catch (InvalidProtocolBufferException e) { String errMsg = "Cannot parse application at " + appZNode; FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); @@ -502,28 +507,30 @@ private SubClusterId getApp(final ApplicationId appId) throws YarnException { /** * Put an application. - * @param appId Application identifier. + * + * @param appId Application identifier. * @param subClusterId Subcluster identifier. * @throws Exception If it cannot contact ZooKeeper. */ private void putApp(final ApplicationId appId, - final SubClusterId subClusterId, boolean update) + final SubClusterId subClusterId, boolean update) throws YarnException { String appZNode = getNodePath(appsZNode, appId.toString()); SubClusterIdProto proto = - ((SubClusterIdPBImpl)subClusterId).getProto(); + ((SubClusterIdPBImpl) subClusterId).getProto(); byte[] data = proto.toByteArray(); put(appZNode, data, update); } /** * Get the current information for a subcluster from Zookeeper. + * * @param subclusterId Subcluster identifier. * @return Subcluster information or null if it doesn't exist. * @throws Exception If it cannot contact ZooKeeper. */ private SubClusterInfo getSubclusterInfo(final SubClusterId subclusterId) - throws YarnException { + throws YarnException { String memberZNode = getNodePath(membershipZNode, subclusterId.toString()); SubClusterInfo policy = null; @@ -531,7 +538,7 @@ private SubClusterInfo getSubclusterInfo(final SubClusterId subclusterId) if (data != null) { try { policy = new SubClusterInfoPBImpl( - SubClusterInfoProto.parseFrom(data)); + SubClusterInfoProto.parseFrom(data)); } catch (InvalidProtocolBufferException e) { String errMsg = "Cannot parse subcluster info at " + memberZNode; FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); @@ -542,28 +549,30 @@ private SubClusterInfo getSubclusterInfo(final SubClusterId subclusterId) /** * Put the subcluster information in Zookeeper. - * @param subclusterId Subcluster identifier. + * + * @param subclusterId Subcluster identifier. * @param subClusterInfo Subcluster information. * @throws Exception If it cannot contact ZooKeeper. */ private void putSubclusterInfo(final SubClusterId subclusterId, - final SubClusterInfo subClusterInfo, final boolean update) + final SubClusterInfo subClusterInfo, final boolean update) throws YarnException { String memberZNode = getNodePath(membershipZNode, subclusterId.toString()); SubClusterInfoProto proto = - ((SubClusterInfoPBImpl)subClusterInfo).getProto(); + ((SubClusterInfoPBImpl) subClusterInfo).getProto(); byte[] data = proto.toByteArray(); put(memberZNode, data, update); } /** * Get the queue policy from Zookeeper. + * * @param queue Name of the queue. * @return Subcluster policy configuration. * @throws YarnException If it cannot contact ZooKeeper. */ private SubClusterPolicyConfiguration getPolicy(final String queue) - throws YarnException { + throws YarnException { String policyZNode = getNodePath(policiesZNode, queue); SubClusterPolicyConfiguration policy = null; @@ -571,7 +580,7 @@ private SubClusterPolicyConfiguration getPolicy(final String queue) if (data != null) { try { policy = new SubClusterPolicyConfigurationPBImpl( - SubClusterPolicyConfigurationProto.parseFrom(data)); + SubClusterPolicyConfigurationProto.parseFrom(data)); } catch (InvalidProtocolBufferException e) { String errMsg = "Cannot parse policy at " + policyZNode; FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); @@ -582,23 +591,25 @@ private SubClusterPolicyConfiguration getPolicy(final String queue) /** * Put the subcluster information in Zookeeper. - * @param queue Name of the queue. + * + * @param queue Name of the queue. * @param policy Subcluster policy configuration. * @throws YarnException If it cannot contact ZooKeeper. */ private void putPolicy(final String queue, - final SubClusterPolicyConfiguration policy, boolean update) + final SubClusterPolicyConfiguration policy, boolean update) throws YarnException { String policyZNode = getNodePath(policiesZNode, queue); SubClusterPolicyConfigurationProto proto = - ((SubClusterPolicyConfigurationPBImpl)policy).getProto(); + ((SubClusterPolicyConfigurationPBImpl) policy).getProto(); byte[] data = proto.toByteArray(); put(policyZNode, data, update); } /** * Get data from a znode in Zookeeper. + * * @param znode Path of the znode. * @return Data in the znode. * @throws YarnException If it cannot contact ZooKeeper. @@ -621,7 +632,7 @@ private byte[] get(String znode) throws YarnException { data = zkManager.getData(znode); } catch (Exception e) { String errMsg = "Cannot get data from znode " + znode - + ": " + e.getMessage(); + + ": " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } return data; @@ -629,12 +640,13 @@ private byte[] get(String znode) throws YarnException { /** * Put data into a znode in Zookeeper. + * * @param znode Path of the znode. - * @param data Data to write. + * @param data Data to write. * @throws YarnException If it cannot contact ZooKeeper. */ private void put(String znode, byte[] data, boolean update) - throws YarnException { + throws YarnException { // Create the znode boolean created = false; try { @@ -656,13 +668,14 @@ private void put(String znode, byte[] data, boolean update) zkManager.setData(znode, data, -1); } catch (Exception e) { String errMsg = "Cannot write data into znode " + znode - + ": " + e.getMessage(); + + ": " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } } /** * Get the current time. + * * @return Current time in milliseconds. */ private static long getCurrentTime() { @@ -673,21 +686,22 @@ private static long getCurrentTime() { @VisibleForTesting public ZKFederationStateStoreOpDurations getOpDurations() { return opDurations; + } @Override - public AddReservationHomeSubClusterResponse addReservationHomeSubCluster( + public AddReservationHomeSubClusterResponse addReservationHomeSubCluster ( AddReservationHomeSubClusterRequest request) throws YarnException { throw new NotImplementedException("Code is not implemented"); } @Override - public GetReservationHomeSubClusterResponse getReservationHomeSubCluster( + public GetReservationHomeSubClusterResponse getReservationHomeSubCluster ( GetReservationHomeSubClusterRequest request) throws YarnException { throw new NotImplementedException("Code is not implemented"); } @Override - public GetReservationsHomeSubClusterResponse getReservationsHomeSubCluster( + public GetReservationsHomeSubClusterResponse getReservationsHomeSubCluster ( GetReservationsHomeSubClusterRequest request) throws YarnException { throw new NotImplementedException("Code is not implemented"); } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java index bb117ea36ff6c..616740a0e08a4 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java @@ -98,16 +98,16 @@ public void testMetricsInited() throws Exception { MetricsCollectorImpl collector = new MetricsCollectorImpl(); long anyDuration = 10; - zkStateStoreOpDurations.addAppHomeSubClusterCallDuration(anyDuration); - zkStateStoreOpDurations.addUpdateAppHomeSubClusterCallDuration(anyDuration); - zkStateStoreOpDurations.addGetAppHomeSubClusterCallDuration(anyDuration); - zkStateStoreOpDurations.addGetAppsHomeSubClusterCallDuration(anyDuration); - zkStateStoreOpDurations.addDeleteAppHomeSubClusterCallDuration(anyDuration); - zkStateStoreOpDurations.addRegisterSubClusterCallDuration(anyDuration); - zkStateStoreOpDurations.addDeregisterSubClusterCallDuration(anyDuration); - zkStateStoreOpDurations.addSubClusterHeartbeatCallDuration(anyDuration); - zkStateStoreOpDurations.addGetSubClusterCallDuration(anyDuration); - zkStateStoreOpDurations.addGetSubClustersCallDuration(anyDuration); + zkStateStoreOpDurations.addAppHomeSubClusterDuration(anyDuration); + zkStateStoreOpDurations.addUpdateAppHomeSubClusterDuration(anyDuration); + zkStateStoreOpDurations.addGetAppHomeSubClusterDuration(anyDuration); + zkStateStoreOpDurations.addGetAppsHomeSubClusterDuration(anyDuration); + zkStateStoreOpDurations.addDeleteAppHomeSubClusterDuration(anyDuration); + zkStateStoreOpDurations.addRegisterSubClusterDuration(anyDuration); + zkStateStoreOpDurations.addDeregisterSubClusterDuration(anyDuration); + zkStateStoreOpDurations.addSubClusterHeartbeatDuration(anyDuration); + zkStateStoreOpDurations.addGetSubClusterDuration(anyDuration); + zkStateStoreOpDurations.addGetSubClustersDuration(anyDuration); zkStateStoreOpDurations.addGetPolicyConfigurationDuration(anyDuration); zkStateStoreOpDurations.addSetPolicyConfigurationDuration(anyDuration); zkStateStoreOpDurations.addGetPoliciesConfigurationsDuration(anyDuration); @@ -120,33 +120,33 @@ public void testMetricsInited() throws Exception { "ZKFederationStateStoreOpDurations"); double expectAvgTime = anyDuration; - MetricsRecords.assertMetric(record, "AddAppHomeSubClusterCallAvgTime", expectAvgTime); - MetricsRecords.assertMetric(record, "UpdateAppHomeSubClusterCallAvgTime", expectAvgTime); - MetricsRecords.assertMetric(record, "GetAppHomeSubClusterCallAvgTime", expectAvgTime); - MetricsRecords.assertMetric(record, "GetAppsHomeSubClusterCallAvgTime", expectAvgTime); - MetricsRecords.assertMetric(record, "DeleteAppHomeSubClusterCallAvgTime", expectAvgTime); - MetricsRecords.assertMetric(record, "RegisterSubClusterCallAvgTime", expectAvgTime); - MetricsRecords.assertMetric(record, "DeregisterSubClusterCallAvgTime", expectAvgTime); - MetricsRecords.assertMetric(record, "SubClusterHeartbeatCallAvgTime", expectAvgTime); - MetricsRecords.assertMetric(record, "GetSubClusterCallAvgTime", expectAvgTime); - MetricsRecords.assertMetric(record, "GetSubClustersCallAvgTime", expectAvgTime); - MetricsRecords.assertMetric(record, "GetPolicyConfigurationCallAvgTime", expectAvgTime); - MetricsRecords.assertMetric(record, "SetPolicyConfigurationCallAvgTime", expectAvgTime); - MetricsRecords.assertMetric(record, "GetPoliciesConfigurationsCallAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "AddAppHomeSubClusterAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "UpdateAppHomeSubClusterAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "GetAppHomeSubClusterAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "GetAppsHomeSubClusterAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "DeleteAppHomeSubClusterAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "RegisterSubClusterAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "DeregisterSubClusterAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "SubClusterHeartbeatAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "GetSubClusterAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "GetSubClustersAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "GetPolicyConfigurationAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "SetPolicyConfigurationAvgTime", expectAvgTime); + MetricsRecords.assertMetric(record, "GetPoliciesConfigurationsAvgTime", expectAvgTime); long expectOps = 1; - MetricsRecords.assertMetric(record, "AddAppHomeSubClusterCallNumOps", expectOps); - MetricsRecords.assertMetric(record, "UpdateAppHomeSubClusterCallNumOps", expectOps); - MetricsRecords.assertMetric(record, "GetAppHomeSubClusterCallNumOps", expectOps); - MetricsRecords.assertMetric(record, "GetAppsHomeSubClusterCallNumOps", expectOps); - MetricsRecords.assertMetric(record, "DeleteAppHomeSubClusterCallNumOps", expectOps); - MetricsRecords.assertMetric(record, "RegisterSubClusterCallNumOps", expectOps); - MetricsRecords.assertMetric(record, "DeregisterSubClusterCallNumOps", expectOps); - MetricsRecords.assertMetric(record, "SubClusterHeartbeatCallNumOps", expectOps); - MetricsRecords.assertMetric(record, "GetSubClusterCallNumOps", expectOps); - MetricsRecords.assertMetric(record, "GetSubClustersCallNumOps", expectOps); - MetricsRecords.assertMetric(record, "GetPolicyConfigurationCallNumOps", expectOps); - MetricsRecords.assertMetric(record, "SetPolicyConfigurationCallNumOps", expectOps); - MetricsRecords.assertMetric(record, "GetPoliciesConfigurationsCallNumOps", expectOps); + MetricsRecords.assertMetric(record, "AddAppHomeSubClusterNumOps", expectOps); + MetricsRecords.assertMetric(record, "UpdateAppHomeSubClusterNumOps", expectOps); + MetricsRecords.assertMetric(record, "GetAppHomeSubClusterNumOps", expectOps); + MetricsRecords.assertMetric(record, "GetAppsHomeSubClusterNumOps", expectOps); + MetricsRecords.assertMetric(record, "DeleteAppHomeSubClusterNumOps", expectOps); + MetricsRecords.assertMetric(record, "RegisterSubClusterNumOps", expectOps); + MetricsRecords.assertMetric(record, "DeregisterSubClusterNumOps", expectOps); + MetricsRecords.assertMetric(record, "SubClusterHeartbeatNumOps", expectOps); + MetricsRecords.assertMetric(record, "GetSubClusterNumOps", expectOps); + MetricsRecords.assertMetric(record, "GetSubClustersNumOps", expectOps); + MetricsRecords.assertMetric(record, "GetPolicyConfigurationNumOps", expectOps); + MetricsRecords.assertMetric(record, "SetPolicyConfigurationNumOps", expectOps); + MetricsRecords.assertMetric(record, "GetPoliciesConfigurationsNumOps", expectOps); } } \ No newline at end of file From df668f96d96c76206599caf5336877cd217c9cba Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Wed, 17 Aug 2022 08:18:41 +0800 Subject: [PATCH 04/13] YARN-11250. Fix CheckStyle. --- .../impl/ZookeeperFederationStateStore.java | 18 +++++++----------- 1 file changed, 7 insertions(+), 11 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java index dd07943c0dee7..60f4cbb44f325 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java @@ -107,20 +107,16 @@ public class ZookeeperFederationStateStore implements FederationStateStore { private static final Logger LOG = - LoggerFactory.getLogger(ZookeeperFederationStateStore.class); + LoggerFactory.getLogger(ZookeeperFederationStateStore.class); private final static String ROOT_ZNODE_NAME_MEMBERSHIP = "memberships"; private final static String ROOT_ZNODE_NAME_APPLICATION = "applications"; private final static String ROOT_ZNODE_NAME_POLICY = "policies"; - /** - * Interface to Zookeeper. - */ + /** Interface to Zookeeper. */ private ZKCuratorManager zkManager; - /** - * Directory to store the state store data. - */ + /** Directory to store the state store data.*/ private String baseZNode; private String appsZNode; @@ -131,15 +127,15 @@ public class ZookeeperFederationStateStore implements FederationStateStore { @VisibleForTesting private ZKFederationStateStoreOpDurations opDurations = - ZKFederationStateStoreOpDurations.getInstance(); + ZKFederationStateStoreOpDurations.getInstance(); @Override public void init(Configuration conf) throws YarnException { LOG.info("Initializing ZooKeeper connection"); baseZNode = conf.get( - YarnConfiguration.FEDERATION_STATESTORE_ZK_PARENT_PATH, - YarnConfiguration.DEFAULT_FEDERATION_STATESTORE_ZK_PARENT_PATH); + YarnConfiguration.FEDERATION_STATESTORE_ZK_PARENT_PATH, + YarnConfiguration.DEFAULT_FEDERATION_STATESTORE_ZK_PARENT_PATH); try { this.zkManager = new ZKCuratorManager(conf); this.zkManager.start(); @@ -217,7 +213,7 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } SubClusterId newSubClusterId = - request.getApplicationHomeSubCluster().getHomeSubCluster(); + request.getApplicationHomeSubCluster().getHomeSubCluster(); putApp(appId, newSubClusterId, true); opDurations.addUpdateAppHomeSubClusterDuration(clock.getTime() - start); From c7c90999c673a0e73720022dc4c5aceffa9dcd2e Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Wed, 17 Aug 2022 11:18:21 -0700 Subject: [PATCH 05/13] YARN-11250. Fix CheckStyle. --- .../impl/ZookeeperFederationStateStore.java | 36 +++++++++---------- 1 file changed, 18 insertions(+), 18 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java index 60f4cbb44f325..d1bdfb2149aac 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java @@ -116,7 +116,7 @@ public class ZookeeperFederationStateStore implements FederationStateStore { /** Interface to Zookeeper. */ private ZKCuratorManager zkManager; - /** Directory to store the state store data.*/ + /** Directory to store the state store data. */ private String baseZNode; private String appsZNode; @@ -169,7 +169,7 @@ public void close() throws Exception { @Override public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( - AddApplicationHomeSubClusterRequest request) throws YarnException { + AddApplicationHomeSubClusterRequest request) throws YarnException { long start = clock.getTime(); FederationApplicationHomeSubClusterStoreInputValidator.validate(request); @@ -199,9 +199,9 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( @Override public UpdateApplicationHomeSubClusterResponse - updateApplicationHomeSubCluster( + updateApplicationHomeSubCluster( UpdateApplicationHomeSubClusterRequest request) - throws YarnException { + throws YarnException { long start = clock.getTime(); FederationApplicationHomeSubClusterStoreInputValidator.validate(request); @@ -213,7 +213,7 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } SubClusterId newSubClusterId = - request.getApplicationHomeSubCluster().getHomeSubCluster(); + request.getApplicationHomeSubCluster().getHomeSubCluster(); putApp(appId, newSubClusterId, true); opDurations.addUpdateAppHomeSubClusterDuration(clock.getTime() - start); @@ -222,7 +222,7 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( @Override public GetApplicationHomeSubClusterResponse getApplicationHomeSubCluster( - GetApplicationHomeSubClusterRequest request) throws YarnException { + GetApplicationHomeSubClusterRequest request) throws YarnException { long start = clock.getTime(); FederationApplicationHomeSubClusterStoreInputValidator.validate(request); @@ -234,12 +234,12 @@ public GetApplicationHomeSubClusterResponse getApplicationHomeSubCluster( } opDurations.addGetAppHomeSubClusterDuration(clock.getTime() - start); return GetApplicationHomeSubClusterResponse.newInstance( - ApplicationHomeSubCluster.newInstance(appId, homeSubCluster)); + ApplicationHomeSubCluster.newInstance(appId, homeSubCluster)); } @Override public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( - GetApplicationsHomeSubClusterRequest request) throws YarnException { + GetApplicationsHomeSubClusterRequest request) throws YarnException { long start = clock.getTime(); List result = new ArrayList<>(); @@ -261,7 +261,7 @@ public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( @Override public DeleteApplicationHomeSubClusterResponse - deleteApplicationHomeSubCluster( + deleteApplicationHomeSubCluster( DeleteApplicationHomeSubClusterRequest request) throws YarnException { long start = clock.getTime(); @@ -293,7 +293,7 @@ public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( @Override public SubClusterRegisterResponse registerSubCluster( - SubClusterRegisterRequest request) throws YarnException { + SubClusterRegisterRequest request) throws YarnException { long start = clock.getTime(); FederationMembershipStateStoreInputValidator.validate(request); SubClusterInfo subClusterInfo = request.getSubClusterInfo(); @@ -315,7 +315,7 @@ public SubClusterRegisterResponse registerSubCluster( @Override public SubClusterDeregisterResponse deregisterSubCluster( - SubClusterDeregisterRequest request) throws YarnException { + SubClusterDeregisterRequest request) throws YarnException { long start = clock.getTime(); FederationMembershipStateStoreInputValidator.validate(request); SubClusterId subClusterId = request.getSubClusterId(); @@ -336,7 +336,7 @@ public SubClusterDeregisterResponse deregisterSubCluster( @Override public SubClusterHeartbeatResponse subClusterHeartbeat( - SubClusterHeartbeatRequest request) throws YarnException { + SubClusterHeartbeatRequest request) throws YarnException { long start = clock.getTime(); FederationMembershipStateStoreInputValidator.validate(request); SubClusterId subClusterId = request.getSubClusterId(); @@ -360,7 +360,7 @@ public SubClusterHeartbeatResponse subClusterHeartbeat( @Override public GetSubClusterInfoResponse getSubCluster( - GetSubClusterInfoRequest request) throws YarnException { + GetSubClusterInfoRequest request) throws YarnException { long start = clock.getTime(); FederationMembershipStateStoreInputValidator.validate(request); SubClusterId subClusterId = request.getSubClusterId(); @@ -381,7 +381,7 @@ public GetSubClusterInfoResponse getSubCluster( @Override public GetSubClustersInfoResponse getSubClusters( - GetSubClustersInfoRequest request) throws YarnException { + GetSubClustersInfoRequest request) throws YarnException { long start = clock.getTime(); List result = new ArrayList<>(); @@ -405,7 +405,7 @@ public GetSubClustersInfoResponse getSubClusters( @Override public GetSubClusterPolicyConfigurationResponse getPolicyConfiguration( - GetSubClusterPolicyConfigurationRequest request) throws YarnException { + GetSubClusterPolicyConfigurationRequest request) throws YarnException { long start = clock.getTime(); FederationPolicyStoreInputValidator.validate(request); String queue = request.getQueue(); @@ -423,12 +423,12 @@ public GetSubClusterPolicyConfigurationResponse getPolicyConfiguration( } opDurations.addGetPolicyConfigurationDuration(clock.getTime() - start); return GetSubClusterPolicyConfigurationResponse - .newInstance(policy); + .newInstance(policy); } @Override public SetSubClusterPolicyConfigurationResponse setPolicyConfiguration( - SetSubClusterPolicyConfigurationRequest request) throws YarnException { + SetSubClusterPolicyConfigurationRequest request) throws YarnException { long start = clock.getTime(); FederationPolicyStoreInputValidator.validate(request); SubClusterPolicyConfiguration policy = @@ -446,7 +446,7 @@ public SetSubClusterPolicyConfigurationResponse setPolicyConfiguration( @Override public GetSubClusterPoliciesConfigurationsResponse getPoliciesConfigurations( - GetSubClusterPoliciesConfigurationsRequest request) throws YarnException { + GetSubClusterPoliciesConfigurationsRequest request) throws YarnException { long start = clock.getTime(); List result = new ArrayList<>(); From a27b862b39e2aaec47d4c1c9aa17b4eed7bcf433 Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Wed, 17 Aug 2022 11:23:55 -0700 Subject: [PATCH 06/13] YARN-11250. Fix CheckStyle. --- .../impl/ZookeeperFederationStateStore.java | 33 +++++++++---------- 1 file changed, 16 insertions(+), 17 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java index d1bdfb2149aac..9abbaf952f5bc 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java @@ -248,7 +248,7 @@ public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( ApplicationId appId = ApplicationId.fromString(child); SubClusterId homeSubCluster = getApp(appId); ApplicationHomeSubCluster app = - ApplicationHomeSubCluster.newInstance(appId, homeSubCluster); + ApplicationHomeSubCluster.newInstance(appId, homeSubCluster); result.add(app); } } catch (Exception e) { @@ -263,7 +263,7 @@ public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( public DeleteApplicationHomeSubClusterResponse deleteApplicationHomeSubCluster( DeleteApplicationHomeSubClusterRequest request) - throws YarnException { + throws YarnException { long start = clock.getTime(); FederationApplicationHomeSubClusterStoreInputValidator.validate(request); ApplicationId appId = request.getApplicationId(); @@ -390,7 +390,7 @@ public GetSubClustersInfoResponse getSubClusters( SubClusterId subClusterId = SubClusterId.newInstance(child); SubClusterInfo info = getSubclusterInfo(subClusterId); if (!request.getFilterInactiveSubClusters() || - info.getState().isActive()) { + info.getState().isActive()) { result.add(info); } } @@ -432,7 +432,7 @@ public SetSubClusterPolicyConfigurationResponse setPolicyConfiguration( long start = clock.getTime(); FederationPolicyStoreInputValidator.validate(request); SubClusterPolicyConfiguration policy = - request.getPolicyConfiguration(); + request.getPolicyConfiguration(); try { String queue = policy.getQueue(); putPolicy(queue, policy, true); @@ -492,7 +492,7 @@ private SubClusterId getApp(final ApplicationId appId) throws YarnException { if (data != null) { try { subClusterId = new SubClusterIdPBImpl( - SubClusterIdProto.parseFrom(data)); + SubClusterIdProto.parseFrom(data)); } catch (InvalidProtocolBufferException e) { String errMsg = "Cannot parse application at " + appZNode; FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); @@ -509,11 +509,11 @@ private SubClusterId getApp(final ApplicationId appId) throws YarnException { * @throws Exception If it cannot contact ZooKeeper. */ private void putApp(final ApplicationId appId, - final SubClusterId subClusterId, boolean update) + final SubClusterId subClusterId, boolean update) throws YarnException { String appZNode = getNodePath(appsZNode, appId.toString()); SubClusterIdProto proto = - ((SubClusterIdPBImpl) subClusterId).getProto(); + ((SubClusterIdPBImpl) subClusterId).getProto(); byte[] data = proto.toByteArray(); put(appZNode, data, update); } @@ -526,7 +526,7 @@ private void putApp(final ApplicationId appId, * @throws Exception If it cannot contact ZooKeeper. */ private SubClusterInfo getSubclusterInfo(final SubClusterId subclusterId) - throws YarnException { + throws YarnException { String memberZNode = getNodePath(membershipZNode, subclusterId.toString()); SubClusterInfo policy = null; @@ -534,7 +534,7 @@ private SubClusterInfo getSubclusterInfo(final SubClusterId subclusterId) if (data != null) { try { policy = new SubClusterInfoPBImpl( - SubClusterInfoProto.parseFrom(data)); + SubClusterInfoProto.parseFrom(data)); } catch (InvalidProtocolBufferException e) { String errMsg = "Cannot parse subcluster info at " + memberZNode; FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); @@ -545,13 +545,12 @@ private SubClusterInfo getSubclusterInfo(final SubClusterId subclusterId) /** * Put the subcluster information in Zookeeper. - * * @param subclusterId Subcluster identifier. * @param subClusterInfo Subcluster information. * @throws Exception If it cannot contact ZooKeeper. */ private void putSubclusterInfo(final SubClusterId subclusterId, - final SubClusterInfo subClusterInfo, final boolean update) + final SubClusterInfo subClusterInfo, final boolean update) throws YarnException { String memberZNode = getNodePath(membershipZNode, subclusterId.toString()); SubClusterInfoProto proto = @@ -568,7 +567,7 @@ private void putSubclusterInfo(final SubClusterId subclusterId, * @throws YarnException If it cannot contact ZooKeeper. */ private SubClusterPolicyConfiguration getPolicy(final String queue) - throws YarnException { + throws YarnException { String policyZNode = getNodePath(policiesZNode, queue); SubClusterPolicyConfiguration policy = null; @@ -593,7 +592,7 @@ private SubClusterPolicyConfiguration getPolicy(final String queue) * @throws YarnException If it cannot contact ZooKeeper. */ private void putPolicy(final String queue, - final SubClusterPolicyConfiguration policy, boolean update) + final SubClusterPolicyConfiguration policy, boolean update) throws YarnException { String policyZNode = getNodePath(policiesZNode, queue); @@ -642,7 +641,7 @@ private byte[] get(String znode) throws YarnException { * @throws YarnException If it cannot contact ZooKeeper. */ private void put(String znode, byte[] data, boolean update) - throws YarnException { + throws YarnException { // Create the znode boolean created = false; try { @@ -685,19 +684,19 @@ public ZKFederationStateStoreOpDurations getOpDurations() { } @Override - public AddReservationHomeSubClusterResponse addReservationHomeSubCluster ( + public AddReservationHomeSubClusterResponse addReservationHomeSubCluster( AddReservationHomeSubClusterRequest request) throws YarnException { throw new NotImplementedException("Code is not implemented"); } @Override - public GetReservationHomeSubClusterResponse getReservationHomeSubCluster ( + public GetReservationHomeSubClusterResponse getReservationHomeSubCluster( GetReservationHomeSubClusterRequest request) throws YarnException { throw new NotImplementedException("Code is not implemented"); } @Override - public GetReservationsHomeSubClusterResponse getReservationsHomeSubCluster ( + public GetReservationsHomeSubClusterResponse getReservationsHomeSubCluster( GetReservationsHomeSubClusterRequest request) throws YarnException { throw new NotImplementedException("Code is not implemented"); } From 95b9e4e14ebb7ad5c6dc74b5bdbdc576083b49cc Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Wed, 17 Aug 2022 11:28:14 -0700 Subject: [PATCH 07/13] YARN-11250. Fix CheckStyle. --- .../impl/ZookeeperFederationStateStore.java | 25 ++++++++----------- 1 file changed, 11 insertions(+), 14 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java index 9abbaf952f5bc..b065d96ed6d1a 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java @@ -201,7 +201,7 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( public UpdateApplicationHomeSubClusterResponse updateApplicationHomeSubCluster( UpdateApplicationHomeSubClusterRequest request) - throws YarnException { + throws YarnException { long start = clock.getTime(); FederationApplicationHomeSubClusterStoreInputValidator.validate(request); @@ -344,7 +344,7 @@ public SubClusterHeartbeatResponse subClusterHeartbeat( SubClusterInfo subClusterInfo = getSubclusterInfo(subClusterId); if (subClusterInfo == null) { String errMsg = "SubCluster " + subClusterId - + " does not exist; cannot heartbeat"; + + " does not exist; cannot heartbeat"; FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } @@ -479,7 +479,6 @@ public Version loadVersion() { /** * Get the subcluster for an application. - * * @param appId Application identifier. * @return Subcluster identifier. * @throws Exception If it cannot contact ZooKeeper. @@ -503,8 +502,7 @@ private SubClusterId getApp(final ApplicationId appId) throws YarnException { /** * Put an application. - * - * @param appId Application identifier. + * @param appId pplication identifier. * @param subClusterId Subcluster identifier. * @throws Exception If it cannot contact ZooKeeper. */ @@ -513,14 +511,13 @@ private void putApp(final ApplicationId appId, throws YarnException { String appZNode = getNodePath(appsZNode, appId.toString()); SubClusterIdProto proto = - ((SubClusterIdPBImpl) subClusterId).getProto(); + ((SubClusterIdPBImpl)subClusterId).getProto(); byte[] data = proto.toByteArray(); put(appZNode, data, update); } /** * Get the current information for a subcluster from Zookeeper. - * * @param subclusterId Subcluster identifier. * @return Subcluster information or null if it doesn't exist. * @throws Exception If it cannot contact ZooKeeper. @@ -534,7 +531,7 @@ private SubClusterInfo getSubclusterInfo(final SubClusterId subclusterId) if (data != null) { try { policy = new SubClusterInfoPBImpl( - SubClusterInfoProto.parseFrom(data)); + SubClusterInfoProto.parseFrom(data)); } catch (InvalidProtocolBufferException e) { String errMsg = "Cannot parse subcluster info at " + memberZNode; FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); @@ -545,7 +542,7 @@ private SubClusterInfo getSubclusterInfo(final SubClusterId subclusterId) /** * Put the subcluster information in Zookeeper. - * @param subclusterId Subcluster identifier. + * @param subclusterId Subcluster identifier. * @param subClusterInfo Subcluster information. * @throws Exception If it cannot contact ZooKeeper. */ @@ -554,7 +551,7 @@ private void putSubclusterInfo(final SubClusterId subclusterId, throws YarnException { String memberZNode = getNodePath(membershipZNode, subclusterId.toString()); SubClusterInfoProto proto = - ((SubClusterInfoPBImpl) subClusterInfo).getProto(); + ((SubClusterInfoPBImpl)subClusterInfo).getProto(); byte[] data = proto.toByteArray(); put(memberZNode, data, update); } @@ -575,7 +572,7 @@ private SubClusterPolicyConfiguration getPolicy(final String queue) if (data != null) { try { policy = new SubClusterPolicyConfigurationPBImpl( - SubClusterPolicyConfigurationProto.parseFrom(data)); + SubClusterPolicyConfigurationProto.parseFrom(data)); } catch (InvalidProtocolBufferException e) { String errMsg = "Cannot parse policy at " + policyZNode; FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); @@ -597,7 +594,7 @@ private void putPolicy(final String queue, String policyZNode = getNodePath(policiesZNode, queue); SubClusterPolicyConfigurationProto proto = - ((SubClusterPolicyConfigurationPBImpl) policy).getProto(); + ((SubClusterPolicyConfigurationPBImpl) policy).getProto(); byte[] data = proto.toByteArray(); put(policyZNode, data, update); } @@ -627,7 +624,7 @@ private byte[] get(String znode) throws YarnException { data = zkManager.getData(znode); } catch (Exception e) { String errMsg = "Cannot get data from znode " + znode - + ": " + e.getMessage(); + + ": " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } return data; @@ -663,7 +660,7 @@ private void put(String znode, byte[] data, boolean update) zkManager.setData(znode, data, -1); } catch (Exception e) { String errMsg = "Cannot write data into znode " + znode - + ": " + e.getMessage(); + + ": " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } } From 9e0bea98301dd35defc94cd65fee90990b30d9f1 Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Wed, 17 Aug 2022 11:31:23 -0700 Subject: [PATCH 08/13] YARN-11250. Fix CheckStyle. --- .../store/impl/ZookeeperFederationStateStore.java | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java index b065d96ed6d1a..8fb9a678eb025 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java @@ -502,7 +502,7 @@ private SubClusterId getApp(final ApplicationId appId) throws YarnException { /** * Put an application. - * @param appId pplication identifier. + * @param appId Application identifier. * @param subClusterId Subcluster identifier. * @throws Exception If it cannot contact ZooKeeper. */ @@ -583,7 +583,6 @@ private SubClusterPolicyConfiguration getPolicy(final String queue) /** * Put the subcluster information in Zookeeper. - * * @param queue Name of the queue. * @param policy Subcluster policy configuration. * @throws YarnException If it cannot contact ZooKeeper. @@ -594,14 +593,13 @@ private void putPolicy(final String queue, String policyZNode = getNodePath(policiesZNode, queue); SubClusterPolicyConfigurationProto proto = - ((SubClusterPolicyConfigurationPBImpl) policy).getProto(); + ((SubClusterPolicyConfigurationPBImpl)policy).getProto(); byte[] data = proto.toByteArray(); put(policyZNode, data, update); } /** * Get data from a znode in Zookeeper. - * * @param znode Path of the znode. * @return Data in the znode. * @throws YarnException If it cannot contact ZooKeeper. @@ -634,7 +632,7 @@ private byte[] get(String znode) throws YarnException { * Put data into a znode in Zookeeper. * * @param znode Path of the znode. - * @param data Data to write. + * @param data Data to write. * @throws YarnException If it cannot contact ZooKeeper. */ private void put(String znode, byte[] data, boolean update) @@ -667,7 +665,6 @@ private void put(String znode, byte[] data, boolean update) /** * Get the current time. - * * @return Current time in milliseconds. */ private static long getCurrentTime() { From dec1841376308d0dc9cb5646bd22c0b36a60e2d0 Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Wed, 17 Aug 2022 11:33:41 -0700 Subject: [PATCH 09/13] YARN-11250. Fix CheckStyle. --- .../federation/store/impl/ZookeeperFederationStateStore.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java index 8fb9a678eb025..c36fa6404bf91 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java @@ -583,7 +583,7 @@ private SubClusterPolicyConfiguration getPolicy(final String queue) /** * Put the subcluster information in Zookeeper. - * @param queue Name of the queue. + * @param queue Name of the queue. * @param policy Subcluster policy configuration. * @throws YarnException If it cannot contact ZooKeeper. */ @@ -630,7 +630,6 @@ private byte[] get(String znode) throws YarnException { /** * Put data into a znode in Zookeeper. - * * @param znode Path of the znode. * @param data Data to write. * @throws YarnException If it cannot contact ZooKeeper. From e78395f338a28c9ef99572f669bc49fcbd3391bd Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Wed, 17 Aug 2022 11:46:12 -0700 Subject: [PATCH 10/13] YARN-11250. Fix CheckStyle. --- .../federation/store/impl/MemoryFederationStateStore.java | 3 +-- .../federation/store/impl/SQLFederationStateStore.java | 3 +-- .../store/impl/ZookeeperFederationStateStore.java | 8 ++++---- .../records/GetApplicationHomeSubClusterResponse.java | 5 ++++- 4 files changed, 10 insertions(+), 9 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/MemoryFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/MemoryFederationStateStore.java index 24ed54d63668b..4b07b7b1b93f2 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/MemoryFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/MemoryFederationStateStore.java @@ -245,8 +245,7 @@ public GetApplicationHomeSubClusterResponse getApplicationHomeSubCluster( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - return GetApplicationHomeSubClusterResponse.newInstance( - ApplicationHomeSubCluster.newInstance(appId, applications.get(appId))); + return GetApplicationHomeSubClusterResponse.newInstance(appId, applications.get(appId)); } @Override diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/SQLFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/SQLFederationStateStore.java index dfcfb06bb46b8..fc1a69fad899e 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/SQLFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/SQLFederationStateStore.java @@ -716,8 +716,7 @@ public GetApplicationHomeSubClusterResponse getApplicationHomeSubCluster( FederationStateStoreUtils.returnToPool(LOG, cstmt); } return GetApplicationHomeSubClusterResponse - .newInstance(ApplicationHomeSubCluster - .newInstance(request.getApplicationId(), homeRM)); + .newInstance(request.getApplicationId(), homeRM); } @Override diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java index c36fa6404bf91..41d3247f9533a 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java @@ -194,7 +194,9 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( } opDurations.addAppHomeSubClusterDuration(clock.getTime() - start); - return AddApplicationHomeSubClusterResponse.newInstance(homeSubCluster); + + return AddApplicationHomeSubClusterResponse + .newInstance(homeSubCluster); } @Override @@ -233,8 +235,7 @@ public GetApplicationHomeSubClusterResponse getApplicationHomeSubCluster( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } opDurations.addGetAppHomeSubClusterDuration(clock.getTime() - start); - return GetApplicationHomeSubClusterResponse.newInstance( - ApplicationHomeSubCluster.newInstance(appId, homeSubCluster)); + return GetApplicationHomeSubClusterResponse.newInstance(appId, homeSubCluster); } @Override @@ -558,7 +559,6 @@ private void putSubclusterInfo(final SubClusterId subclusterId, /** * Get the queue policy from Zookeeper. - * * @param queue Name of the queue. * @return Subcluster policy configuration. * @throws YarnException If it cannot contact ZooKeeper. diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/records/GetApplicationHomeSubClusterResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/records/GetApplicationHomeSubClusterResponse.java index 60735b382f16a..6144b01e86060 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/records/GetApplicationHomeSubClusterResponse.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/records/GetApplicationHomeSubClusterResponse.java @@ -20,6 +20,7 @@ import org.apache.hadoop.classification.InterfaceAudience.Private; 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.util.Records; /** @@ -42,7 +43,9 @@ public abstract class GetApplicationHomeSubClusterResponse { @Private @Unstable public static GetApplicationHomeSubClusterResponse newInstance( - ApplicationHomeSubCluster applicationHomeSubCluster) { + ApplicationId appId, SubClusterId homeSubCluster) { + ApplicationHomeSubCluster applicationHomeSubCluster = + ApplicationHomeSubCluster.newInstance(appId, homeSubCluster); GetApplicationHomeSubClusterResponse mapResponse = Records.newRecord(GetApplicationHomeSubClusterResponse.class); mapResponse.setApplicationHomeSubCluster(applicationHomeSubCluster); From e7b6b694d661357d288253ad6c0996825e855a46 Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Wed, 17 Aug 2022 12:09:01 -0700 Subject: [PATCH 11/13] YARN-11250. Fix CheckStyle. --- .../ZKFederationStateStoreOpDurations.java | 52 +++++++++---------- .../impl/ZookeeperFederationStateStore.java | 41 +++++++++------ .../TestZookeeperFederationStateStore.java | 30 ++++++----- 3 files changed, 69 insertions(+), 54 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java index b429f4c6db580..47fbf2f4ee630 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java @@ -101,55 +101,55 @@ public synchronized void getMetrics(MetricsCollector collector, boolean all) { registry.snapshot(collector.addRecord(registry.info()), all); } - public void addAppHomeSubClusterDuration(long value) { - addAppHomeSubCluster.add(value); + public void addAppHomeSubClusterDuration(long endTime, long startTime) { + addAppHomeSubCluster.add(endTime - startTime); } - public void addUpdateAppHomeSubClusterDuration(long value) { - updateAppHomeSubCluster.add(value); + public void addUpdateAppHomeSubClusterDuration(long endTime, long startTime) { + updateAppHomeSubCluster.add(endTime - startTime); } - public void addGetAppHomeSubClusterDuration(long value) { - getAppHomeSubCluster.add(value); + public void addGetAppHomeSubClusterDuration(long endTime, long startTime) { + getAppHomeSubCluster.add(endTime - startTime); } - public void addGetAppsHomeSubClusterDuration(long value) { - getAppsHomeSubCluster.add(value); + public void addGetAppsHomeSubClusterDuration(long endTime, long startTime) { + getAppsHomeSubCluster.add(endTime - startTime); } - public void addDeleteAppHomeSubClusterDuration(long value) { - deleteAppHomeSubCluster.add(value); + public void addDeleteAppHomeSubClusterDuration(long endTime, long startTime) { + deleteAppHomeSubCluster.add(endTime - startTime); } - public void addRegisterSubClusterDuration(long value) { - registerSubCluster.add(value); + public void addRegisterSubClusterDuration(long endTime, long startTime) { + registerSubCluster.add(endTime - startTime); } - public void addDeregisterSubClusterDuration(long value) { - deregisterSubCluster.add(value); + public void addDeregisterSubClusterDuration(long endTime, long startTime) { + deregisterSubCluster.add(endTime - startTime); } - public void addSubClusterHeartbeatDuration(long value) { - subClusterHeartbeat.add(value); + public void addSubClusterHeartbeatDuration(long endTime, long startTime) { + subClusterHeartbeat.add(endTime - startTime); } - public void addGetSubClusterDuration(long value) { - getSubCluster.add(value); + public void addGetSubClusterDuration(long endTime, long startTime) { + getSubCluster.add(endTime - startTime); } - public void addGetSubClustersDuration(long value) { - getSubClusters.add(value); + public void addGetSubClustersDuration(long endTime, long startTime) { + getSubClusters.add(endTime - startTime); } - public void addGetPolicyConfigurationDuration(long value) { - getPolicyConfiguration.add(value); + public void addGetPolicyConfigurationDuration(long endTime, long startTime) { + getPolicyConfiguration.add(endTime - startTime); } - public void addSetPolicyConfigurationDuration(long value) { - setPolicyConfiguration.add(value); + public void addSetPolicyConfigurationDuration(long endTime, long startTime) { + setPolicyConfiguration.add(endTime - startTime); } - public void addGetPoliciesConfigurationsDuration(long value) { - getPoliciesConfigurations.add(value); + public void addGetPoliciesConfigurationsDuration(long endTime, long startTime) { + getPoliciesConfigurations.add(endTime - startTime); } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java index 41d3247f9533a..639eddc945420 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java @@ -192,9 +192,8 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( String errMsg = "Cannot check app home subcluster for " + appId; FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - - opDurations.addAppHomeSubClusterDuration(clock.getTime() - start); - + long end = clock.getTime(); + opDurations.addAppHomeSubClusterDuration(end, start); return AddApplicationHomeSubClusterResponse .newInstance(homeSubCluster); } @@ -218,7 +217,8 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( request.getApplicationHomeSubCluster().getHomeSubCluster(); putApp(appId, newSubClusterId, true); - opDurations.addUpdateAppHomeSubClusterDuration(clock.getTime() - start); + long end = clock.getTime(); + opDurations.addUpdateAppHomeSubClusterDuration(end, start); return UpdateApplicationHomeSubClusterResponse.newInstance(); } @@ -234,7 +234,8 @@ public GetApplicationHomeSubClusterResponse getApplicationHomeSubCluster( String errMsg = "Application " + appId + " does not exist"; FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - opDurations.addGetAppHomeSubClusterDuration(clock.getTime() - start); + long end = clock.getTime(); + opDurations.addGetAppHomeSubClusterDuration(end, start); return GetApplicationHomeSubClusterResponse.newInstance(appId, homeSubCluster); } @@ -256,7 +257,8 @@ public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( String errMsg = "Cannot get apps: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - opDurations.addGetAppsHomeSubClusterDuration(clock.getTime() - start); + long end = clock.getTime(); + opDurations.addGetAppsHomeSubClusterDuration(end, start); return GetApplicationsHomeSubClusterResponse.newInstance(result); } @@ -288,7 +290,8 @@ public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( String errMsg = "Cannot delete app: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - opDurations.addDeleteAppHomeSubClusterDuration(clock.getTime() - start); + long end = clock.getTime(); + opDurations.addDeleteAppHomeSubClusterDuration(end, start); return DeleteApplicationHomeSubClusterResponse.newInstance(); } @@ -310,7 +313,8 @@ public SubClusterRegisterResponse registerSubCluster( String errMsg = "Cannot register subcluster: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - opDurations.addRegisterSubClusterDuration(clock.getTime() - start); + long end = clock.getTime(); + opDurations.addRegisterSubClusterDuration(end, start); return SubClusterRegisterResponse.newInstance(); } @@ -331,7 +335,8 @@ public SubClusterDeregisterResponse deregisterSubCluster( subClusterInfo.setState(state); putSubclusterInfo(subClusterId, subClusterInfo, true); } - opDurations.addDeregisterSubClusterDuration(clock.getTime() - start); + long end = clock.getTime(); + opDurations.addDeregisterSubClusterDuration(end, start); return SubClusterDeregisterResponse.newInstance(); } @@ -355,7 +360,8 @@ public SubClusterHeartbeatResponse subClusterHeartbeat( subClusterInfo.setCapability(request.getCapability()); putSubclusterInfo(subClusterId, subClusterInfo, true); - opDurations.addSubClusterHeartbeatDuration(clock.getTime() - start); + long end = clock.getTime(); + opDurations.addSubClusterHeartbeatDuration(clock.getTime(), start); return SubClusterHeartbeatResponse.newInstance(); } @@ -376,7 +382,8 @@ public GetSubClusterInfoResponse getSubCluster( String errMsg = "Cannot get subcluster: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - opDurations.addGetSubClusterDuration(clock.getTime() - start); + long end = clock.getTime(); + opDurations.addGetSubClusterDuration(end, start); return GetSubClusterInfoResponse.newInstance(subClusterInfo); } @@ -399,7 +406,8 @@ public GetSubClustersInfoResponse getSubClusters( String errMsg = "Cannot get subclusters: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - opDurations.addGetSubClustersDuration(clock.getTime() - start); + long end = clock.getTime(); + opDurations.addGetSubClustersDuration(end, start); return GetSubClustersInfoResponse.newInstance(result); } @@ -422,7 +430,8 @@ public GetSubClusterPolicyConfigurationResponse getPolicyConfiguration( LOG.warn("Policy for queue: {} does not exist.", queue); return null; } - opDurations.addGetPolicyConfigurationDuration(clock.getTime() - start); + long end = clock.getTime(); + opDurations.addGetPolicyConfigurationDuration(end, start); return GetSubClusterPolicyConfigurationResponse .newInstance(policy); } @@ -441,7 +450,8 @@ public SetSubClusterPolicyConfigurationResponse setPolicyConfiguration( String errMsg = "Cannot set policy: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - opDurations.addSetPolicyConfigurationDuration(clock.getTime() - start); + long end = clock.getTime(); + opDurations.addSetPolicyConfigurationDuration(end, start); return SetSubClusterPolicyConfigurationResponse.newInstance(); } @@ -464,7 +474,8 @@ public GetSubClusterPoliciesConfigurationsResponse getPoliciesConfigurations( String errMsg = "Cannot get policies: " + e.getMessage(); FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } - opDurations.addGetPoliciesConfigurationsDuration(clock.getTime() - start); + long end = clock.getTime(); + opDurations.addGetPoliciesConfigurationsDuration(end, start); return GetSubClusterPoliciesConfigurationsResponse.newInstance(result); } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java index 616740a0e08a4..68050ef2a04cf 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java @@ -28,6 +28,7 @@ import org.apache.hadoop.metrics2.MetricsRecord; import org.apache.hadoop.metrics2.impl.MetricsCollectorImpl; import org.apache.hadoop.metrics2.impl.MetricsRecords; +import org.apache.hadoop.util.Time; import org.apache.hadoop.yarn.conf.YarnConfiguration; import org.apache.hadoop.yarn.exceptions.YarnException; import org.apache.hadoop.yarn.server.federation.store.FederationStateStore; @@ -98,19 +99,22 @@ public void testMetricsInited() throws Exception { MetricsCollectorImpl collector = new MetricsCollectorImpl(); long anyDuration = 10; - zkStateStoreOpDurations.addAppHomeSubClusterDuration(anyDuration); - zkStateStoreOpDurations.addUpdateAppHomeSubClusterDuration(anyDuration); - zkStateStoreOpDurations.addGetAppHomeSubClusterDuration(anyDuration); - zkStateStoreOpDurations.addGetAppsHomeSubClusterDuration(anyDuration); - zkStateStoreOpDurations.addDeleteAppHomeSubClusterDuration(anyDuration); - zkStateStoreOpDurations.addRegisterSubClusterDuration(anyDuration); - zkStateStoreOpDurations.addDeregisterSubClusterDuration(anyDuration); - zkStateStoreOpDurations.addSubClusterHeartbeatDuration(anyDuration); - zkStateStoreOpDurations.addGetSubClusterDuration(anyDuration); - zkStateStoreOpDurations.addGetSubClustersDuration(anyDuration); - zkStateStoreOpDurations.addGetPolicyConfigurationDuration(anyDuration); - zkStateStoreOpDurations.addSetPolicyConfigurationDuration(anyDuration); - zkStateStoreOpDurations.addGetPoliciesConfigurationsDuration(anyDuration); + long start = Time.now(); + long end = start + anyDuration; + + zkStateStoreOpDurations.addAppHomeSubClusterDuration(end, start); + zkStateStoreOpDurations.addUpdateAppHomeSubClusterDuration(end, start); + zkStateStoreOpDurations.addGetAppHomeSubClusterDuration(end, start); + zkStateStoreOpDurations.addGetAppsHomeSubClusterDuration(end, start); + zkStateStoreOpDurations.addDeleteAppHomeSubClusterDuration(end, start); + zkStateStoreOpDurations.addRegisterSubClusterDuration(end, start); + zkStateStoreOpDurations.addDeregisterSubClusterDuration(end, start); + zkStateStoreOpDurations.addSubClusterHeartbeatDuration(end, start); + zkStateStoreOpDurations.addGetSubClusterDuration(end, start); + zkStateStoreOpDurations.addGetSubClustersDuration(end, start); + zkStateStoreOpDurations.addGetPolicyConfigurationDuration(end, start); + zkStateStoreOpDurations.addSetPolicyConfigurationDuration(end, start); + zkStateStoreOpDurations.addGetPoliciesConfigurationsDuration(end, start); zkStateStoreOpDurations.getMetrics(collector, true); assertEquals("Incorrect number of perf metrics", 1, collector.getRecords().size()); From 644fcebbac81cea52c20644e4341e188a130aa2c Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Thu, 18 Aug 2022 20:32:10 +0800 Subject: [PATCH 12/13] YARN-11250. Fix CodeStyle. --- .../ZKFederationStateStoreOpDurations.java | 26 +++++++++---------- .../impl/ZookeeperFederationStateStore.java | 26 +++++++++---------- 2 files changed, 26 insertions(+), 26 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java index 47fbf2f4ee630..6ce5e2ef46185 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZKFederationStateStoreOpDurations.java @@ -101,55 +101,55 @@ public synchronized void getMetrics(MetricsCollector collector, boolean all) { registry.snapshot(collector.addRecord(registry.info()), all); } - public void addAppHomeSubClusterDuration(long endTime, long startTime) { + public void addAppHomeSubClusterDuration(long startTime, long endTime) { addAppHomeSubCluster.add(endTime - startTime); } - public void addUpdateAppHomeSubClusterDuration(long endTime, long startTime) { + public void addUpdateAppHomeSubClusterDuration(long startTime, long endTime) { updateAppHomeSubCluster.add(endTime - startTime); } - public void addGetAppHomeSubClusterDuration(long endTime, long startTime) { + public void addGetAppHomeSubClusterDuration(long startTime, long endTime) { getAppHomeSubCluster.add(endTime - startTime); } - public void addGetAppsHomeSubClusterDuration(long endTime, long startTime) { + public void addGetAppsHomeSubClusterDuration(long startTime, long endTime) { getAppsHomeSubCluster.add(endTime - startTime); } - public void addDeleteAppHomeSubClusterDuration(long endTime, long startTime) { + public void addDeleteAppHomeSubClusterDuration(long startTime, long endTime) { deleteAppHomeSubCluster.add(endTime - startTime); } - public void addRegisterSubClusterDuration(long endTime, long startTime) { + public void addRegisterSubClusterDuration(long startTime, long endTime) { registerSubCluster.add(endTime - startTime); } - public void addDeregisterSubClusterDuration(long endTime, long startTime) { + public void addDeregisterSubClusterDuration(long startTime, long endTime) { deregisterSubCluster.add(endTime - startTime); } - public void addSubClusterHeartbeatDuration(long endTime, long startTime) { + public void addSubClusterHeartbeatDuration(long startTime, long endTime) { subClusterHeartbeat.add(endTime - startTime); } - public void addGetSubClusterDuration(long endTime, long startTime) { + public void addGetSubClusterDuration(long startTime, long endTime) { getSubCluster.add(endTime - startTime); } - public void addGetSubClustersDuration(long endTime, long startTime) { + public void addGetSubClustersDuration(long startTime, long endTime) { getSubClusters.add(endTime - startTime); } - public void addGetPolicyConfigurationDuration(long endTime, long startTime) { + public void addGetPolicyConfigurationDuration(long startTime, long endTime) { getPolicyConfiguration.add(endTime - startTime); } - public void addSetPolicyConfigurationDuration(long endTime, long startTime) { + public void addSetPolicyConfigurationDuration(long startTime, long endTime) { setPolicyConfiguration.add(endTime - startTime); } - public void addGetPoliciesConfigurationsDuration(long endTime, long startTime) { + public void addGetPoliciesConfigurationsDuration(long startTime, long endTime) { getPoliciesConfigurations.add(endTime - startTime); } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java index 639eddc945420..f394655de8a60 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/federation/store/impl/ZookeeperFederationStateStore.java @@ -193,7 +193,7 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } long end = clock.getTime(); - opDurations.addAppHomeSubClusterDuration(end, start); + opDurations.addAppHomeSubClusterDuration(start, end); return AddApplicationHomeSubClusterResponse .newInstance(homeSubCluster); } @@ -218,7 +218,7 @@ public AddApplicationHomeSubClusterResponse addApplicationHomeSubCluster( putApp(appId, newSubClusterId, true); long end = clock.getTime(); - opDurations.addUpdateAppHomeSubClusterDuration(end, start); + opDurations.addUpdateAppHomeSubClusterDuration(start, end); return UpdateApplicationHomeSubClusterResponse.newInstance(); } @@ -235,7 +235,7 @@ public GetApplicationHomeSubClusterResponse getApplicationHomeSubCluster( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } long end = clock.getTime(); - opDurations.addGetAppHomeSubClusterDuration(end, start); + opDurations.addGetAppHomeSubClusterDuration(start, end); return GetApplicationHomeSubClusterResponse.newInstance(appId, homeSubCluster); } @@ -258,7 +258,7 @@ public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } long end = clock.getTime(); - opDurations.addGetAppsHomeSubClusterDuration(end, start); + opDurations.addGetAppsHomeSubClusterDuration(start, end); return GetApplicationsHomeSubClusterResponse.newInstance(result); } @@ -291,7 +291,7 @@ public GetApplicationsHomeSubClusterResponse getApplicationsHomeSubCluster( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } long end = clock.getTime(); - opDurations.addDeleteAppHomeSubClusterDuration(end, start); + opDurations.addDeleteAppHomeSubClusterDuration(start, end); return DeleteApplicationHomeSubClusterResponse.newInstance(); } @@ -314,7 +314,7 @@ public SubClusterRegisterResponse registerSubCluster( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } long end = clock.getTime(); - opDurations.addRegisterSubClusterDuration(end, start); + opDurations.addRegisterSubClusterDuration(start, end); return SubClusterRegisterResponse.newInstance(); } @@ -336,7 +336,7 @@ public SubClusterDeregisterResponse deregisterSubCluster( putSubclusterInfo(subClusterId, subClusterInfo, true); } long end = clock.getTime(); - opDurations.addDeregisterSubClusterDuration(end, start); + opDurations.addDeregisterSubClusterDuration(start, end); return SubClusterDeregisterResponse.newInstance(); } @@ -361,7 +361,7 @@ public SubClusterHeartbeatResponse subClusterHeartbeat( putSubclusterInfo(subClusterId, subClusterInfo, true); long end = clock.getTime(); - opDurations.addSubClusterHeartbeatDuration(clock.getTime(), start); + opDurations.addSubClusterHeartbeatDuration(start, end); return SubClusterHeartbeatResponse.newInstance(); } @@ -383,7 +383,7 @@ public GetSubClusterInfoResponse getSubCluster( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } long end = clock.getTime(); - opDurations.addGetSubClusterDuration(end, start); + opDurations.addGetSubClusterDuration(start, end); return GetSubClusterInfoResponse.newInstance(subClusterInfo); } @@ -407,7 +407,7 @@ public GetSubClustersInfoResponse getSubClusters( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } long end = clock.getTime(); - opDurations.addGetSubClustersDuration(end, start); + opDurations.addGetSubClustersDuration(start, end); return GetSubClustersInfoResponse.newInstance(result); } @@ -431,7 +431,7 @@ public GetSubClusterPolicyConfigurationResponse getPolicyConfiguration( return null; } long end = clock.getTime(); - opDurations.addGetPolicyConfigurationDuration(end, start); + opDurations.addGetPolicyConfigurationDuration(start, end); return GetSubClusterPolicyConfigurationResponse .newInstance(policy); } @@ -451,7 +451,7 @@ public SetSubClusterPolicyConfigurationResponse setPolicyConfiguration( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } long end = clock.getTime(); - opDurations.addSetPolicyConfigurationDuration(end, start); + opDurations.addSetPolicyConfigurationDuration(start, end); return SetSubClusterPolicyConfigurationResponse.newInstance(); } @@ -475,7 +475,7 @@ public GetSubClusterPoliciesConfigurationsResponse getPoliciesConfigurations( FederationStateStoreUtils.logAndThrowStoreException(LOG, errMsg); } long end = clock.getTime(); - opDurations.addGetPoliciesConfigurationsDuration(end, start); + opDurations.addGetPoliciesConfigurationsDuration(start, end); return GetSubClusterPoliciesConfigurationsResponse.newInstance(result); } From 15d0370f523e0508a4e5af9c6257f68dff1a9ab6 Mon Sep 17 00:00:00 2001 From: slfan1989 Date: Thu, 18 Aug 2022 20:33:51 +0800 Subject: [PATCH 13/13] YARN-11250. Fix CodeStyle. --- .../TestZookeeperFederationStateStore.java | 26 +++++++++---------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java index 68050ef2a04cf..584f3355ff5d6 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/test/java/org/apache/hadoop/yarn/server/federation/store/impl/TestZookeeperFederationStateStore.java @@ -102,19 +102,19 @@ public void testMetricsInited() throws Exception { long start = Time.now(); long end = start + anyDuration; - zkStateStoreOpDurations.addAppHomeSubClusterDuration(end, start); - zkStateStoreOpDurations.addUpdateAppHomeSubClusterDuration(end, start); - zkStateStoreOpDurations.addGetAppHomeSubClusterDuration(end, start); - zkStateStoreOpDurations.addGetAppsHomeSubClusterDuration(end, start); - zkStateStoreOpDurations.addDeleteAppHomeSubClusterDuration(end, start); - zkStateStoreOpDurations.addRegisterSubClusterDuration(end, start); - zkStateStoreOpDurations.addDeregisterSubClusterDuration(end, start); - zkStateStoreOpDurations.addSubClusterHeartbeatDuration(end, start); - zkStateStoreOpDurations.addGetSubClusterDuration(end, start); - zkStateStoreOpDurations.addGetSubClustersDuration(end, start); - zkStateStoreOpDurations.addGetPolicyConfigurationDuration(end, start); - zkStateStoreOpDurations.addSetPolicyConfigurationDuration(end, start); - zkStateStoreOpDurations.addGetPoliciesConfigurationsDuration(end, start); + zkStateStoreOpDurations.addAppHomeSubClusterDuration(start, end); + zkStateStoreOpDurations.addUpdateAppHomeSubClusterDuration(start, end); + zkStateStoreOpDurations.addGetAppHomeSubClusterDuration(start, end); + zkStateStoreOpDurations.addGetAppsHomeSubClusterDuration(start, end); + zkStateStoreOpDurations.addDeleteAppHomeSubClusterDuration(start, end); + zkStateStoreOpDurations.addRegisterSubClusterDuration(start, end); + zkStateStoreOpDurations.addDeregisterSubClusterDuration(start, end); + zkStateStoreOpDurations.addSubClusterHeartbeatDuration(start, end); + zkStateStoreOpDurations.addGetSubClusterDuration(start, end); + zkStateStoreOpDurations.addGetSubClustersDuration(start, end); + zkStateStoreOpDurations.addGetPolicyConfigurationDuration(start, end); + zkStateStoreOpDurations.addSetPolicyConfigurationDuration(start, end); + zkStateStoreOpDurations.addGetPoliciesConfigurationsDuration(start, end); zkStateStoreOpDurations.getMetrics(collector, true); assertEquals("Incorrect number of perf metrics", 1, collector.getRecords().size());