diff --git hadoop-yarn-project/hadoop-yarn/bin/yarn hadoop-yarn-project/hadoop-yarn/bin/yarn
index 128ab91..915e162 100644
--- hadoop-yarn-project/hadoop-yarn/bin/yarn
+++ hadoop-yarn-project/hadoop-yarn/bin/yarn
@@ -34,6 +34,7 @@ function hadoop_usage
echo " resourcemanager -format-state-store deletes the RMStateStore"
echo " rmadmin admin tools"
echo " sharedcachemanager run the SharedCacheManager daemon"
+ echo " scmadmin SharedCacheManager admin tools"
echo " timelineserver run the timeline server"
echo " version print the version"
echo " or"
@@ -144,6 +145,10 @@ case "${COMMAND}" in
CLASS='org.apache.hadoop.yarn.server.sharedcachemanager.SharedCacheManager'
YARN_OPTS="$YARN_OPTS $YARN_SHAREDCACHEMANAGER_OPTS"
;;
+ scmadmin)
+ CLASS='org.apache.hadoop.yarn.client.SCMAdmin'
+ YARN_OPTS="$YARN_OPTS $YARN_CLIENT_OPTS"
+ ;;
version)
CLASS=org.apache.hadoop.util.VersionInfo
YARN_OPTS="${YARN_OPTS} ${YARN_CLIENT_OPTS}"
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/pom.xml hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/pom.xml
index 5e2278d..600bb54 100644
--- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/pom.xml
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/pom.xml
@@ -97,6 +97,7 @@
application_history_client.proto
server/application_history_server.proto
client_SCM_protocol.proto
+ SCM_Admin_protocol.proto
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/SCMAdminProtocol.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/SCMAdminProtocol.java
new file mode 100644
index 0000000..1bc618d
--- /dev/null
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/SCMAdminProtocol.java
@@ -0,0 +1,51 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.api;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience.Public;
+import org.apache.hadoop.classification.InterfaceStability.Stable;
+import org.apache.hadoop.yarn.api.protocolrecords.RunSharedCacheCleanerTaskRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.RunSharedCacheCleanerTaskResponse;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+
+/**
+ *
+ * The protocol between administrators and the SharedCacheManager
+ *
+ */
+@Public
+@Stable
+public interface SCMAdminProtocol {
+ /**
+ *
+ * The method used by administrators to ask SCM to run cleaner task right away
+ *
+ *
+ * @param request request SharedCacheManager to run a cleaner task
+ * @return SharedCacheManager returns an empty response
+ * on success and throws an exception on rejecting the request
+ * @throws YarnException
+ * @throws IOException
+ */
+ public RunSharedCacheCleanerTaskResponse runCleanerTask(
+ RunSharedCacheCleanerTaskRequest request) throws YarnException, IOException;
+
+}
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/SCMAdminProtocolPB.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/SCMAdminProtocolPB.java
new file mode 100644
index 0000000..1883c16c
--- /dev/null
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/SCMAdminProtocolPB.java
@@ -0,0 +1,27 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.yarn.api;
+
+import org.apache.hadoop.ipc.ProtocolInfo;
+import org.apache.hadoop.yarn.proto.SCMAdminProtocol.SCMAdminProtocolService;
+
+@ProtocolInfo(protocolName = "org.apache.hadoop.yarn.api.SCMAdminProtocolPB",
+ protocolVersion = 1)
+public interface SCMAdminProtocolPB extends
+ SCMAdminProtocolService.BlockingInterface {
+}
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/RunSharedCacheCleanerTaskRequest.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/RunSharedCacheCleanerTaskRequest.java
new file mode 100644
index 0000000..7407b28
--- /dev/null
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/RunSharedCacheCleanerTaskRequest.java
@@ -0,0 +1,37 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.api.protocolrecords;
+
+import org.apache.hadoop.classification.InterfaceAudience.Public;
+import org.apache.hadoop.classification.InterfaceStability.Stable;
+
+/**
+ *
+ * The request from admin to ask the SharedCacheManager to run
+ * cleaner service right away.
+ *
+ *
+ *
+ * Currently, this is empty.
+ *
+ */
+@Public
+@Stable
+public abstract class RunSharedCacheCleanerTaskRequest {
+}
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/RunSharedCacheCleanerTaskResponse.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/RunSharedCacheCleanerTaskResponse.java
new file mode 100644
index 0000000..e68baa7
--- /dev/null
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/RunSharedCacheCleanerTaskResponse.java
@@ -0,0 +1,58 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.api.protocolrecords;
+
+import org.apache.hadoop.classification.InterfaceAudience.Public;
+import org.apache.hadoop.classification.InterfaceStability.Stable;
+
+/**
+ *
+ * The response to admin from the SharedCacheManager when
+ * is asked to run the cleaner service.
+ *
+ *
+ *
+ * Currently, this is empty.
+ *
+ */
+@Public
+@Stable
+public abstract class RunSharedCacheCleanerTaskResponse {
+
+ /**
+ * Get whether or not the shared cache manager has accepted the request.
+ * Shared cache manager will reject the request if there is an ongoing task
+ *
+ * @return boolean True if the resource has been accepted, false otherwise.
+ */
+ @Public
+ @Stable
+ public abstract boolean getAccepted();
+
+ /**
+ * Set whether or not the shared cache manager has accepted the request
+ * Shared cache manager will reject the request if there is an ongoing task
+ *
+ * @param b True if the resource has been accepted, false otherwise.
+ */
+ @Public
+ @Stable
+ public abstract void setAccepted(boolean b);
+
+}
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
index a164d46..eb795ad 100644
--- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
@@ -1313,6 +1313,17 @@
+ "client.thread-count";
public static final int DEFAULT_SCM_CLIENT_THREAD_COUNT = 50;
+ /** The address of the SCM admin interface. */
+ public static final String SCM_ADMIN_ADDRESS = SCM_PREFIX + "admin.address";
+ public static final int DEFAULT_SCM_ADMIN_PORT = 8047;
+ public static final String DEFAULT_SCM_ADMIN_ADDRESS = "0.0.0.0:"
+ + DEFAULT_SCM_ADMIN_PORT;
+
+ /** Number of threads used to handle SCM admin interface. */
+ public static final String SCM_ADMIN_CLIENT_THREAD_COUNT = SCM_PREFIX
+ + "admin.client.thread-count";
+ public static final int DEFAULT_SCM_ADMIN_CLIENT_THREAD_COUNT = 1;
+
////////////////////////////////
// Other Configs
////////////////////////////////
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/SCM_Admin_protocol.proto hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/SCM_Admin_protocol.proto
new file mode 100644
index 0000000..4e46c57
--- /dev/null
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/SCM_Admin_protocol.proto
@@ -0,0 +1,29 @@
+/**
+ * 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.
+ */
+
+option java_package = "org.apache.hadoop.yarn.proto";
+option java_outer_classname = "SCMAdminProtocol";
+option java_generic_services = true;
+option java_generate_equals_and_hash = true;
+package hadoop.yarn;
+
+import "yarn_service_protos.proto";
+
+service SCMAdminProtocolService {
+ rpc runCleanerTask (RunSharedCacheCleanerTaskRequestProto) returns (RunSharedCacheCleanerTaskResponseProto);
+}
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
index 51d58a7..56d66a5 100644
--- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
@@ -307,3 +307,14 @@ message ReleaseSharedCacheResourceRequestProto {
message ReleaseSharedCacheResourceResponseProto {
}
+
+//////////////////////////////////////////////////////
+/////// SCM_Admin_Protocol //////////////////////////
+//////////////////////////////////////////////////////
+
+message RunSharedCacheCleanerTaskRequestProto {
+}
+
+message RunSharedCacheCleanerTaskResponseProto {
+ optional bool accepted = 1;
+}
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/SCMAdmin.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/SCMAdmin.java
new file mode 100644
index 0000000..a743e11
--- /dev/null
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/SCMAdmin.java
@@ -0,0 +1,207 @@
+/**
+ * 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.client;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.security.PrivilegedAction;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.ipc.RemoteException;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.hadoop.yarn.api.SCMAdminProtocol;
+import org.apache.hadoop.yarn.api.protocolrecords.RunSharedCacheCleanerTaskRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.RunSharedCacheCleanerTaskResponse;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.factories.RecordFactory;
+import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
+import org.apache.hadoop.yarn.ipc.YarnRPC;
+
+public class SCMAdmin extends Configured implements Tool {
+
+ private final RecordFactory recordFactory =
+ RecordFactoryProvider.getRecordFactory(null);
+
+ public SCMAdmin() {
+ super();
+ }
+
+ public SCMAdmin(Configuration conf) {
+ super(conf);
+ }
+
+ private static void printHelp(String cmd) {
+ String summary = "scmadmin is the command to execute shared cache manager" +
+ "administrative commands.\n" +
+ "The full syntax is: \n\n" +
+ "hadoop scmadmin" +
+ " [-runCleanerTask]" +
+ " [-help [cmd]]\n";
+
+ String runCleanerTask =
+ "-runCleanerTask: Run cleaner task right away.\n";
+
+ String help = "-help [cmd]: \tDisplays help for the given command or all commands if none\n" +
+ "\t\tis specified.\n";
+
+ if ("runCleanerTask".equals(cmd)) {
+ System.out.println(runCleanerTask);
+ } else if ("help".equals(cmd)) {
+ System.out.println(help);
+ } else {
+ System.out.println(summary);
+ System.out.println(runCleanerTask);
+ System.out.println(help);
+ System.out.println();
+ ToolRunner.printGenericCommandUsage(System.out);
+ }
+ }
+
+ /**
+ * Displays format of commands.
+ * @param cmd The command that is being executed.
+ */
+ private static void printUsage(String cmd) {
+ if ("-runCleanerTask".equals(cmd)) {
+ System.err.println("Usage: yarn scmadmin" + " [-runCleanerTask]");
+ } else {
+ System.err.println("Usage: yarn scmadmin");
+ System.err.println(" [-runCleanerTask]");
+ System.err.println(" [-help [cmd]]");
+ System.err.println();
+ ToolRunner.printGenericCommandUsage(System.err);
+ }
+ }
+
+ private static UserGroupInformation getUGI(Configuration conf
+ ) throws IOException {
+ return UserGroupInformation.getCurrentUser();
+ }
+
+ private SCMAdminProtocol createSCMAdminProtocol() throws IOException {
+ // Get the current configuration
+ final YarnConfiguration conf = new YarnConfiguration(getConf());
+
+ // Create the admin client
+ final InetSocketAddress addr = conf.getSocketAddr(
+ YarnConfiguration.SCM_ADMIN_ADDRESS,
+ YarnConfiguration.DEFAULT_SCM_ADMIN_ADDRESS,
+ YarnConfiguration.DEFAULT_SCM_ADMIN_PORT);
+ final YarnRPC rpc = YarnRPC.create(conf);
+
+ SCMAdminProtocol scmAdminProtocol =
+ getUGI(conf).doAs(new PrivilegedAction() {
+ @Override
+ public SCMAdminProtocol run() {
+ return (SCMAdminProtocol) rpc.getProxy(SCMAdminProtocol.class,
+ addr, conf);
+ }
+ });
+
+ return scmAdminProtocol;
+ }
+
+ private int runCleanerTask() throws YarnException, IOException {
+ // run cleaner task right away
+ SCMAdminProtocol scmAdminProtocol = createSCMAdminProtocol();
+ RunSharedCacheCleanerTaskRequest request =
+ recordFactory.newRecordInstance(RunSharedCacheCleanerTaskRequest.class);
+ RunSharedCacheCleanerTaskResponse response =
+ scmAdminProtocol.runCleanerTask(request);
+ if (response.getAccepted()) {
+ System.out.println("request accepted by shared cache manager");
+ return 0;
+ } else {
+ System.out.println("request rejected by shared cache manager");
+ return 1;
+ }
+ }
+
+ @Override
+ public int run(String[] args) throws Exception {
+ if (args.length < 1) {
+ printUsage("");
+ return -1;
+ }
+
+ int exitCode = -1;
+ int i = 0;
+ String cmd = args[i++];
+ //
+ // verify that we have enough command line parameters
+ //
+ if ("-runCleanerTask".equals(cmd)) {
+ if (args.length != 1) {
+ printUsage(cmd);
+ return exitCode;
+ }
+ }
+
+ exitCode = 0;
+ try {
+ if ("-runCleanerTask".equals(cmd)) {
+ exitCode = runCleanerTask();
+ } else if ("-help".equals(cmd)) {
+ if (i < args.length) {
+ printUsage(args[i]);
+ } else {
+ printHelp("");
+ }
+ } else {
+ exitCode = -1;
+ System.err.println(cmd.substring(1) + ": Unknown command");
+ printUsage("");
+ printUsage("");
+ }
+
+ } catch (IllegalArgumentException arge) {
+ exitCode = -1;
+ System.err.println(cmd.substring(1) + ": " + arge.getLocalizedMessage());
+ printUsage(cmd);
+ } catch (RemoteException e) {
+ //
+ // This is a error returned by hadoop server. Print
+ // out the first line of the error mesage, ignore the stack trace.
+ exitCode = -1;
+ try {
+ String[] content;
+ content = e.getLocalizedMessage().split("\n");
+ System.err.println(cmd.substring(1) + ": "
+ + content[0]);
+ } catch (Exception ex) {
+ System.err.println(cmd.substring(1) + ": "
+ + ex.getLocalizedMessage());
+ }
+ } catch (Exception e) {
+ exitCode = -1;
+ System.err.println(cmd.substring(1) + ": "
+ + e.getLocalizedMessage());
+ }
+ return exitCode;
+ }
+
+ public static void main(String[] args) throws Exception {
+ int result = ToolRunner.run(new SCMAdmin(), args);
+ System.exit(result);
+ }
+}
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/SCMAdminProtocolPBClientImpl.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/SCMAdminProtocolPBClientImpl.java
new file mode 100644
index 0000000..aabc2a3
--- /dev/null
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/SCMAdminProtocolPBClientImpl.java
@@ -0,0 +1,74 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.api.impl.pb.client;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.ProtobufRpcEngine;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.yarn.api.ClientSCMProtocolPB;
+import org.apache.hadoop.yarn.api.SCMAdminProtocol;
+import org.apache.hadoop.yarn.api.SCMAdminProtocolPB;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.RunSharedCacheCleanerTaskRequestPBImpl;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.RunSharedCacheCleanerTaskResponsePBImpl;
+import org.apache.hadoop.yarn.api.protocolrecords.RunSharedCacheCleanerTaskRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.RunSharedCacheCleanerTaskResponse;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.ipc.RPCUtil;
+import org.apache.hadoop.yarn.proto.YarnServiceProtos;
+
+import com.google.protobuf.ServiceException;
+
+public class SCMAdminProtocolPBClientImpl implements SCMAdminProtocol,
+ Closeable {
+
+ private SCMAdminProtocolPB proxy;
+
+ public SCMAdminProtocolPBClientImpl(long clientVersion,
+ InetSocketAddress addr, Configuration conf) throws IOException {
+ RPC.setProtocolEngine(conf, SCMAdminProtocolPB.class,
+ ProtobufRpcEngine.class);
+ proxy = RPC.getProxy(SCMAdminProtocolPB.class, clientVersion, addr, conf);
+ }
+
+ @Override
+ public void close() {
+ if (this.proxy != null) {
+ RPC.stopProxy(this.proxy);
+ }
+ }
+
+ @Override
+ public RunSharedCacheCleanerTaskResponse runCleanerTask(
+ RunSharedCacheCleanerTaskRequest request) throws YarnException,
+ IOException {
+ YarnServiceProtos.RunSharedCacheCleanerTaskRequestProto requestProto =
+ ((RunSharedCacheCleanerTaskRequestPBImpl) request).getProto();
+ try {
+ return new RunSharedCacheCleanerTaskResponsePBImpl(proxy.runCleanerTask(null,
+ requestProto));
+ } catch (ServiceException e) {
+ RPCUtil.unwrapAndThrowException(e);
+ return null;
+ }
+ }
+}
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/service/SCMAdminProtocolPBServiceImpl.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/service/SCMAdminProtocolPBServiceImpl.java
new file mode 100644
index 0000000..937e363
--- /dev/null
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/service/SCMAdminProtocolPBServiceImpl.java
@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.api.impl.pb.service;
+
+import java.io.IOException;
+
+import org.apache.hadoop.yarn.api.SCMAdminProtocol;
+import org.apache.hadoop.yarn.api.SCMAdminProtocolPB;
+import org.apache.hadoop.yarn.api.protocolrecords.RunSharedCacheCleanerTaskResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.RunSharedCacheCleanerTaskRequestPBImpl;
+import org.apache.hadoop.yarn.api.protocolrecords.impl.pb.RunSharedCacheCleanerTaskResponsePBImpl;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.proto.YarnServiceProtos.RunSharedCacheCleanerTaskRequestProto;
+import org.apache.hadoop.yarn.proto.YarnServiceProtos.RunSharedCacheCleanerTaskResponseProto;
+
+import com.google.protobuf.RpcController;
+import com.google.protobuf.ServiceException;
+
+public class SCMAdminProtocolPBServiceImpl implements SCMAdminProtocolPB {
+
+ private SCMAdminProtocol real;
+
+ public SCMAdminProtocolPBServiceImpl(SCMAdminProtocol impl) {
+ this.real = impl;
+ }
+
+ @Override
+ public RunSharedCacheCleanerTaskResponseProto runCleanerTask(RpcController controller,
+ RunSharedCacheCleanerTaskRequestProto proto) throws ServiceException {
+ RunSharedCacheCleanerTaskRequestPBImpl request =
+ new RunSharedCacheCleanerTaskRequestPBImpl(proto);
+ try {
+ RunSharedCacheCleanerTaskResponse response = real.runCleanerTask(request);
+ return ((RunSharedCacheCleanerTaskResponsePBImpl) response).getProto();
+ } catch (YarnException e) {
+ throw new ServiceException(e);
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ }
+ }
+}
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/RunSharedCacheCleanerTaskRequestPBImpl.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/RunSharedCacheCleanerTaskRequestPBImpl.java
new file mode 100644
index 0000000..d432332
--- /dev/null
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/RunSharedCacheCleanerTaskRequestPBImpl.java
@@ -0,0 +1,53 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.api.protocolrecords.impl.pb;
+
+import org.apache.hadoop.yarn.api.protocolrecords.RunSharedCacheCleanerTaskRequest;
+import org.apache.hadoop.yarn.proto.YarnServiceProtos.RunSharedCacheCleanerTaskRequestProto;
+
+public class RunSharedCacheCleanerTaskRequestPBImpl extends
+ RunSharedCacheCleanerTaskRequest {
+ RunSharedCacheCleanerTaskRequestProto proto =
+ RunSharedCacheCleanerTaskRequestProto.getDefaultInstance();
+ RunSharedCacheCleanerTaskRequestProto.Builder builder = null;
+ boolean viaProto = false;
+
+ public RunSharedCacheCleanerTaskRequestPBImpl() {
+ builder = RunSharedCacheCleanerTaskRequestProto.newBuilder();
+ }
+
+ public RunSharedCacheCleanerTaskRequestPBImpl(
+ RunSharedCacheCleanerTaskRequestProto proto) {
+ this.proto = proto;
+ viaProto = true;
+ }
+
+ public RunSharedCacheCleanerTaskRequestProto getProto() {
+ proto = viaProto ? proto : builder.build();
+ viaProto = true;
+ return proto;
+ }
+
+ private void maybeInitBuilder() {
+ if (viaProto || builder == null) {
+ builder = RunSharedCacheCleanerTaskRequestProto.newBuilder(proto);
+ }
+ viaProto = false;
+ }
+}
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/RunSharedCacheCleanerTaskResponsePBImpl.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/RunSharedCacheCleanerTaskResponsePBImpl.java
new file mode 100644
index 0000000..c7da352
--- /dev/null
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/RunSharedCacheCleanerTaskResponsePBImpl.java
@@ -0,0 +1,66 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.api.protocolrecords.impl.pb;
+
+import org.apache.hadoop.yarn.api.protocolrecords.RunSharedCacheCleanerTaskResponse;
+import org.apache.hadoop.yarn.proto.YarnServiceProtos.RunSharedCacheCleanerTaskResponseProto;
+import org.apache.hadoop.yarn.proto.YarnServiceProtos.RunSharedCacheCleanerTaskResponseProtoOrBuilder;
+
+public class RunSharedCacheCleanerTaskResponsePBImpl extends
+ RunSharedCacheCleanerTaskResponse {
+ RunSharedCacheCleanerTaskResponseProto proto =
+ RunSharedCacheCleanerTaskResponseProto.getDefaultInstance();
+ RunSharedCacheCleanerTaskResponseProto.Builder builder = null;
+ boolean viaProto = false;
+
+ public RunSharedCacheCleanerTaskResponsePBImpl() {
+ builder = RunSharedCacheCleanerTaskResponseProto.newBuilder();
+ }
+
+ public RunSharedCacheCleanerTaskResponsePBImpl(
+ RunSharedCacheCleanerTaskResponseProto proto) {
+ this.proto = proto;
+ viaProto = true;
+ }
+
+ @Override
+ public boolean getAccepted() {
+ RunSharedCacheCleanerTaskResponseProtoOrBuilder p = viaProto ? proto : builder;
+ return (p.hasAccepted()) ? p.getAccepted() : false;
+ }
+
+ @Override
+ public void setAccepted(boolean b) {
+ maybeInitBuilder();
+ builder.setAccepted(b);
+ }
+
+ public RunSharedCacheCleanerTaskResponseProto getProto() {
+ proto = viaProto ? proto : builder.build();
+ viaProto = true;
+ return proto;
+ }
+
+ private void maybeInitBuilder() {
+ if (viaProto || builder == null) {
+ builder = RunSharedCacheCleanerTaskResponseProto.newBuilder(proto);
+ }
+ viaProto = false;
+ }
+}
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
index 7fd4d61..379e8d8 100644
--- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml
@@ -1403,6 +1403,18 @@
yarn.sharedcache.manager.client.thread-count
50
+
+
+ The address of the admin interface in the SCM (shared cache manager)
+ yarn.sharedcache.manager.admin.address
+ 0.0.0.0:8047
+
+
+
+ The number of threads used to handle SCM admin interface (1 by default)
+ yarn.sharedcache.manager.admin.thread-count
+ 1
+
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/SCMAdminProtocolService.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/SCMAdminProtocolService.java
new file mode 100644
index 0000000..1838932
--- /dev/null
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/SCMAdminProtocolService.java
@@ -0,0 +1,141 @@
+/**
+ * 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.sharedcachemanager;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.Server;
+import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.authorize.AccessControlList;
+import org.apache.hadoop.service.AbstractService;
+import org.apache.hadoop.yarn.api.SCMAdminProtocol;
+import org.apache.hadoop.yarn.api.protocolrecords.RunSharedCacheCleanerTaskRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.RunSharedCacheCleanerTaskResponse;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.factories.RecordFactory;
+import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
+import org.apache.hadoop.yarn.ipc.RPCUtil;
+import org.apache.hadoop.yarn.ipc.YarnRPC;
+
+/**
+ * This service handles all SCMAdminProtocol rpc calls from administrators
+ * to the shared cache manager.
+ */
+public class SCMAdminProtocolService extends AbstractService implements
+ SCMAdminProtocol {
+
+ private static final Log LOG = LogFactory.getLog(SCMAdminProtocolService.class);
+
+ private final RecordFactory recordFactory = RecordFactoryProvider
+ .getRecordFactory(null);
+
+ private Server server;
+ InetSocketAddress clientBindAddress;
+ private final CleanerService cleanerService;
+ private AccessControlList adminAcl;
+
+ public SCMAdminProtocolService(CleanerService cleanerService) {
+ super(SCMAdminProtocolService.class.getName());
+ this.cleanerService = cleanerService;
+ }
+
+ @Override
+ protected void serviceInit(Configuration conf) throws Exception {
+ this.clientBindAddress = getBindAddress(conf);
+ adminAcl = new AccessControlList(conf.get(
+ YarnConfiguration.YARN_ADMIN_ACL,
+ YarnConfiguration.DEFAULT_YARN_ADMIN_ACL));
+
+
+ super.serviceInit(conf);
+ }
+
+ InetSocketAddress getBindAddress(Configuration conf) {
+ return conf.getSocketAddr(YarnConfiguration.SCM_ADMIN_ADDRESS,
+ YarnConfiguration.DEFAULT_SCM_ADMIN_ADDRESS,
+ YarnConfiguration.DEFAULT_SCM_ADMIN_PORT);
+ }
+
+ @Override
+ protected void serviceStart() throws Exception {
+ Configuration conf = getConfig();
+ YarnRPC rpc = YarnRPC.create(conf);
+ this.server =
+ rpc.getServer(SCMAdminProtocol.class, this,
+ clientBindAddress,
+ conf, null, // Secret manager null for now (security not supported)
+ conf.getInt(YarnConfiguration.SCM_ADMIN_CLIENT_THREAD_COUNT,
+ YarnConfiguration.DEFAULT_SCM_ADMIN_CLIENT_THREAD_COUNT));
+
+ // TODO: Enable service authorization
+
+ this.server.start();
+ clientBindAddress =
+ conf.updateConnectAddr(YarnConfiguration.SCM_ADMIN_ADDRESS,
+ server.getListenerAddress());
+
+ super.serviceStart();
+ }
+
+ @Override
+ protected void serviceStop() throws Exception {
+ if (this.server != null) {
+ this.server.stop();
+ }
+
+ super.serviceStop();
+ }
+
+ private void checkAcls(String method) throws YarnException {
+ UserGroupInformation user;
+ try {
+ user = UserGroupInformation.getCurrentUser();
+ } catch (IOException ioe) {
+ LOG.warn("Couldn't get current user", ioe);
+ throw RPCUtil.getRemoteException(ioe);
+ }
+
+ if (!adminAcl.isUserAllowed(user)) {
+ LOG.warn("User " + user.getShortUserName() + " doesn't have permission" +
+ " to call '" + method + "'");
+
+ throw RPCUtil.getRemoteException(
+ new AccessControlException("User " + user.getShortUserName() +
+ " doesn't have permission" + " to call '" + method + "'"));
+ }
+ LOG.info("SCM Admin: " + method + " invoked by user " +
+ user.getShortUserName());
+ }
+
+ @Override
+ public RunSharedCacheCleanerTaskResponse runCleanerTask(
+ RunSharedCacheCleanerTaskRequest request) throws YarnException {
+ checkAcls("runCleanerTask");
+ RunSharedCacheCleanerTaskResponse response =
+ recordFactory.newRecordInstance(RunSharedCacheCleanerTaskResponse.class);
+ response.setAccepted(this.cleanerService.runCleanerTask());
+ return response;
+ }
+}
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/SharedCacheManager.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/SharedCacheManager.java
index 42468af..426e625 100644
--- hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/SharedCacheManager.java
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/main/java/org/apache/hadoop/yarn/server/sharedcachemanager/SharedCacheManager.java
@@ -87,6 +87,9 @@ protected synchronized void serviceInit(Configuration conf) throws Exception {
ClientProtocolService cps = createClientProtocolService(store);
addService(cps);
+ SCMAdminProtocolService saps = createSCMAdminProtocolService(cs);
+ addService(saps);
+
} catch (IOException e) {
LOG.error("Encountered unexpected exception while initializing the shared cache manager",
e);
@@ -208,6 +211,11 @@ private ClientProtocolService createClientProtocolService(SCMStore store) {
return new ClientProtocolService(store);
}
+ private SCMAdminProtocolService createSCMAdminProtocolService(
+ CleanerService cleanerService) {
+ return new SCMAdminProtocolService(cleanerService);
+ }
+
@Override
protected synchronized void serviceStart() throws Exception {
// Start metrics
diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/test/java/org/apache/hadoop/yarn/server/sharedcachemanager/TestSCMAdminProtocolService.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/test/java/org/apache/hadoop/yarn/server/sharedcachemanager/TestSCMAdminProtocolService.java
new file mode 100644
index 0000000..d18ddea
--- /dev/null
+++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-sharedcachemanager/src/test/java/org/apache/hadoop/yarn/server/sharedcachemanager/TestSCMAdminProtocolService.java
@@ -0,0 +1,101 @@
+/**
+ * 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.sharedcachemanager;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.net.InetSocketAddress;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.yarn.api.SCMAdminProtocol;
+import org.apache.hadoop.yarn.api.protocolrecords.RunSharedCacheCleanerTaskRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.RunSharedCacheCleanerTaskResponse;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.factories.RecordFactory;
+import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
+import org.apache.hadoop.yarn.ipc.YarnRPC;
+import org.apache.hadoop.yarn.server.sharedcachemanager.store.InMemorySCMStore;
+import org.apache.hadoop.yarn.server.sharedcachemanager.store.SCMStore;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Basic unit tests for the SCM Admin Protocol Service.
+ */
+public class TestSCMAdminProtocolService {
+
+ static SCMAdminProtocolService service;
+ static SCMAdminProtocol SCMAdminProxy;
+ static SCMStore store;
+ static CleanerService cleaner;
+ private final RecordFactory recordFactory = RecordFactoryProvider
+ .getRecordFactory(null);
+
+ @Before
+ public void startUp() {
+ Configuration conf = new Configuration();
+ conf.set(YarnConfiguration.SCM_STORE_IMPL, InMemorySCMStore.class.getName());
+
+ cleaner = mock(CleanerService.class);
+
+ service = spy(new SCMAdminProtocolService(cleaner));
+ service.init(conf);
+ service.start();
+
+ YarnRPC rpc = YarnRPC.create(new Configuration());
+
+ InetSocketAddress scmAddress =
+ conf.getSocketAddr(YarnConfiguration.SCM_ADMIN_ADDRESS,
+ YarnConfiguration.DEFAULT_SCM_ADMIN_ADDRESS,
+ YarnConfiguration.DEFAULT_SCM_ADMIN_PORT);
+
+ SCMAdminProxy =
+ (SCMAdminProtocol) rpc.getProxy(SCMAdminProtocol.class, scmAddress,
+ conf);
+ }
+
+ @After
+ public void cleanUpTest() {
+ if (service != null) {
+ service.stop();
+ }
+
+ if (SCMAdminProxy != null) {
+ RPC.stopProxy(SCMAdminProxy);
+ }
+ }
+
+ @Test
+ public void testRunCleanerTask() throws Exception {
+ when(cleaner.runCleanerTask()).thenReturn(true);
+ RunSharedCacheCleanerTaskRequest request =
+ recordFactory.newRecordInstance(RunSharedCacheCleanerTaskRequest.class);
+ RunSharedCacheCleanerTaskResponse response = SCMAdminProxy.runCleanerTask(request);
+ Assert.assertTrue("cleaner task request isn't accepted", response.getAccepted());
+ verify(service, times(1)).runCleanerTask(any(RunSharedCacheCleanerTaskRequest.class));
+ }
+}