- Usage: hadoop balancer [-threshold <threshold>]
+ Usage: hadoop balancer [-policy <blockpool|datanode>] [-threshold <threshold>]
| COMMAND_OPTION | Description | |
|---|---|---|
-policy <blockpool|datanode> |
+ The balancing policy.
+ datanode: Cluster is balance if the disk usage of each datanode is balance.
+ blockpool: Cluster is balance if the disk usage of each block pool in each datanode is balance.
+ Note that blockpool is a condition stronger than datanode.
+ The default policy is datanode.
+ |
+ |
-threshold <threshold> |
- Percentage of disk capacity. This overwrites the default threshold. | +Percentage of disk capacity. This default threshold is 10%. |
param, to the IPC server running at
* address, returning the value. Throws exceptions if there are
* network problems or if the remote code threw an exception.
- * @deprecated Use {@link #call(RpcKind, Writable, ConnectionId)} instead
+ * @deprecated Use {@link #call(RpcPayloadHeader.RpcKind, Writable,
+ * ConnectionId)} instead
*/
@Deprecated
public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress address)
@@ -1025,7 +1027,8 @@ public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress address)
* the value.
* Throws exceptions if there are network problems or if the remote code
* threw an exception.
- * @deprecated Use {@link #call(RpcKind, Writable, ConnectionId)} instead
+ * @deprecated Use {@link #call(RpcPayloadHeader.RpcKind, Writable,
+ * ConnectionId)} instead
*/
@Deprecated
public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress addr,
@@ -1042,7 +1045,8 @@ public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress addr,
* timeout, returning the value.
* Throws exceptions if there are network problems or if the remote code
* threw an exception.
- * @deprecated Use {@link #call(RpcKind, Writable, ConnectionId)} instead
+ * @deprecated Use {@link #call(RpcPayloadHeader.RpcKind, Writable,
+ * ConnectionId)} instead
*/
@Deprecated
public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress addr,
@@ -1056,7 +1060,7 @@ public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress addr,
/**
- * Same as {@link #call(RpcKind, Writable, InetSocketAddress,
+ * Same as {@link #call(RpcPayloadHeader.RpcKind, Writable, InetSocketAddress,
* Class, UserGroupInformation, int, Configuration)}
* except that rpcKind is writable.
*/
@@ -1066,7 +1070,7 @@ public Writable call(Writable param, InetSocketAddress addr,
throws InterruptedException, IOException {
ConnectionId remoteId = ConnectionId.getConnectionId(addr, protocol,
ticket, rpcTimeout, conf);
- return call(RpcKind.RPC_WRITABLE, param, remoteId);
+ return call(RpcKind.RPC_BUILTIN, param, remoteId);
}
/**
@@ -1087,21 +1091,28 @@ public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress addr,
}
/**
- * Same as {link {@link #call(RpcKind, Writable, ConnectionId)}
- * except the rpcKind is RPC_WRITABLE
+ * Same as {link {@link #call(RpcPayloadHeader.RpcKind, Writable, ConnectionId)}
+ * except the rpcKind is RPC_BUILTIN
*/
public Writable call(Writable param, ConnectionId remoteId)
throws InterruptedException, IOException {
- return call(RpcKind.RPC_WRITABLE, param, remoteId);
+ return call(RpcKind.RPC_BUILTIN, param, remoteId);
}
- /** Make a call, passing param, to the IPC server defined by
- * remoteId, returning the value.
+ /**
+ * Make a call, passing rpcRequest, to the IPC server defined by
+ * remoteId, returning the rpc respond.
+ *
+ * @param rpcKind
+ * @param rpcRequest - contains serialized method and method parameters
+ * @param remoteId - the target rpc server
+ * @returns the rpc response
* Throws exceptions if there are network problems or if the remote code
- * threw an exception. */
- public Writable call(RpcKind rpcKind, Writable param, ConnectionId remoteId)
- throws InterruptedException, IOException {
- Call call = new Call(rpcKind, param);
+ * threw an exception.
+ */
+ public Writable call(RpcKind rpcKind, Writable rpcRequest,
+ ConnectionId remoteId) throws InterruptedException, IOException {
+ Call call = new Call(rpcKind, rpcRequest);
Connection connection = getConnection(remoteId, call);
connection.sendParam(call); // send the parameter
boolean interrupted = false;
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufHelper.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufHelper.java
new file mode 100644
index 0000000000..7f029618fa
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufHelper.java
@@ -0,0 +1,46 @@
+/**
+ * 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.ipc;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+
+import com.google.protobuf.ServiceException;
+
+/**
+ * Helper methods for protobuf related RPC implementation
+ */
+@InterfaceAudience.Private
+public class ProtobufHelper {
+ private ProtobufHelper() {
+ // Hidden constructor for class with only static helper methods
+ }
+
+ /**
+ * Return the RemoteException wrapped in ServiceException as cause.
+ * @param se ServiceException that wraps RemoteException
+ * @return RemoteException wrapped in ServiceException or
+ * a new IOException that wraps unexpected ServiceException.
+ */
+ public static IOException getRemoteException(ServiceException se) {
+ Throwable e = se.getCause();
+ return ((e instanceof RemoteException) ? (IOException) e :
+ new IOException(se));
+ }
+}
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
index dad94227ca..3db73859e5 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
@@ -37,6 +37,7 @@
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.ipc.RPC.RpcInvoker;
import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRpcExceptionProto;
import org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRpcRequestProto;
@@ -60,6 +61,12 @@
@InterfaceStability.Evolving
public class ProtobufRpcEngine implements RpcEngine {
private static final Log LOG = LogFactory.getLog(ProtobufRpcEngine.class);
+
+ static { // Register the rpcRequest deserializer for WritableRpcEngine
+ org.apache.hadoop.ipc.Server.registerProtocolEngine(
+ RpcKind.RPC_PROTOCOL_BUFFER, RpcRequestWritable.class,
+ new Server.ProtoBufRpcInvoker());
+ }
private static final ClientCache CLIENTS = new ClientCache();
@@ -75,10 +82,13 @@ public handlerCount determines
* the number of handler threads that will be used to process calls.
* If queueSizePerHandler or numReaders are not -1 they will be used instead of parameters
* from configuration. Otherwise the configuration will be picked up.
+ *
+ * If rpcRequestClass is null then the rpcRequestClass must have been
+ * registered via {@link #registerProtocolEngine(RpcPayloadHeader.RpcKind,
+ * Class, RPC.RpcInvoker)}
+ * This parameter has been retained for compatibility with existing tests
+ * and usage.
*/
@SuppressWarnings("unchecked")
- protected Server(String bindAddress, int port,
- Class paramClass, int handlerCount, int numReaders, int queueSizePerHandler,
- Configuration conf, String serverName, SecretManager secretManager)
+ protected Server(String bindAddress, int port,
+ Class rpcRequestClass, int handlerCount,
+ int numReaders, int queueSizePerHandler, Configuration conf,
+ String serverName, SecretManager secretManager)
throws IOException {
this.bindAddress = bindAddress;
this.conf = conf;
this.port = port;
- this.paramClass = paramClass;
+ this.rpcRequestClass = rpcRequestClass;
this.handlerCount = handlerCount;
this.socketSendBufferSize = 0;
if (queueSizePerHandler != -1) {
@@ -1797,17 +1879,17 @@ public synchronized InetSocketAddress getListenerAddress() {
/**
* Called for each call.
- * @deprecated Use {@link #call(String, Writable, long)} instead
+ * @deprecated Use {@link #call(RpcPayloadHeader.RpcKind, String,
+ * Writable, long)} instead
*/
@Deprecated
public Writable call(Writable param, long receiveTime) throws IOException {
- return call(null, param, receiveTime);
+ return call(RpcKind.RPC_BUILTIN, null, param, receiveTime);
}
/** Called for each call. */
- public abstract Writable call(String protocol,
- Writable param, long receiveTime)
- throws IOException;
+ public abstract Writable call(RpcKind rpcKind, String protocol,
+ Writable param, long receiveTime) throws IOException;
/**
* Authorize the incoming client connection.
@@ -1957,5 +2039,5 @@ private static int channelIO(ReadableByteChannel readCh,
int nBytes = initialRemaining - buf.remaining();
return (nBytes > 0) ? nBytes : ret;
- }
+ }
}
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java
index 6953814891..25f46f13e2 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java
@@ -18,7 +18,6 @@
package org.apache.hadoop.ipc;
-import java.lang.reflect.Field;
import java.lang.reflect.Proxy;
import java.lang.reflect.Method;
import java.lang.reflect.Array;
@@ -27,18 +26,14 @@
import java.net.InetSocketAddress;
import java.io.*;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
import java.io.Closeable;
-import java.util.Map;
-import java.util.HashMap;
import javax.net.SocketFactory;
import org.apache.commons.logging.*;
import org.apache.hadoop.io.*;
+import org.apache.hadoop.ipc.RPC.RpcInvoker;
import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.ipc.VersionedProtocol;
import org.apache.hadoop.security.UserGroupInformation;
@@ -53,36 +48,9 @@
public class WritableRpcEngine implements RpcEngine {
private static final Log LOG = LogFactory.getLog(RPC.class);
-
- /**
- * Get all superInterfaces that extend VersionedProtocol
- * @param childInterfaces
- * @return the super interfaces that extend VersionedProtocol
- */
- private static Class[] getSuperInterfaces(Class[] childInterfaces) {
- List> allInterfaces = new ArrayList>();
-
- for (Class childInterface : childInterfaces) {
- if (VersionedProtocol.class.isAssignableFrom(childInterface)) {
- allInterfaces.add(childInterface);
- allInterfaces.addAll(
- Arrays.asList(
- getSuperInterfaces(childInterface.getInterfaces())));
- } else {
- LOG.warn("Interface " + childInterface +
- " ignored because it does not extend VersionedProtocol");
- }
- }
- return (Class[]) allInterfaces.toArray(new Class[allInterfaces.size()]);
- }
-
- /**
- * Get all interfaces that the given protocol implements or extends
- * which are assignable from VersionedProtocol.
- */
- private static Class[] getProtocolInterfaces(Class protocol) {
- Class[] interfaces = protocol.getInterfaces();
- return getSuperInterfaces(interfaces);
+ static { // Register the rpcRequest deserializer for WritableRpcEngine
+ org.apache.hadoop.ipc.Server.registerProtocolEngine(RpcKind.RPC_WRITABLE,
+ Invocation.class, new Server.WritableRpcInvoker());
}
@@ -120,15 +88,7 @@ public Invocation(Method method, Object[] parameters) {
clientVersion = 0;
clientMethodsHash = 0;
} else {
- try {
- Field versionField = method.getDeclaringClass().getField("versionID");
- versionField.setAccessible(true);
- this.clientVersion = versionField.getLong(method.getDeclaringClass());
- } catch (NoSuchFieldException ex) {
- throw new RuntimeException(ex);
- } catch (IllegalAccessException ex) {
- throw new RuntimeException(ex);
- }
+ this.clientVersion = RPC.getProtocolVersion(method.getDeclaringClass());
this.clientMethodsHash = ProtocolSignature.getFingerprint(method
.getDeclaringClass().getMethods());
}
@@ -329,140 +289,25 @@ public RPC.Server getServer(Class protocolClass,
/** An RPC Server. */
public static class Server extends RPC.Server {
- private boolean verbose;
-
- /**
- * The key in Map
- */
- static class ProtoNameVer {
- final String protocol;
- final long version;
- ProtoNameVer(String protocol, long ver) {
- this.protocol = protocol;
- this.version = ver;
- }
- @Override
- public boolean equals(Object o) {
- if (o == null)
- return false;
- if (this == o)
- return true;
- if (! (o instanceof ProtoNameVer))
- return false;
- ProtoNameVer pv = (ProtoNameVer) o;
- return ((pv.protocol.equals(this.protocol)) &&
- (pv.version == this.version));
- }
- @Override
- public int hashCode() {
- return protocol.hashCode() * 37 + (int) version;
- }
- }
-
- /**
- * The value in map
- */
- static class ProtoClassProtoImpl {
- final Class protocolClass;
- final Object protocolImpl;
- ProtoClassProtoImpl(Class protocolClass, Object protocolImpl) {
- this.protocolClass = protocolClass;
- this.protocolImpl = protocolImpl;
- }
- }
-
- private Map protocolImplMap =
- new HashMap(10);
-
- // Register protocol and its impl for rpc calls
- private void registerProtocolAndImpl(Class protocolClass,
- Object protocolImpl) throws IOException {
- String protocolName = RPC.getProtocolName(protocolClass);
- VersionedProtocol vp = (VersionedProtocol) protocolImpl;
- long version;
- try {
- version = vp.getProtocolVersion(protocolName, 0);
- } catch (Exception ex) {
- LOG.warn("Protocol " + protocolClass +
- " NOT registered as getProtocolVersion throws exception ");
- return;
- }
- protocolImplMap.put(new ProtoNameVer(protocolName, version),
- new ProtoClassProtoImpl(protocolClass, protocolImpl));
- LOG.debug("Protocol Name = " + protocolName + " version=" + version +
- " ProtocolImpl=" + protocolImpl.getClass().getName() +
- " protocolClass=" + protocolClass.getName());
- }
-
- private static class VerProtocolImpl {
- final long version;
- final ProtoClassProtoImpl protocolTarget;
- VerProtocolImpl(long ver, ProtoClassProtoImpl protocolTarget) {
- this.version = ver;
- this.protocolTarget = protocolTarget;
- }
- }
-
-
- @SuppressWarnings("unused") // will be useful later.
- private VerProtocolImpl[] getSupportedProtocolVersions(
- String protocolName) {
- VerProtocolImpl[] resultk = new VerProtocolImpl[protocolImplMap.size()];
- int i = 0;
- for (Map.Entry pv :
- protocolImplMap.entrySet()) {
- if (pv.getKey().protocol.equals(protocolName)) {
- resultk[i++] =
- new VerProtocolImpl(pv.getKey().version, pv.getValue());
- }
- }
- if (i == 0) {
- return null;
- }
- VerProtocolImpl[] result = new VerProtocolImpl[i];
- System.arraycopy(resultk, 0, result, 0, i);
- return result;
- }
-
- private VerProtocolImpl getHighestSupportedProtocol(String protocolName) {
- Long highestVersion = 0L;
- ProtoClassProtoImpl highest = null;
- for (Map.Entry pv : protocolImplMap
- .entrySet()) {
- if (pv.getKey().protocol.equals(protocolName)) {
- if ((highest == null) || (pv.getKey().version > highestVersion)) {
- highest = pv.getValue();
- highestVersion = pv.getKey().version;
- }
- }
- }
- if (highest == null) {
- return null;
- }
- return new VerProtocolImpl(highestVersion, highest);
- }
-
-
- /** Construct an RPC server.
+ /**
+ * Construct an RPC server.
* @param instance the instance whose methods will be called
* @param conf the configuration to use
* @param bindAddress the address to bind on to listen for connection
* @param port the port to listen for connections on
*
- * @deprecated Use #Server(Class, Object, Configuration, String, int)
- *
+ * @deprecated Use #Server(Class, Object, Configuration, String, int)
*/
@Deprecated
public Server(Object instance, Configuration conf, String bindAddress,
- int port)
- throws IOException {
+ int port) throws IOException {
this(null, instance, conf, bindAddress, port);
}
/** Construct an RPC server.
- * @param protocol class
- * @param instance the instance whose methods will be called
+ * @param protocolClass class
+ * @param protocolImpl the instance whose methods will be called
* @param conf the configuration to use
* @param bindAddress the address to bind on to listen for connection
* @param port the port to listen for connections on
@@ -474,16 +319,8 @@ public Server(Class protocolClass, Object protocolImpl,
false, null);
}
- private static String classNameBase(String className) {
- String[] names = className.split("\\.", -1);
- if (names == null || names.length == 0) {
- return className;
- }
- return names[names.length-1];
- }
-
-
- /** Construct an RPC server.
+ /**
+ * Construct an RPC server.
* @param protocolImpl the instance whose methods will be called
* @param conf the configuration to use
* @param bindAddress the address to bind on to listen for connection
@@ -505,7 +342,8 @@ public Server(Object protocolImpl, Configuration conf, String bindAddress,
}
- /** Construct an RPC server.
+ /**
+ * Construct an RPC server.
* @param protocolClass - the protocol being registered
* can be null for compatibility with old usage (see below for details)
* @param protocolImpl the protocol impl that will be called
@@ -520,7 +358,7 @@ public Server(Class protocolClass, Object protocolImpl,
int numHandlers, int numReaders, int queueSizePerHandler,
boolean verbose, SecretManager secretManager)
throws IOException {
- super(bindAddress, port, Invocation.class, numHandlers, numReaders,
+ super(bindAddress, port, null, numHandlers, numReaders,
queueSizePerHandler, conf,
classNameBase(protocolImpl.getClass().getName()), secretManager);
@@ -535,7 +373,7 @@ public Server(Class protocolClass, Object protocolImpl,
* the protocolImpl is derived from the protocolClass(es)
* we register all interfaces extended by the protocolImpl
*/
- protocols = getProtocolInterfaces(protocolImpl.getClass());
+ protocols = RPC.getProtocolInterfaces(protocolImpl.getClass());
} else {
if (!protocolClass.isAssignableFrom(protocolImpl.getClass())) {
@@ -544,132 +382,125 @@ public Server(Class protocolClass, Object protocolImpl,
protocolImpl.getClass());
}
// register protocol class and its super interfaces
- registerProtocolAndImpl(protocolClass, protocolImpl);
- protocols = getProtocolInterfaces(protocolClass);
+ registerProtocolAndImpl(RpcKind.RPC_WRITABLE, protocolClass, protocolImpl);
+ protocols = RPC.getProtocolInterfaces(protocolClass);
}
for (Class p : protocols) {
if (!p.equals(VersionedProtocol.class)) {
- registerProtocolAndImpl(p, protocolImpl);
+ registerProtocolAndImpl(RpcKind.RPC_WRITABLE, p, protocolImpl);
}
}
}
-
- @Override
- public Server
- addProtocol(
- Class protocolClass, IMPL protocolImpl) throws IOException {
- registerProtocolAndImpl(protocolClass, protocolImpl);
- return this;
+ private static void log(String value) {
+ if (value!= null && value.length() > 55)
+ value = value.substring(0, 55)+"...";
+ LOG.info(value);
}
- /**
- * Process a client call
- * @param protocolName - the protocol name (the class of the client proxy
- * used to make calls to the rpc server.
- * @param param parameters
- * @param receivedTime time at which the call receoved (for metrics)
- * @return the call's return
- * @throws IOException
- */
- public Writable call(String protocolName, Writable param, long receivedTime)
- throws IOException {
- try {
- Invocation call = (Invocation)param;
- if (verbose) log("Call: " + call);
+ static class WritableRpcInvoker implements RpcInvoker {
- // Verify rpc version
- if (call.getRpcVersion() != writableRpcVersion) {
- // Client is using a different version of WritableRpc
- throw new IOException(
- "WritableRpc version mismatch, client side version="
- + call.getRpcVersion() + ", server side version="
- + writableRpcVersion);
- }
+ @Override
+ public Writable call(org.apache.hadoop.ipc.RPC.Server server,
+ String protocolName, Writable rpcRequest, long receivedTime)
+ throws IOException {
+ try {
+ Invocation call = (Invocation)rpcRequest;
+ if (server.verbose) log("Call: " + call);
- long clientVersion = call.getProtocolVersion();
- final String protoName;
- ProtoClassProtoImpl protocolImpl;
- if (call.declaringClassProtocolName.equals(VersionedProtocol.class.getName())) {
- // VersionProtocol methods are often used by client to figure out
- // which version of protocol to use.
- //
- // Versioned protocol methods should go the protocolName protocol
- // rather than the declaring class of the method since the
- // the declaring class is VersionedProtocol which is not
- // registered directly.
- // Send the call to the highest protocol version
- protocolImpl =
- getHighestSupportedProtocol(protocolName).protocolTarget;
- } else {
- protoName = call.declaringClassProtocolName;
+ // Verify rpc version
+ if (call.getRpcVersion() != writableRpcVersion) {
+ // Client is using a different version of WritableRpc
+ throw new IOException(
+ "WritableRpc version mismatch, client side version="
+ + call.getRpcVersion() + ", server side version="
+ + writableRpcVersion);
+ }
- // Find the right impl for the protocol based on client version.
- ProtoNameVer pv =
- new ProtoNameVer(call.declaringClassProtocolName, clientVersion);
- protocolImpl = protocolImplMap.get(pv);
- if (protocolImpl == null) { // no match for Protocol AND Version
- VerProtocolImpl highest =
- getHighestSupportedProtocol(protoName);
+ long clientVersion = call.getProtocolVersion();
+ final String protoName;
+ ProtoClassProtoImpl protocolImpl;
+ if (call.declaringClassProtocolName.equals(VersionedProtocol.class.getName())) {
+ // VersionProtocol methods are often used by client to figure out
+ // which version of protocol to use.
+ //
+ // Versioned protocol methods should go the protocolName protocol
+ // rather than the declaring class of the method since the
+ // the declaring class is VersionedProtocol which is not
+ // registered directly.
+ // Send the call to the highest protocol version
+ VerProtocolImpl highest = server.getHighestSupportedProtocol(
+ RpcKind.RPC_WRITABLE, protocolName);
if (highest == null) {
- throw new IOException("Unknown protocol: " + protoName);
- } else { // protocol supported but not the version that client wants
- throw new RPC.VersionMismatch(protoName, clientVersion,
- highest.version);
+ throw new IOException("Unknown protocol: " + protocolName);
+ }
+ protocolImpl = highest.protocolTarget;
+ } else {
+ protoName = call.declaringClassProtocolName;
+
+ // Find the right impl for the protocol based on client version.
+ ProtoNameVer pv =
+ new ProtoNameVer(call.declaringClassProtocolName, clientVersion);
+ protocolImpl =
+ server.getProtocolImplMap(RpcKind.RPC_WRITABLE).get(pv);
+ if (protocolImpl == null) { // no match for Protocol AND Version
+ VerProtocolImpl highest =
+ server.getHighestSupportedProtocol(RpcKind.RPC_WRITABLE,
+ protoName);
+ if (highest == null) {
+ throw new IOException("Unknown protocol: " + protoName);
+ } else { // protocol supported but not the version that client wants
+ throw new RPC.VersionMismatch(protoName, clientVersion,
+ highest.version);
+ }
}
}
- }
-
+
- // Invoke the protocol method
+ // Invoke the protocol method
- long startTime = System.currentTimeMillis();
- Method method =
- protocolImpl.protocolClass.getMethod(call.getMethodName(),
- call.getParameterClasses());
- method.setAccessible(true);
- rpcDetailedMetrics.init(protocolImpl.protocolClass);
- Object value =
- method.invoke(protocolImpl.protocolImpl, call.getParameters());
- int processingTime = (int) (System.currentTimeMillis() - startTime);
- int qTime = (int) (startTime-receivedTime);
- if (LOG.isDebugEnabled()) {
- LOG.debug("Served: " + call.getMethodName() +
- " queueTime= " + qTime +
- " procesingTime= " + processingTime);
- }
- rpcMetrics.addRpcQueueTime(qTime);
- rpcMetrics.addRpcProcessingTime(processingTime);
- rpcDetailedMetrics.addProcessingTime(call.getMethodName(),
- processingTime);
- if (verbose) log("Return: "+value);
+ long startTime = System.currentTimeMillis();
+ Method method =
+ protocolImpl.protocolClass.getMethod(call.getMethodName(),
+ call.getParameterClasses());
+ method.setAccessible(true);
+ server.rpcDetailedMetrics.init(protocolImpl.protocolClass);
+ Object value =
+ method.invoke(protocolImpl.protocolImpl, call.getParameters());
+ int processingTime = (int) (System.currentTimeMillis() - startTime);
+ int qTime = (int) (startTime-receivedTime);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Served: " + call.getMethodName() +
+ " queueTime= " + qTime +
+ " procesingTime= " + processingTime);
+ }
+ server.rpcMetrics.addRpcQueueTime(qTime);
+ server.rpcMetrics.addRpcProcessingTime(processingTime);
+ server.rpcDetailedMetrics.addProcessingTime(call.getMethodName(),
+ processingTime);
+ if (server.verbose) log("Return: "+value);
- return new ObjectWritable(method.getReturnType(), value);
+ return new ObjectWritable(method.getReturnType(), value);
- } catch (InvocationTargetException e) {
- Throwable target = e.getTargetException();
- if (target instanceof IOException) {
- throw (IOException)target;
- } else {
- IOException ioe = new IOException(target.toString());
- ioe.setStackTrace(target.getStackTrace());
+ } catch (InvocationTargetException e) {
+ Throwable target = e.getTargetException();
+ if (target instanceof IOException) {
+ throw (IOException)target;
+ } else {
+ IOException ioe = new IOException(target.toString());
+ ioe.setStackTrace(target.getStackTrace());
+ throw ioe;
+ }
+ } catch (Throwable e) {
+ if (!(e instanceof IOException)) {
+ LOG.error("Unexpected throwable object ", e);
+ }
+ IOException ioe = new IOException(e.toString());
+ ioe.setStackTrace(e.getStackTrace());
throw ioe;
}
- } catch (Throwable e) {
- if (!(e instanceof IOException)) {
- LOG.error("Unexpected throwable object ", e);
- }
- IOException ioe = new IOException(e.toString());
- ioe.setStackTrace(e.getStackTrace());
- throw ioe;
}
}
}
-
- private static void log(String value) {
- if (value!= null && value.length() > 55)
- value = value.substring(0, 55)+"...";
- LOG.info(value);
- }
}
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/protobuf/HadoopRpcProtos.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/protobuf/HadoopRpcProtos.java
index 2086f3d86b..b6d1577eb2 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/protobuf/HadoopRpcProtos.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/protobuf/HadoopRpcProtos.java
@@ -1,3 +1,20 @@
+/**
+ * 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.
+ */
// Generated by the protocol buffer compiler. DO NOT EDIT!
// source: hadoop_rpc.proto
@@ -18,6 +35,14 @@ public interface HadoopRpcRequestProtoOrBuilder
// optional bytes request = 2;
boolean hasRequest();
com.google.protobuf.ByteString getRequest();
+
+ // required string declaringClassProtocolName = 3;
+ boolean hasDeclaringClassProtocolName();
+ String getDeclaringClassProtocolName();
+
+ // required uint64 clientProtocolVersion = 4;
+ boolean hasClientProtocolVersion();
+ long getClientProtocolVersion();
}
public static final class HadoopRpcRequestProto extends
com.google.protobuf.GeneratedMessage
@@ -90,9 +115,53 @@ public com.google.protobuf.ByteString getRequest() {
return request_;
}
+ // required string declaringClassProtocolName = 3;
+ public static final int DECLARINGCLASSPROTOCOLNAME_FIELD_NUMBER = 3;
+ private java.lang.Object declaringClassProtocolName_;
+ public boolean hasDeclaringClassProtocolName() {
+ return ((bitField0_ & 0x00000004) == 0x00000004);
+ }
+ public String getDeclaringClassProtocolName() {
+ java.lang.Object ref = declaringClassProtocolName_;
+ if (ref instanceof String) {
+ return (String) ref;
+ } else {
+ com.google.protobuf.ByteString bs =
+ (com.google.protobuf.ByteString) ref;
+ String s = bs.toStringUtf8();
+ if (com.google.protobuf.Internal.isValidUtf8(bs)) {
+ declaringClassProtocolName_ = s;
+ }
+ return s;
+ }
+ }
+ private com.google.protobuf.ByteString getDeclaringClassProtocolNameBytes() {
+ java.lang.Object ref = declaringClassProtocolName_;
+ if (ref instanceof String) {
+ com.google.protobuf.ByteString b =
+ com.google.protobuf.ByteString.copyFromUtf8((String) ref);
+ declaringClassProtocolName_ = b;
+ return b;
+ } else {
+ return (com.google.protobuf.ByteString) ref;
+ }
+ }
+
+ // required uint64 clientProtocolVersion = 4;
+ public static final int CLIENTPROTOCOLVERSION_FIELD_NUMBER = 4;
+ private long clientProtocolVersion_;
+ public boolean hasClientProtocolVersion() {
+ return ((bitField0_ & 0x00000008) == 0x00000008);
+ }
+ public long getClientProtocolVersion() {
+ return clientProtocolVersion_;
+ }
+
private void initFields() {
methodName_ = "";
request_ = com.google.protobuf.ByteString.EMPTY;
+ declaringClassProtocolName_ = "";
+ clientProtocolVersion_ = 0L;
}
private byte memoizedIsInitialized = -1;
public final boolean isInitialized() {
@@ -103,6 +172,14 @@ public final boolean isInitialized() {
memoizedIsInitialized = 0;
return false;
}
+ if (!hasDeclaringClassProtocolName()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
+ if (!hasClientProtocolVersion()) {
+ memoizedIsInitialized = 0;
+ return false;
+ }
memoizedIsInitialized = 1;
return true;
}
@@ -116,6 +193,12 @@ public void writeTo(com.google.protobuf.CodedOutputStream output)
if (((bitField0_ & 0x00000002) == 0x00000002)) {
output.writeBytes(2, request_);
}
+ if (((bitField0_ & 0x00000004) == 0x00000004)) {
+ output.writeBytes(3, getDeclaringClassProtocolNameBytes());
+ }
+ if (((bitField0_ & 0x00000008) == 0x00000008)) {
+ output.writeUInt64(4, clientProtocolVersion_);
+ }
getUnknownFields().writeTo(output);
}
@@ -133,6 +216,14 @@ public int getSerializedSize() {
size += com.google.protobuf.CodedOutputStream
.computeBytesSize(2, request_);
}
+ if (((bitField0_ & 0x00000004) == 0x00000004)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeBytesSize(3, getDeclaringClassProtocolNameBytes());
+ }
+ if (((bitField0_ & 0x00000008) == 0x00000008)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeUInt64Size(4, clientProtocolVersion_);
+ }
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
return size;
@@ -166,6 +257,16 @@ public boolean equals(final java.lang.Object obj) {
result = result && getRequest()
.equals(other.getRequest());
}
+ result = result && (hasDeclaringClassProtocolName() == other.hasDeclaringClassProtocolName());
+ if (hasDeclaringClassProtocolName()) {
+ result = result && getDeclaringClassProtocolName()
+ .equals(other.getDeclaringClassProtocolName());
+ }
+ result = result && (hasClientProtocolVersion() == other.hasClientProtocolVersion());
+ if (hasClientProtocolVersion()) {
+ result = result && (getClientProtocolVersion()
+ == other.getClientProtocolVersion());
+ }
result = result &&
getUnknownFields().equals(other.getUnknownFields());
return result;
@@ -183,6 +284,14 @@ public int hashCode() {
hash = (37 * hash) + REQUEST_FIELD_NUMBER;
hash = (53 * hash) + getRequest().hashCode();
}
+ if (hasDeclaringClassProtocolName()) {
+ hash = (37 * hash) + DECLARINGCLASSPROTOCOLNAME_FIELD_NUMBER;
+ hash = (53 * hash) + getDeclaringClassProtocolName().hashCode();
+ }
+ if (hasClientProtocolVersion()) {
+ hash = (37 * hash) + CLIENTPROTOCOLVERSION_FIELD_NUMBER;
+ hash = (53 * hash) + hashLong(getClientProtocolVersion());
+ }
hash = (29 * hash) + getUnknownFields().hashCode();
return hash;
}
@@ -303,6 +412,10 @@ public Builder clear() {
bitField0_ = (bitField0_ & ~0x00000001);
request_ = com.google.protobuf.ByteString.EMPTY;
bitField0_ = (bitField0_ & ~0x00000002);
+ declaringClassProtocolName_ = "";
+ bitField0_ = (bitField0_ & ~0x00000004);
+ clientProtocolVersion_ = 0L;
+ bitField0_ = (bitField0_ & ~0x00000008);
return this;
}
@@ -349,6 +462,14 @@ public org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRpcRequestProto buil
to_bitField0_ |= 0x00000002;
}
result.request_ = request_;
+ if (((from_bitField0_ & 0x00000004) == 0x00000004)) {
+ to_bitField0_ |= 0x00000004;
+ }
+ result.declaringClassProtocolName_ = declaringClassProtocolName_;
+ if (((from_bitField0_ & 0x00000008) == 0x00000008)) {
+ to_bitField0_ |= 0x00000008;
+ }
+ result.clientProtocolVersion_ = clientProtocolVersion_;
result.bitField0_ = to_bitField0_;
onBuilt();
return result;
@@ -371,6 +492,12 @@ public Builder mergeFrom(org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRp
if (other.hasRequest()) {
setRequest(other.getRequest());
}
+ if (other.hasDeclaringClassProtocolName()) {
+ setDeclaringClassProtocolName(other.getDeclaringClassProtocolName());
+ }
+ if (other.hasClientProtocolVersion()) {
+ setClientProtocolVersion(other.getClientProtocolVersion());
+ }
this.mergeUnknownFields(other.getUnknownFields());
return this;
}
@@ -380,6 +507,14 @@ public final boolean isInitialized() {
return false;
}
+ if (!hasDeclaringClassProtocolName()) {
+
+ return false;
+ }
+ if (!hasClientProtocolVersion()) {
+
+ return false;
+ }
return true;
}
@@ -416,6 +551,16 @@ public Builder mergeFrom(
request_ = input.readBytes();
break;
}
+ case 26: {
+ bitField0_ |= 0x00000004;
+ declaringClassProtocolName_ = input.readBytes();
+ break;
+ }
+ case 32: {
+ bitField0_ |= 0x00000008;
+ clientProtocolVersion_ = input.readUInt64();
+ break;
+ }
}
}
}
@@ -482,6 +627,63 @@ public Builder clearRequest() {
return this;
}
+ // required string declaringClassProtocolName = 3;
+ private java.lang.Object declaringClassProtocolName_ = "";
+ public boolean hasDeclaringClassProtocolName() {
+ return ((bitField0_ & 0x00000004) == 0x00000004);
+ }
+ public String getDeclaringClassProtocolName() {
+ java.lang.Object ref = declaringClassProtocolName_;
+ if (!(ref instanceof String)) {
+ String s = ((com.google.protobuf.ByteString) ref).toStringUtf8();
+ declaringClassProtocolName_ = s;
+ return s;
+ } else {
+ return (String) ref;
+ }
+ }
+ public Builder setDeclaringClassProtocolName(String value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ bitField0_ |= 0x00000004;
+ declaringClassProtocolName_ = value;
+ onChanged();
+ return this;
+ }
+ public Builder clearDeclaringClassProtocolName() {
+ bitField0_ = (bitField0_ & ~0x00000004);
+ declaringClassProtocolName_ = getDefaultInstance().getDeclaringClassProtocolName();
+ onChanged();
+ return this;
+ }
+ void setDeclaringClassProtocolName(com.google.protobuf.ByteString value) {
+ bitField0_ |= 0x00000004;
+ declaringClassProtocolName_ = value;
+ onChanged();
+ }
+
+ // required uint64 clientProtocolVersion = 4;
+ private long clientProtocolVersion_ ;
+ public boolean hasClientProtocolVersion() {
+ return ((bitField0_ & 0x00000008) == 0x00000008);
+ }
+ public long getClientProtocolVersion() {
+ return clientProtocolVersion_;
+ }
+ public Builder setClientProtocolVersion(long value) {
+ bitField0_ |= 0x00000008;
+ clientProtocolVersion_ = value;
+ onChanged();
+ return this;
+ }
+ public Builder clearClientProtocolVersion() {
+ bitField0_ = (bitField0_ & ~0x00000008);
+ clientProtocolVersion_ = 0L;
+ onChanged();
+ return this;
+ }
+
// @@protoc_insertion_point(builder_scope:HadoopRpcRequestProto)
}
@@ -1706,16 +1908,18 @@ public org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRpcExceptionProtoOrB
descriptor;
static {
java.lang.String[] descriptorData = {
- "\n\020hadoop_rpc.proto\"<\n\025HadoopRpcRequestPr" +
+ "\n\020hadoop_rpc.proto\"\177\n\025HadoopRpcRequestPr" +
"oto\022\022\n\nmethodName\030\001 \002(\t\022\017\n\007request\030\002 \001(\014" +
- "\"D\n\027HadoopRpcExceptionProto\022\025\n\rexception" +
- "Name\030\001 \001(\t\022\022\n\nstackTrace\030\002 \001(\t\"\272\001\n\026Hadoo" +
- "pRpcResponseProto\0226\n\006status\030\001 \002(\0162&.Hado" +
- "opRpcResponseProto.ResponseStatus\022\020\n\010res" +
- "ponse\030\002 \001(\014\022+\n\texception\030\003 \001(\0132\030.HadoopR" +
- "pcExceptionProto\")\n\016ResponseStatus\022\013\n\007SU" +
- "CCESS\020\001\022\n\n\006ERRROR\020\002B4\n\036org.apache.hadoop" +
- ".ipc.protobufB\017HadoopRpcProtos\240\001\001"
+ "\022\"\n\032declaringClassProtocolName\030\003 \002(\t\022\035\n\025" +
+ "clientProtocolVersion\030\004 \002(\004\"D\n\027HadoopRpc" +
+ "ExceptionProto\022\025\n\rexceptionName\030\001 \001(\t\022\022\n" +
+ "\nstackTrace\030\002 \001(\t\"\272\001\n\026HadoopRpcResponseP" +
+ "roto\0226\n\006status\030\001 \002(\0162&.HadoopRpcResponse" +
+ "Proto.ResponseStatus\022\020\n\010response\030\002 \001(\014\022+" +
+ "\n\texception\030\003 \001(\0132\030.HadoopRpcExceptionPr" +
+ "oto\")\n\016ResponseStatus\022\013\n\007SUCCESS\020\001\022\n\n\006ER",
+ "RROR\020\002B4\n\036org.apache.hadoop.ipc.protobuf" +
+ "B\017HadoopRpcProtos\240\001\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -1727,7 +1931,7 @@ public com.google.protobuf.ExtensionRegistry assignDescriptors(
internal_static_HadoopRpcRequestProto_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_HadoopRpcRequestProto_descriptor,
- new java.lang.String[] { "MethodName", "Request", },
+ new java.lang.String[] { "MethodName", "Request", "DeclaringClassProtocolName", "ClientProtocolVersion", },
org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRpcRequestProto.class,
org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRpcRequestProto.Builder.class);
internal_static_HadoopRpcExceptionProto_descriptor =
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/DelegationKey.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/DelegationKey.java
index 28d82537fb..7cea67923e 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/DelegationKey.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/delegation/DelegationKey.java
@@ -42,15 +42,20 @@ public class DelegationKey implements Writable {
@Nullable
private byte[] keyBytes = null;
+ /** Default constructore required for Writable */
public DelegationKey() {
- this(0, 0L, null);
+ this(0, 0L, (SecretKey)null);
}
public DelegationKey(int keyId, long expiryDate, SecretKey key) {
+ this(keyId, expiryDate, key != null ? key.getEncoded() : null);
+ }
+
+ public DelegationKey(int keyId, long expiryDate, byte[] encodedKey) {
this.keyId = keyId;
this.expiryDate = expiryDate;
- if (key!=null) {
- this.keyBytes = key.getEncoded();
+ if (encodedKey != null) {
+ this.keyBytes = encodedKey;
}
}
@@ -70,6 +75,10 @@ public SecretKey getKey() {
return key;
}
}
+
+ public byte[] getEncodedKey() {
+ return keyBytes;
+ }
public void setExpiryDate(long expiryDate) {
this.expiryDate = expiryDate;
diff --git a/hadoop-common-project/hadoop-common/src/proto/hadoop_rpc.proto b/hadoop-common-project/hadoop-common/src/proto/hadoop_rpc.proto
index d37455434d..7af8eae182 100644
--- a/hadoop-common-project/hadoop-common/src/proto/hadoop_rpc.proto
+++ b/hadoop-common-project/hadoop-common/src/proto/hadoop_rpc.proto
@@ -34,6 +34,12 @@ message HadoopRpcRequestProto {
/** Bytes corresponding to the client protobuf request */
optional bytes request = 2;
+
+ /** protocol name of class declaring the called method */
+ required string declaringClassProtocolName = 3;
+
+ /** protocol version of class declaring the called method */
+ required uint64 clientProtocolVersion = 4;
}
/**
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAvroRpc.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAvroRpc.java
index e7b6657a5c..5ce3359428 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAvroRpc.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAvroRpc.java
@@ -34,6 +34,7 @@
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.ipc.TestSaslRPC.CustomSecurityInfo;
import org.apache.hadoop.ipc.TestSaslRPC.TestTokenIdentifier;
import org.apache.hadoop.ipc.TestSaslRPC.TestTokenSecretManager;
@@ -101,7 +102,8 @@ private void testReflect(boolean secure) throws Exception {
RPC.setProtocolEngine(conf, AvroTestProtocol.class, AvroRpcEngine.class);
RPC.Server server = RPC.getServer(EmptyProtocol.class, new EmptyImpl(),
ADDRESS, 0, 5, true, conf, sm);
- server.addProtocol(AvroTestProtocol.class, new TestImpl());
+ server.addProtocol(RpcKind.RPC_WRITABLE,
+ AvroTestProtocol.class, new TestImpl());
try {
server.start();
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java
index 1515ba6216..1f3e67a4f9 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPC.java
@@ -23,6 +23,7 @@
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.net.NetUtils;
@@ -96,8 +97,8 @@ public TestServer(int handlerCount, boolean sleep,
}
@Override
- public Writable call(String protocol, Writable param, long receiveTime)
- throws IOException {
+ public Writable call(RpcKind rpcKind, String protocol, Writable param,
+ long receiveTime) throws IOException {
if (sleep) {
// sleep a bit
try {
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPCServerResponder.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPCServerResponder.java
index d4400effa7..5675cbfddf 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPCServerResponder.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestIPCServerResponder.java
@@ -30,6 +30,7 @@
import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.net.NetUtils;
/**
@@ -72,8 +73,8 @@ public TestServer(final int handlerCount, final boolean sleep)
}
@Override
- public Writable call(String protocol, Writable param, long receiveTime)
- throws IOException {
+ public Writable call(RpcKind rpcKind, String protocol, Writable param,
+ long receiveTime) throws IOException {
if (sleep) {
try {
Thread.sleep(RANDOM.nextInt(20)); // sleep a bit
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestMultipleProtocolServer.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestMultipleProtocolServer.java
index 203c2855bc..f5acd93eb2 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestMultipleProtocolServer.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestMultipleProtocolServer.java
@@ -23,10 +23,15 @@
import org.junit.Assert;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
+import org.apache.hadoop.ipc.TestProtoBufRpc.PBServerImpl;
+import org.apache.hadoop.ipc.TestProtoBufRpc.TestRpcService;
+import org.apache.hadoop.ipc.protobuf.TestRpcServiceProtos.TestProtobufRpcProto;
import org.apache.hadoop.net.NetUtils;
import org.junit.Before;
import org.junit.After;
import org.junit.Test;
+import com.google.protobuf.BlockingService;
public class TestMultipleProtocolServer {
private static final String ADDRESS = "0.0.0.0";
@@ -173,9 +178,19 @@ public void setUp() throws Exception {
// create a server with two handlers
server = RPC.getServer(Foo0.class,
new Foo0Impl(), ADDRESS, 0, 2, false, conf, null);
- server.addProtocol(Foo1.class, new Foo1Impl());
- server.addProtocol(Bar.class, new BarImpl());
- server.addProtocol(Mixin.class, new BarImpl());
+ server.addProtocol(RpcKind.RPC_WRITABLE, Foo1.class, new Foo1Impl());
+ server.addProtocol(RpcKind.RPC_WRITABLE, Bar.class, new BarImpl());
+ server.addProtocol(RpcKind.RPC_WRITABLE, Mixin.class, new BarImpl());
+
+
+ // Add Protobuf server
+ // Create server side implementation
+ PBServerImpl pbServerImpl =
+ new PBServerImpl();
+ BlockingService service = TestProtobufRpcProto
+ .newReflectiveBlockingService(pbServerImpl);
+ server.addProtocol(RpcKind.RPC_PROTOCOL_BUFFER, TestRpcService.class,
+ service);
server.start();
addr = NetUtils.getConnectAddress(server);
}
@@ -251,5 +266,16 @@ public void testNonExistingProtocol2() throws IOException {
public void testIncorrectServerCreation() throws IOException {
RPC.getServer(Foo1.class,
new Foo0Impl(), ADDRESS, 0, 2, false, conf, null);
+ }
+
+ // Now test a PB service - a server hosts both PB and Writable Rpcs.
+ @Test
+ public void testPBService() throws Exception {
+ // Set RPC engine to protobuf RPC engine
+ Configuration conf2 = new Configuration();
+ RPC.setProtocolEngine(conf2, TestRpcService.class,
+ ProtobufRpcEngine.class);
+ TestRpcService client = RPC.getProxy(TestRpcService.class, 0, addr, conf2);
+ TestProtoBufRpc.testProtoBufRpc(client);
}
}
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestProtoBufRpc.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestProtoBufRpc.java
index 95083ab2fa..ea48b98db4 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestProtoBufRpc.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestProtoBufRpc.java
@@ -21,14 +21,18 @@
import java.net.InetSocketAddress;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.ipc.protobuf.TestProtos.EchoRequestProto;
import org.apache.hadoop.ipc.protobuf.TestProtos.EchoResponseProto;
import org.apache.hadoop.ipc.protobuf.TestProtos.EmptyRequestProto;
import org.apache.hadoop.ipc.protobuf.TestProtos.EmptyResponseProto;
import org.apache.hadoop.ipc.protobuf.TestRpcServiceProtos.TestProtobufRpcProto;
-import org.apache.hadoop.ipc.protobuf.TestRpcServiceProtos.TestProtobufRpcProto.BlockingInterface;
+import org.apache.hadoop.ipc.protobuf.TestRpcServiceProtos.TestProtobufRpc2Proto;
+import org.apache.hadoop.net.NetUtils;
import org.junit.Assert;
import org.junit.Test;
+import org.junit.Before;
+import org.junit.After;
import com.google.protobuf.BlockingService;
import com.google.protobuf.RpcController;
@@ -42,8 +46,21 @@
public class TestProtoBufRpc {
public final static String ADDRESS = "0.0.0.0";
public final static int PORT = 0;
+ private static InetSocketAddress addr;
+ private static Configuration conf;
+ private static RPC.Server server;
+
+ @ProtocolInfo(protocolName = "testProto", protocolVersion = 1)
+ public interface TestRpcService
+ extends TestProtobufRpcProto.BlockingInterface {
+ }
- public static class ServerImpl implements BlockingInterface {
+ @ProtocolInfo(protocolName = "testProto2", protocolVersion = 1)
+ public interface TestRpcService2 extends
+ TestProtobufRpc2Proto.BlockingInterface {
+ }
+
+ public static class PBServerImpl implements TestRpcService {
@Override
public EmptyResponseProto ping(RpcController unused,
@@ -64,40 +81,78 @@ public EmptyResponseProto error(RpcController unused,
throw new ServiceException("error", new RpcServerException("error"));
}
}
+
+ public static class PBServer2Impl implements TestRpcService2 {
- private static RPC.Server startRPCServer(Configuration conf)
- throws IOException {
+ @Override
+ public EmptyResponseProto ping2(RpcController unused,
+ EmptyRequestProto request) throws ServiceException {
+ return EmptyResponseProto.newBuilder().build();
+ }
+
+ @Override
+ public EchoResponseProto echo2(RpcController unused, EchoRequestProto request)
+ throws ServiceException {
+ return EchoResponseProto.newBuilder().setMessage(request.getMessage())
+ .build();
+ }
+ }
+
+ @Before
+ public void setUp() throws IOException { // Setup server for both protocols
+ conf = new Configuration();
// Set RPC engine to protobuf RPC engine
- RPC.setProtocolEngine(conf, BlockingService.class, ProtobufRpcEngine.class);
+ RPC.setProtocolEngine(conf, TestRpcService.class, ProtobufRpcEngine.class);
// Create server side implementation
- ServerImpl serverImpl = new ServerImpl();
+ PBServerImpl serverImpl = new PBServerImpl();
BlockingService service = TestProtobufRpcProto
.newReflectiveBlockingService(serverImpl);
- // Get RPC server for serer side implementation
- RPC.Server server = RPC.getServer(BlockingService.class, service, ADDRESS,
- PORT, conf);
+ // Get RPC server for server side implementation
+ server = RPC.getServer(TestRpcService.class, service, ADDRESS, PORT, conf);
+ addr = NetUtils.getConnectAddress(server);
+
+ // now the second protocol
+ PBServer2Impl server2Impl = new PBServer2Impl();
+ BlockingService service2 = TestProtobufRpc2Proto
+ .newReflectiveBlockingService(server2Impl);
+
+ server.addProtocol(RpcKind.RPC_PROTOCOL_BUFFER, TestRpcService2.class,
+ service2);
server.start();
- return server;
+ }
+
+
+ @After
+ public void tearDown() throws Exception {
+ server.stop();
}
- private static BlockingInterface getClient(Configuration conf,
- InetSocketAddress addr) throws IOException {
+ private static TestRpcService getClient() throws IOException {
// Set RPC engine to protobuf RPC engine
- RPC.setProtocolEngine(conf, BlockingInterface.class,
+ RPC.setProtocolEngine(conf, TestRpcService.class,
ProtobufRpcEngine.class);
- BlockingInterface client = RPC.getProxy(BlockingInterface.class, 0, addr,
+ return RPC.getProxy(TestRpcService.class, 0, addr,
+ conf);
+ }
+
+ private static TestRpcService2 getClient2() throws IOException {
+ // Set RPC engine to protobuf RPC engine
+ RPC.setProtocolEngine(conf, TestRpcService2.class,
+ ProtobufRpcEngine.class);
+ return RPC.getProxy(TestRpcService2.class, 0, addr,
conf);
- return client;
}
@Test
public void testProtoBufRpc() throws Exception {
- Configuration conf = new Configuration();
- RPC.Server server = startRPCServer(conf);
- BlockingInterface client = getClient(conf, server.getListenerAddress());
-
+ TestRpcService client = getClient();
+ testProtoBufRpc(client);
+ }
+
+ // separated test out so that other tests can call it.
+ public static void testProtoBufRpc(TestRpcService client) throws Exception {
// Test ping method
EmptyRequestProto emptyRequest = EmptyRequestProto.newBuilder().build();
client.ping(null, emptyRequest);
@@ -108,16 +163,29 @@ public void testProtoBufRpc() throws Exception {
EchoResponseProto echoResponse = client.echo(null, echoRequest);
Assert.assertEquals(echoResponse.getMessage(), "hello");
- // Test error method - it should be thrown as RemoteException
+ // Test error method - error should be thrown as RemoteException
try {
client.error(null, emptyRequest);
Assert.fail("Expected exception is not thrown");
} catch (ServiceException e) {
RemoteException re = (RemoteException)e.getCause();
- re.printStackTrace();
RpcServerException rse = (RpcServerException) re
.unwrapRemoteException(RpcServerException.class);
- rse.printStackTrace();
}
}
+
+ @Test
+ public void testProtoBufRpc2() throws Exception {
+ TestRpcService2 client = getClient2();
+
+ // Test ping method
+ EmptyRequestProto emptyRequest = EmptyRequestProto.newBuilder().build();
+ client.ping2(null, emptyRequest);
+
+ // Test echo method
+ EchoRequestProto echoRequest = EchoRequestProto.newBuilder()
+ .setMessage("hello").build();
+ EchoResponseProto echoResponse = client.echo2(null, echoRequest);
+ Assert.assertEquals(echoResponse.getMessage(), "hello");
+ }
}
\ No newline at end of file
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPCCompatibility.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPCCompatibility.java
index 85e60dde9f..d38d823200 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPCCompatibility.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRPCCompatibility.java
@@ -31,6 +31,7 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.net.NetUtils;
import org.junit.After;
import org.junit.Test;
@@ -56,6 +57,8 @@ public interface TestProtocol1 extends VersionedProtocol, TestProtocol0 {
String echo(String value) throws IOException;
}
+
+ // TestProtocol2 is a compatible impl of TestProtocol1 - hence use its name
@ProtocolInfo(protocolName=
"org.apache.hadoop.ipc.TestRPCCompatibility$TestProtocol1")
public interface TestProtocol2 extends TestProtocol1 {
@@ -114,9 +117,11 @@ public long getProtocolVersion(String protocol,
public void tearDown() throws IOException {
if (proxy != null) {
RPC.stopProxy(proxy.getProxy());
+ proxy = null;
}
if (server != null) {
server.stop();
+ server = null;
}
}
@@ -126,7 +131,7 @@ public void testVersion0ClientVersion1Server() throws Exception {
TestImpl1 impl = new TestImpl1();
server = RPC.getServer(TestProtocol1.class,
impl, ADDRESS, 0, 2, false, conf, null);
- server.addProtocol(TestProtocol0.class, impl);
+ server.addProtocol(RpcKind.RPC_WRITABLE, TestProtocol0.class, impl);
server.start();
addr = NetUtils.getConnectAddress(server);
@@ -170,8 +175,10 @@ private Version2Client() throws IOException {
public int echo(int value) throws IOException, NumberFormatException {
if (serverInfo.isMethodSupported("echo", int.class)) {
+System.out.println("echo int is supported");
return -value; // use version 3 echo long
} else { // server is version 2
+System.out.println("echo int is NOT supported");
return Integer.parseInt(proxy2.echo(String.valueOf(value)));
}
}
@@ -191,7 +198,7 @@ public void testVersion2ClientVersion1Server() throws Exception {
TestImpl1 impl = new TestImpl1();
server = RPC.getServer(TestProtocol1.class,
impl, ADDRESS, 0, 2, false, conf, null);
- server.addProtocol(TestProtocol0.class, impl);
+ server.addProtocol(RpcKind.RPC_WRITABLE, TestProtocol0.class, impl);
server.start();
addr = NetUtils.getConnectAddress(server);
@@ -207,11 +214,12 @@ public void testVersion2ClientVersion1Server() throws Exception {
@Test // equal version client and server
public void testVersion2ClientVersion2Server() throws Exception {
+ ProtocolSignature.resetCache();
// create a server with two handlers
TestImpl2 impl = new TestImpl2();
server = RPC.getServer(TestProtocol2.class,
impl, ADDRESS, 0, 2, false, conf, null);
- server.addProtocol(TestProtocol0.class, impl);
+ server.addProtocol(RpcKind.RPC_WRITABLE, TestProtocol0.class, impl);
server.start();
addr = NetUtils.getConnectAddress(server);
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/protobuf/TestRpcServiceProtos.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/protobuf/TestRpcServiceProtos.java
index 214b04f6b9..8b192971ab 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/protobuf/TestRpcServiceProtos.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/protobuf/TestRpcServiceProtos.java
@@ -359,6 +359,292 @@ public org.apache.hadoop.ipc.protobuf.TestProtos.EmptyResponseProto error(
}
}
+ public static abstract class TestProtobufRpc2Proto
+ implements com.google.protobuf.Service {
+ protected TestProtobufRpc2Proto() {}
+
+ public interface Interface {
+ public abstract void ping2(
+ com.google.protobuf.RpcController controller,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EmptyRequestProto request,
+ com.google.protobuf.RpcCallback done);
+
+ public abstract void echo2(
+ com.google.protobuf.RpcController controller,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EchoRequestProto request,
+ com.google.protobuf.RpcCallback done);
+
+ }
+
+ public static com.google.protobuf.Service newReflectiveService(
+ final Interface impl) {
+ return new TestProtobufRpc2Proto() {
+ @java.lang.Override
+ public void ping2(
+ com.google.protobuf.RpcController controller,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EmptyRequestProto request,
+ com.google.protobuf.RpcCallback done) {
+ impl.ping2(controller, request, done);
+ }
+
+ @java.lang.Override
+ public void echo2(
+ com.google.protobuf.RpcController controller,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EchoRequestProto request,
+ com.google.protobuf.RpcCallback done) {
+ impl.echo2(controller, request, done);
+ }
+
+ };
+ }
+
+ public static com.google.protobuf.BlockingService
+ newReflectiveBlockingService(final BlockingInterface impl) {
+ return new com.google.protobuf.BlockingService() {
+ public final com.google.protobuf.Descriptors.ServiceDescriptor
+ getDescriptorForType() {
+ return getDescriptor();
+ }
+
+ public final com.google.protobuf.Message callBlockingMethod(
+ com.google.protobuf.Descriptors.MethodDescriptor method,
+ com.google.protobuf.RpcController controller,
+ com.google.protobuf.Message request)
+ throws com.google.protobuf.ServiceException {
+ if (method.getService() != getDescriptor()) {
+ throw new java.lang.IllegalArgumentException(
+ "Service.callBlockingMethod() given method descriptor for " +
+ "wrong service type.");
+ }
+ switch(method.getIndex()) {
+ case 0:
+ return impl.ping2(controller, (org.apache.hadoop.ipc.protobuf.TestProtos.EmptyRequestProto)request);
+ case 1:
+ return impl.echo2(controller, (org.apache.hadoop.ipc.protobuf.TestProtos.EchoRequestProto)request);
+ default:
+ throw new java.lang.AssertionError("Can't get here.");
+ }
+ }
+
+ public final com.google.protobuf.Message
+ getRequestPrototype(
+ com.google.protobuf.Descriptors.MethodDescriptor method) {
+ if (method.getService() != getDescriptor()) {
+ throw new java.lang.IllegalArgumentException(
+ "Service.getRequestPrototype() given method " +
+ "descriptor for wrong service type.");
+ }
+ switch(method.getIndex()) {
+ case 0:
+ return org.apache.hadoop.ipc.protobuf.TestProtos.EmptyRequestProto.getDefaultInstance();
+ case 1:
+ return org.apache.hadoop.ipc.protobuf.TestProtos.EchoRequestProto.getDefaultInstance();
+ default:
+ throw new java.lang.AssertionError("Can't get here.");
+ }
+ }
+
+ public final com.google.protobuf.Message
+ getResponsePrototype(
+ com.google.protobuf.Descriptors.MethodDescriptor method) {
+ if (method.getService() != getDescriptor()) {
+ throw new java.lang.IllegalArgumentException(
+ "Service.getResponsePrototype() given method " +
+ "descriptor for wrong service type.");
+ }
+ switch(method.getIndex()) {
+ case 0:
+ return org.apache.hadoop.ipc.protobuf.TestProtos.EmptyResponseProto.getDefaultInstance();
+ case 1:
+ return org.apache.hadoop.ipc.protobuf.TestProtos.EchoResponseProto.getDefaultInstance();
+ default:
+ throw new java.lang.AssertionError("Can't get here.");
+ }
+ }
+
+ };
+ }
+
+ public abstract void ping2(
+ com.google.protobuf.RpcController controller,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EmptyRequestProto request,
+ com.google.protobuf.RpcCallback done);
+
+ public abstract void echo2(
+ com.google.protobuf.RpcController controller,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EchoRequestProto request,
+ com.google.protobuf.RpcCallback done);
+
+ public static final
+ com.google.protobuf.Descriptors.ServiceDescriptor
+ getDescriptor() {
+ return org.apache.hadoop.ipc.protobuf.TestRpcServiceProtos.getDescriptor().getServices().get(1);
+ }
+ public final com.google.protobuf.Descriptors.ServiceDescriptor
+ getDescriptorForType() {
+ return getDescriptor();
+ }
+
+ public final void callMethod(
+ com.google.protobuf.Descriptors.MethodDescriptor method,
+ com.google.protobuf.RpcController controller,
+ com.google.protobuf.Message request,
+ com.google.protobuf.RpcCallback<
+ com.google.protobuf.Message> done) {
+ if (method.getService() != getDescriptor()) {
+ throw new java.lang.IllegalArgumentException(
+ "Service.callMethod() given method descriptor for wrong " +
+ "service type.");
+ }
+ switch(method.getIndex()) {
+ case 0:
+ this.ping2(controller, (org.apache.hadoop.ipc.protobuf.TestProtos.EmptyRequestProto)request,
+ com.google.protobuf.RpcUtil.specializeCallback(
+ done));
+ return;
+ case 1:
+ this.echo2(controller, (org.apache.hadoop.ipc.protobuf.TestProtos.EchoRequestProto)request,
+ com.google.protobuf.RpcUtil.specializeCallback(
+ done));
+ return;
+ default:
+ throw new java.lang.AssertionError("Can't get here.");
+ }
+ }
+
+ public final com.google.protobuf.Message
+ getRequestPrototype(
+ com.google.protobuf.Descriptors.MethodDescriptor method) {
+ if (method.getService() != getDescriptor()) {
+ throw new java.lang.IllegalArgumentException(
+ "Service.getRequestPrototype() given method " +
+ "descriptor for wrong service type.");
+ }
+ switch(method.getIndex()) {
+ case 0:
+ return org.apache.hadoop.ipc.protobuf.TestProtos.EmptyRequestProto.getDefaultInstance();
+ case 1:
+ return org.apache.hadoop.ipc.protobuf.TestProtos.EchoRequestProto.getDefaultInstance();
+ default:
+ throw new java.lang.AssertionError("Can't get here.");
+ }
+ }
+
+ public final com.google.protobuf.Message
+ getResponsePrototype(
+ com.google.protobuf.Descriptors.MethodDescriptor method) {
+ if (method.getService() != getDescriptor()) {
+ throw new java.lang.IllegalArgumentException(
+ "Service.getResponsePrototype() given method " +
+ "descriptor for wrong service type.");
+ }
+ switch(method.getIndex()) {
+ case 0:
+ return org.apache.hadoop.ipc.protobuf.TestProtos.EmptyResponseProto.getDefaultInstance();
+ case 1:
+ return org.apache.hadoop.ipc.protobuf.TestProtos.EchoResponseProto.getDefaultInstance();
+ default:
+ throw new java.lang.AssertionError("Can't get here.");
+ }
+ }
+
+ public static Stub newStub(
+ com.google.protobuf.RpcChannel channel) {
+ return new Stub(channel);
+ }
+
+ public static final class Stub extends org.apache.hadoop.ipc.protobuf.TestRpcServiceProtos.TestProtobufRpc2Proto implements Interface {
+ private Stub(com.google.protobuf.RpcChannel channel) {
+ this.channel = channel;
+ }
+
+ private final com.google.protobuf.RpcChannel channel;
+
+ public com.google.protobuf.RpcChannel getChannel() {
+ return channel;
+ }
+
+ public void ping2(
+ com.google.protobuf.RpcController controller,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EmptyRequestProto request,
+ com.google.protobuf.RpcCallback done) {
+ channel.callMethod(
+ getDescriptor().getMethods().get(0),
+ controller,
+ request,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EmptyResponseProto.getDefaultInstance(),
+ com.google.protobuf.RpcUtil.generalizeCallback(
+ done,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EmptyResponseProto.class,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EmptyResponseProto.getDefaultInstance()));
+ }
+
+ public void echo2(
+ com.google.protobuf.RpcController controller,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EchoRequestProto request,
+ com.google.protobuf.RpcCallback done) {
+ channel.callMethod(
+ getDescriptor().getMethods().get(1),
+ controller,
+ request,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EchoResponseProto.getDefaultInstance(),
+ com.google.protobuf.RpcUtil.generalizeCallback(
+ done,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EchoResponseProto.class,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EchoResponseProto.getDefaultInstance()));
+ }
+ }
+
+ public static BlockingInterface newBlockingStub(
+ com.google.protobuf.BlockingRpcChannel channel) {
+ return new BlockingStub(channel);
+ }
+
+ public interface BlockingInterface {
+ public org.apache.hadoop.ipc.protobuf.TestProtos.EmptyResponseProto ping2(
+ com.google.protobuf.RpcController controller,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EmptyRequestProto request)
+ throws com.google.protobuf.ServiceException;
+
+ public org.apache.hadoop.ipc.protobuf.TestProtos.EchoResponseProto echo2(
+ com.google.protobuf.RpcController controller,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EchoRequestProto request)
+ throws com.google.protobuf.ServiceException;
+ }
+
+ private static final class BlockingStub implements BlockingInterface {
+ private BlockingStub(com.google.protobuf.BlockingRpcChannel channel) {
+ this.channel = channel;
+ }
+
+ private final com.google.protobuf.BlockingRpcChannel channel;
+
+ public org.apache.hadoop.ipc.protobuf.TestProtos.EmptyResponseProto ping2(
+ com.google.protobuf.RpcController controller,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EmptyRequestProto request)
+ throws com.google.protobuf.ServiceException {
+ return (org.apache.hadoop.ipc.protobuf.TestProtos.EmptyResponseProto) channel.callBlockingMethod(
+ getDescriptor().getMethods().get(0),
+ controller,
+ request,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EmptyResponseProto.getDefaultInstance());
+ }
+
+
+ public org.apache.hadoop.ipc.protobuf.TestProtos.EchoResponseProto echo2(
+ com.google.protobuf.RpcController controller,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EchoRequestProto request)
+ throws com.google.protobuf.ServiceException {
+ return (org.apache.hadoop.ipc.protobuf.TestProtos.EchoResponseProto) channel.callBlockingMethod(
+ getDescriptor().getMethods().get(1),
+ controller,
+ request,
+ org.apache.hadoop.ipc.protobuf.TestProtos.EchoResponseProto.getDefaultInstance());
+ }
+
+ }
+ }
+
public static com.google.protobuf.Descriptors.FileDescriptor
getDescriptor() {
@@ -373,8 +659,11 @@ public org.apache.hadoop.ipc.protobuf.TestProtos.EmptyResponseProto error(
"uestProto\032\023.EmptyResponseProto\022-\n\004echo\022\021" +
".EchoRequestProto\032\022.EchoResponseProto\0220\n" +
"\005error\022\022.EmptyRequestProto\032\023.EmptyRespon" +
- "seProtoB<\n\036org.apache.hadoop.ipc.protobu" +
- "fB\024TestRpcServiceProtos\210\001\001\240\001\001"
+ "seProto2y\n\025TestProtobufRpc2Proto\0220\n\005ping" +
+ "2\022\022.EmptyRequestProto\032\023.EmptyResponsePro" +
+ "to\022.\n\005echo2\022\021.EchoRequestProto\032\022.EchoRes" +
+ "ponseProtoB<\n\036org.apache.hadoop.ipc.prot" +
+ "obufB\024TestRpcServiceProtos\210\001\001\240\001\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
diff --git a/hadoop-common-project/hadoop-common/src/test/proto/test_rpc_service.proto b/hadoop-common-project/hadoop-common/src/test/proto/test_rpc_service.proto
index 14ba0ae170..3dcffe9d09 100644
--- a/hadoop-common-project/hadoop-common/src/test/proto/test_rpc_service.proto
+++ b/hadoop-common-project/hadoop-common/src/test/proto/test_rpc_service.proto
@@ -31,3 +31,8 @@ service TestProtobufRpcProto {
rpc echo(EchoRequestProto) returns (EchoResponseProto);
rpc error(EmptyRequestProto) returns (EmptyResponseProto);
}
+
+service TestProtobufRpc2Proto {
+ rpc ping2(EmptyRequestProto) returns (EmptyResponseProto);
+ rpc echo2(EchoRequestProto) returns (EchoResponseProto);
+}
diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
index d78b79c12b..7bafb1f44d 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
@@ -68,6 +68,10 @@ Trunk (unreleased changes)
HDFS-2410. Further cleanup of hardcoded configuration keys and values.
(suresh)
+ HADOOP-7862 Hdfs changes to work with HADOOP 7862:
+ Move the support for multiple protocols to lower layer so that Writable,
+ PB and Avro can all use it (Sanjay)
+
OPTIMIZATIONS
HDFS-2477. Optimize computing the diff between a block report and the
namenode state. (Tomasz Nykiel via hairong)
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
index b2c974c28b..d86ad92f36 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
@@ -142,6 +142,7 @@
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.ipc.Server;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
import org.apache.hadoop.metrics2.util.MBeans;
import org.apache.hadoop.net.DNS;
@@ -394,7 +395,7 @@ private void initIpcServer(Configuration conf) throws IOException {
InterDatanodeProtocolServerSideTranslatorR23
interDatanodeProtocolServerTranslator =
new InterDatanodeProtocolServerSideTranslatorR23(this);
- ipcServer.addProtocol(InterDatanodeWireProtocol.class,
+ ipcServer.addProtocol(RpcKind.RPC_WRITABLE, InterDatanodeWireProtocol.class,
interDatanodeProtocolServerTranslator);
// set service-level authorization security policy
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/BackupNode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/BackupNode.java
index c26f84a1ed..a833f7fc53 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/BackupNode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/BackupNode.java
@@ -40,6 +40,7 @@
import org.apache.hadoop.hdfs.server.protocol.NamenodeRegistration;
import org.apache.hadoop.hdfs.server.protocol.NamespaceInfo;
import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.net.NetUtils;
/**
@@ -211,7 +212,7 @@ private BackupNodeRpcServer(Configuration conf, BackupNode nn)
super(conf, nn);
JournalProtocolServerSideTranslatorR23 journalProtocolTranslator =
new JournalProtocolServerSideTranslatorR23(this);
- this.clientRpcServer.addProtocol(JournalWireProtocol.class,
+ this.clientRpcServer.addProtocol(RpcKind.RPC_WRITABLE, JournalWireProtocol.class,
journalProtocolTranslator);
nnRpcAddress = nn.nnRpcAddress;
}
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
index d6ba4175c1..f005f8b9d4 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java
@@ -90,6 +90,7 @@
import org.apache.hadoop.ipc.ProtocolSignature;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.Server;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.net.Node;
import org.apache.hadoop.security.AccessControlException;
import org.apache.hadoop.security.Groups;
@@ -152,13 +153,16 @@ public NameNodeRpcServer(Configuration conf, NameNode nn)
dnSocketAddr.getHostName(), dnSocketAddr.getPort(),
serviceHandlerCount,
false, conf, namesystem.getDelegationTokenSecretManager());
- this.serviceRpcServer.addProtocol(DatanodeProtocol.class, this);
- this.serviceRpcServer.addProtocol(NamenodeProtocol.class, this);
- this.serviceRpcServer.addProtocol(
+ this.serviceRpcServer.addProtocol(RpcKind.RPC_WRITABLE,
+ DatanodeProtocol.class, this);
+ this.serviceRpcServer.addProtocol(RpcKind.RPC_WRITABLE,
+ NamenodeProtocol.class, this);
+ this.serviceRpcServer.addProtocol(RpcKind.RPC_WRITABLE,
RefreshAuthorizationPolicyProtocol.class, this);
- this.serviceRpcServer.addProtocol(
+ this.serviceRpcServer.addProtocol(RpcKind.RPC_WRITABLE,
RefreshUserMappingsProtocol.class, this);
- this.serviceRpcServer.addProtocol(GetUserMappingsProtocol.class, this);
+ this.serviceRpcServer.addProtocol(RpcKind.RPC_WRITABLE,
+ GetUserMappingsProtocol.class, this);
this.serviceRpcServer.addProtocol(HAServiceProtocol.class, this);
this.serviceRPCAddress = this.serviceRpcServer.getListenerAddress();
@@ -174,12 +178,16 @@ public NameNodeRpcServer(Configuration conf, NameNode nn)
clientProtocolServerTranslator, socAddr.getHostName(),
socAddr.getPort(), handlerCount, false, conf,
namesystem.getDelegationTokenSecretManager());
- this.clientRpcServer.addProtocol(DatanodeProtocol.class, this);
- this.clientRpcServer.addProtocol(NamenodeProtocol.class, this);
- this.clientRpcServer.addProtocol(
+ this.clientRpcServer.addProtocol(RpcKind.RPC_WRITABLE,
+ DatanodeProtocol.class, this);
+ this.clientRpcServer.addProtocol(RpcKind.RPC_WRITABLE,
+ NamenodeProtocol.class, this);
+ this.clientRpcServer.addProtocol(RpcKind.RPC_WRITABLE,
RefreshAuthorizationPolicyProtocol.class, this);
- this.clientRpcServer.addProtocol(RefreshUserMappingsProtocol.class, this);
- this.clientRpcServer.addProtocol(GetUserMappingsProtocol.class, this);
+ this.clientRpcServer.addProtocol(RpcKind.RPC_WRITABLE,
+ RefreshUserMappingsProtocol.class, this);
+ this.clientRpcServer.addProtocol(RpcKind.RPC_WRITABLE,
+ GetUserMappingsProtocol.class, this);
this.clientRpcServer.addProtocol(HAServiceProtocol.class, this);
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSClientRetries.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSClientRetries.java
index af12badddb..be57bd4e39 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSClientRetries.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSClientRetries.java
@@ -62,6 +62,7 @@
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.Server;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.net.NetUtils;
import org.mockito.internal.stubbing.answers.ThrowsException;
import org.mockito.invocation.InvocationOnMock;
@@ -97,7 +98,7 @@ public TestServer(int handlerCount, boolean sleep,
}
@Override
- public Writable call(String protocol, Writable param, long receiveTime)
+ public Writable call(RpcKind rpcKind, String protocol, Writable param, long receiveTime)
throws IOException {
if (sleep) {
// sleep a bit
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestInterDatanodeProtocol.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestInterDatanodeProtocol.java
index d50376aaff..5f48651c89 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestInterDatanodeProtocol.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestInterDatanodeProtocol.java
@@ -28,6 +28,7 @@
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.ipc.Server;
import org.apache.hadoop.net.NetUtils;
@@ -81,7 +82,7 @@ public TestServer(int handlerCount, boolean sleep,
}
@Override
- public Writable call(String protocol, Writable param, long receiveTime)
+ public Writable call(RpcKind rpcKind, String protocol, Writable param, long receiveTime)
throws IOException {
if (sleep) {
// sleep a bit
diff --git a/hadoop-mapreduce-project/CHANGES.txt b/hadoop-mapreduce-project/CHANGES.txt
index 70a2001f38..fbdcba77da 100644
--- a/hadoop-mapreduce-project/CHANGES.txt
+++ b/hadoop-mapreduce-project/CHANGES.txt
@@ -35,11 +35,15 @@ Trunk (unreleased changes)
uri with no authority. (John George via jitendra)
MAPREDUCE-3169. Create a new MiniMRCluster equivalent which only provides
- client APIs cross MR1 and MR2. (Ahmed via tucu)
+ client APIs cross MR1 and MR2 (Ahmed via tucu)
MAPREDUCE-3415. improve MiniMRYarnCluster & DistributedShell JAR resolution.
(tucu)
+ HADOOP-7862 MR changes to work with HADOOP 7862:
+ Move the support for multiple protocols to lower layer so that Writable,
+ PB and Avro can all use it (Sanjay)
+
BUG FIXES
MAPREDUCE-3412. Fix 'ant docs'. (amarrk)
@@ -68,6 +72,8 @@ Trunk (unreleased changes)
MAPREDUCE-3477. Hadoop site documentation cannot be built anymore. (jeagles via tucu)
+ MAPREDUCE-3500. MRJobConfig creates an LD_LIBRARY_PATH using the platform ARCH. (tucu)
+
Release 0.23.1 - Unreleased
INCOMPATIBLE CHANGES
@@ -129,6 +135,12 @@ Release 0.23.1 - Unreleased
MAPREDUCE-3448. TestCombineOutputCollector javac unchecked warning on mocked
generics (Jonathan Eagles via mahadev)
+ MAPREDUCE-3169 amendment. Deprecate MiniMRCluster. (Ahmed Radwan via
+ sseth)
+
+ MAPREDUCE-3369. Migrate MR1 tests to run on MR2 using the new interfaces
+ introduced in MAPREDUCE-3169. (Ahmed Radwan via tomwhite)
+
OPTIMIZATIONS
BUG FIXES
@@ -211,6 +223,26 @@ Release 0.23.1 - Unreleased
MAPREDUCE-3452. fifoscheduler web ui page always shows 0% used for the queue.
(Jonathan Eagles via mahadev)
+ MAPREDUCE-3443. JobClient and Job should function in the context of the
+ UGI which created them. (Mahadev Konar via sseth)
+
+ MAPREDUCE-3460. MR AM can hang if containers are allocated on a node
+ blacklisted by the AM. (Hitesh Shah and Robert Joseph Evans via sseth)
+
+ MAPREDUCE-3453. RM web ui application details page shows RM cluster about
+ information. (Jonathan Eagles via sseth)
+
+ MAPREDUCE-3479. JobClient#getJob cannot find local jobs. (tomwhite)
+
+ MAPREDUCE-3456. $HADOOP_PREFIX/bin/yarn should set defaults for
+ $HADOOP_*_HOME (Eric Payne via mahadev)
+
+ MAPREDUCE-3458. Fix findbugs warnings in hadoop-examples. (Devaraj K
+ via mahadev)
+
+ MAPREDUCE-3485. DISKS_FAILED -101 error code should be defined in same location as
+ ABORTED_CONTAINER_EXIT_STATUS. (Ravi Gummadi via mahadev)
+
Release 0.23.0 - 2011-11-01
INCOMPATIBLE CHANGES
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerAllocator.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerAllocator.java
index e8588e5cd0..81a5a75b50 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerAllocator.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerAllocator.java
@@ -87,7 +87,7 @@ public class RMContainerAllocator extends RMContainerRequestor
}
/*
- Vocabulory Used:
+ Vocabulary Used:
pending -> requests which are NOT yet sent to RM
scheduled -> requests which are sent to RM but not yet assigned
assigned -> requests which are assigned to a container
@@ -565,6 +565,7 @@ void addMap(ContainerRequestEvent event) {
if (event.getEarlierAttemptFailed()) {
earlierFailedMaps.add(event.getAttemptID());
request = new ContainerRequest(event, PRIORITY_FAST_FAIL_MAP);
+ LOG.info("Added "+event.getAttemptID()+" to list of failed maps");
} else {
for (String host : event.getHosts()) {
LinkedList list = mapsHostMapping.get(host);
@@ -603,7 +604,9 @@ private void assign(List allocatedContainers) {
containersAllocated += allocatedContainers.size();
while (it.hasNext()) {
Container allocated = it.next();
- LOG.info("Assigning container " + allocated);
+ LOG.info("Assigning container " + allocated.getId() +
+ " with priority " + allocated.getPriority() +
+ " to NM " + allocated.getNodeId());
// check if allocated container meets memory requirements
// and whether we have any scheduled tasks that need
@@ -645,7 +648,8 @@ else if (PRIORITY_REDUCE.equals(priority)) {
// we need to request for a new container
// and release the current one
LOG.info("Got allocated container on a blacklisted "
- + " host. Releasing container " + allocated);
+ + " host "+allocated.getNodeId().getHost()
+ +". Releasing container " + allocated);
// find the request matching this allocated container
// and replace it with a new one
@@ -727,10 +731,20 @@ private ContainerRequest assign(Container allocated) {
}
private ContainerRequest getContainerReqToReplace(Container allocated) {
+ LOG.info("Finding containerReq for allocated container: " + allocated);
Priority priority = allocated.getPriority();
ContainerRequest toBeReplaced = null;
- if (PRIORITY_FAST_FAIL_MAP.equals(priority)
- || PRIORITY_MAP.equals(priority)) {
+ if (PRIORITY_FAST_FAIL_MAP.equals(priority)) {
+ LOG.info("Replacing FAST_FAIL_MAP container " + allocated.getId());
+ Iterator iter = earlierFailedMaps.iterator();
+ while (toBeReplaced == null && iter.hasNext()) {
+ toBeReplaced = maps.get(iter.next());
+ }
+ LOG.info("Found replacement: " + toBeReplaced);
+ return toBeReplaced;
+ }
+ else if (PRIORITY_MAP.equals(priority)) {
+ LOG.info("Replacing MAP container " + allocated.getId());
// allocated container was for a map
String host = allocated.getNodeId().getHost();
LinkedList list = mapsHostMapping.get(host);
@@ -749,6 +763,7 @@ else if (PRIORITY_REDUCE.equals(priority)) {
TaskAttemptId tId = reduces.keySet().iterator().next();
toBeReplaced = reduces.remove(tId);
}
+ LOG.info("Found replacement: " + toBeReplaced);
return toBeReplaced;
}
@@ -758,7 +773,7 @@ private ContainerRequest assignToFailedMap(Container allocated) {
//try to assign to earlierFailedMaps if present
ContainerRequest assigned = null;
while (assigned == null && earlierFailedMaps.size() > 0) {
- TaskAttemptId tId = earlierFailedMaps.removeFirst();
+ TaskAttemptId tId = earlierFailedMaps.removeFirst();
if (maps.containsKey(tId)) {
assigned = maps.remove(tId);
JobCounterUpdateEvent jce =
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerRequestor.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerRequestor.java
index ba3c73219d..6c03c6690c 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerRequestor.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMContainerRequestor.java
@@ -105,6 +105,13 @@ public ContainerRequest(TaskAttemptId attemptID,
this.priority = priority;
}
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("AttemptId[").append(attemptID).append("]");
+ sb.append("Capability[").append(capability).append("]");
+ sb.append("Priority[").append(priority).append("]");
+ return sb.toString();
+ }
}
@Override
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestRMContainerAllocator.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestRMContainerAllocator.java
index 9dd877b330..812393c1b5 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestRMContainerAllocator.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestRMContainerAllocator.java
@@ -580,6 +580,135 @@ public void testBlackListedNodes() throws Exception {
}
}
+ @Test
+ public void testBlackListedNodesWithSchedulingToThatNode() throws Exception {
+ LOG.info("Running testBlackListedNodesWithSchedulingToThatNode");
+
+ Configuration conf = new Configuration();
+ conf.setBoolean(MRJobConfig.MR_AM_JOB_NODE_BLACKLISTING_ENABLE, true);
+ conf.setInt(MRJobConfig.MAX_TASK_FAILURES_PER_TRACKER, 1);
+
+ MyResourceManager rm = new MyResourceManager(conf);
+ rm.start();
+ DrainDispatcher dispatcher = (DrainDispatcher) rm.getRMContext()
+ .getDispatcher();
+
+ // Submit the application
+ RMApp app = rm.submitApp(1024);
+ dispatcher.await();
+
+ MockNM amNodeManager = rm.registerNode("amNM:1234", 2048);
+ amNodeManager.nodeHeartbeat(true);
+ dispatcher.await();
+
+ ApplicationAttemptId appAttemptId = app.getCurrentAppAttempt()
+ .getAppAttemptId();
+ rm.sendAMLaunched(appAttemptId);
+ dispatcher.await();
+
+ JobId jobId = MRBuilderUtils.newJobId(appAttemptId.getApplicationId(), 0);
+ Job mockJob = mock(Job.class);
+ when(mockJob.getReport()).thenReturn(
+ MRBuilderUtils.newJobReport(jobId, "job", "user", JobState.RUNNING, 0,
+ 0, 0, 0, 0, 0, 0, "jobfile", null));
+ MyContainerAllocator allocator = new MyContainerAllocator(rm, conf,
+ appAttemptId, mockJob);
+
+ // add resources to scheduler
+ MockNM nodeManager1 = rm.registerNode("h1:1234", 10240);
+ MockNM nodeManager3 = rm.registerNode("h3:1234", 10240);
+ dispatcher.await();
+
+ LOG.info("Requesting 1 Containers _1 on H1");
+ // create the container request
+ ContainerRequestEvent event1 = createReq(jobId, 1, 1024,
+ new String[] { "h1" });
+ allocator.sendRequest(event1);
+
+ LOG.info("RM Heartbeat (to send the container requests)");
+ // this tells the scheduler about the requests
+ // as nodes are not added, no allocations
+ List assigned = allocator.schedule();
+ dispatcher.await();
+ Assert.assertEquals("No of assignments must be 0", 0, assigned.size());
+
+ LOG.info("h1 Heartbeat (To actually schedule the containers)");
+ // update resources in scheduler
+ nodeManager1.nodeHeartbeat(true); // Node heartbeat
+ dispatcher.await();
+
+ LOG.info("RM Heartbeat (To process the scheduled containers)");
+ assigned = allocator.schedule();
+ dispatcher.await();
+ Assert.assertEquals("No of assignments must be 1", 1, assigned.size());
+
+ LOG.info("Failing container _1 on H1 (should blacklist the node)");
+ // Send events to blacklist nodes h1 and h2
+ ContainerFailedEvent f1 = createFailEvent(jobId, 1, "h1", false);
+ allocator.sendFailure(f1);
+
+ //At this stage, a request should be created for a fast fail map
+ //Create a FAST_FAIL request for a previously failed map.
+ ContainerRequestEvent event1f = createReq(jobId, 1, 1024,
+ new String[] { "h1" }, true, false);
+ allocator.sendRequest(event1f);
+
+ //Update the Scheduler with the new requests.
+ assigned = allocator.schedule();
+ dispatcher.await();
+ Assert.assertEquals("No of assignments must be 0", 0, assigned.size());
+
+ // send another request with different resource and priority
+ ContainerRequestEvent event3 = createReq(jobId, 3, 1024,
+ new String[] { "h1", "h3" });
+ allocator.sendRequest(event3);
+
+ //Allocator is aware of prio:5 container, and prio:20 (h1+h3) container.
+ //RM is only aware of the prio:5 container
+
+ LOG.info("h1 Heartbeat (To actually schedule the containers)");
+ // update resources in scheduler
+ nodeManager1.nodeHeartbeat(true); // Node heartbeat
+ dispatcher.await();
+
+ LOG.info("RM Heartbeat (To process the scheduled containers)");
+ assigned = allocator.schedule();
+ dispatcher.await();
+ Assert.assertEquals("No of assignments must be 0", 0, assigned.size());
+
+ //RMContainerAllocator gets assigned a p:5 on a blacklisted node.
+
+ //Send a release for the p:5 container + another request.
+ LOG.info("RM Heartbeat (To process the re-scheduled containers)");
+ assigned = allocator.schedule();
+ dispatcher.await();
+ Assert.assertEquals("No of assignments must be 0", 0, assigned.size());
+
+ //Hearbeat from H3 to schedule on this host.
+ LOG.info("h3 Heartbeat (To re-schedule the containers)");
+ nodeManager3.nodeHeartbeat(true); // Node heartbeat
+ dispatcher.await();
+
+ LOG.info("RM Heartbeat (To process the re-scheduled containers for H3)");
+ assigned = allocator.schedule();
+ dispatcher.await();
+
+ // For debugging
+ for (TaskAttemptContainerAssignedEvent assig : assigned) {
+ LOG.info(assig.getTaskAttemptID() +
+ " assgined to " + assig.getContainer().getId() +
+ " with priority " + assig.getContainer().getPriority());
+ }
+
+ Assert.assertEquals("No of assignments must be 2", 2, assigned.size());
+
+ // validate that all containers are assigned to h3
+ for (TaskAttemptContainerAssignedEvent assig : assigned) {
+ Assert.assertEquals("Assigned container " + assig.getContainer().getId()
+ + " host not correct", "h3", assig.getContainer().getNodeId().getHost());
+ }
+ }
+
private static class MyFifoScheduler extends FifoScheduler {
public MyFifoScheduler(RMContext rmContext) {
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/test/java/org/apache/hadoop/mapred/TestJobClientGetJob.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/test/java/org/apache/hadoop/mapred/TestJobClientGetJob.java
new file mode 100644
index 0000000000..5ba5f25a63
--- /dev/null
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/test/java/org/apache/hadoop/mapred/TestJobClientGetJob.java
@@ -0,0 +1,62 @@
+/**
+ * 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.mapred;
+
+import static junit.framework.Assert.assertNotNull;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.junit.Test;
+
+public class TestJobClientGetJob {
+
+ private static Path TEST_ROOT_DIR =
+ new Path(System.getProperty("test.build.data","/tmp"));
+
+ private Path createTempFile(String filename, String contents)
+ throws IOException {
+ Path path = new Path(TEST_ROOT_DIR, filename);
+ Configuration conf = new Configuration();
+ FSDataOutputStream os = FileSystem.getLocal(conf).create(path);
+ os.writeBytes(contents);
+ os.close();
+ return path;
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testGetRunningJobFromJobClient() throws Exception {
+ JobConf conf = new JobConf();
+ conf.set("mapreduce.framework.name", "local");
+ FileInputFormat.addInputPath(conf, createTempFile("in", "hello"));
+ FileOutputFormat.setOutputPath(conf,
+ new Path(TEST_ROOT_DIR, getClass().getSimpleName()));
+ JobClient jc = new JobClient(conf);
+ RunningJob runningJob = jc.submitJob(conf);
+ assertNotNull("Running job", runningJob);
+ // Check that the running job can be retrieved by ID
+ RunningJob newRunningJob = jc.getJob(runningJob.getID());
+ assertNotNull("New running job", newRunningJob);
+ }
+
+}
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JobClient.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JobClient.java
index 0505e33ce5..1b0b453ddd 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JobClient.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/JobClient.java
@@ -21,6 +21,7 @@
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URL;
+import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
@@ -42,6 +43,7 @@
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.security.token.TokenRenewer;
import org.apache.hadoop.security.token.SecretManager.InvalidToken;
@@ -421,6 +423,11 @@ boolean monitorAndPrintJob() throws IOException, InterruptedException {
}
Cluster cluster;
+ /**
+ * Ugi of the client. We store this ugi when the client is created and
+ * then make sure that the same ugi is used to run the various protocols.
+ */
+ UserGroupInformation clientUgi;
/**
* Create a job client.
@@ -458,6 +465,7 @@ public JobClient(Configuration conf) throws IOException {
public void init(JobConf conf) throws IOException {
setConf(conf);
cluster = new Cluster(conf);
+ clientUgi = UserGroupInformation.getCurrentUser();
}
@InterfaceAudience.Private
@@ -487,8 +495,7 @@ public void cancel(Token token, Configuration conf
@Override
public boolean isManaged(Token token) throws IOException {
return true;
- }
-
+ }
}
/**
@@ -500,6 +507,7 @@ public boolean isManaged(Token token) throws IOException {
public JobClient(InetSocketAddress jobTrackAddr,
Configuration conf) throws IOException {
cluster = new Cluster(jobTrackAddr, conf);
+ clientUgi = UserGroupInformation.getCurrentUser();
}
/**
@@ -562,21 +570,38 @@ public RunningJob submitJob(String jobFile) throws FileNotFoundException,
* @throws FileNotFoundException
* @throws IOException
*/
- public RunningJob submitJob(JobConf conf) throws FileNotFoundException,
+ public RunningJob submitJob(final JobConf conf) throws FileNotFoundException,
IOException {
try {
conf.setBooleanIfUnset("mapred.mapper.new-api", false);
conf.setBooleanIfUnset("mapred.reducer.new-api", false);
- Job job = Job.getInstance(conf);
- job.submit();
+ Job job = clientUgi.doAs(new PrivilegedExceptionAction () {
+ @Override
+ public Job run() throws IOException, ClassNotFoundException,
+ InterruptedException {
+ Job job = Job.getInstance(conf);
+ job.submit();
+ return job;
+ }
+ });
+ // update our Cluster instance with the one created by Job for submission
+ // (we can't pass our Cluster instance to Job, since Job wraps the config
+ // instance, and the two configs would then diverge)
+ cluster = job.getCluster();
return new NetworkedJob(job);
} catch (InterruptedException ie) {
throw new IOException("interrupted", ie);
- } catch (ClassNotFoundException cnfe) {
- throw new IOException("class not found", cnfe);
}
}
+ private Job getJobUsingCluster(final JobID jobid) throws IOException,
+ InterruptedException {
+ return clientUgi.doAs(new PrivilegedExceptionAction() {
+ public Job run() throws IOException, InterruptedException {
+ return cluster.getJob(jobid);
+ }
+ });
+ }
/**
* Get an {@link RunningJob} object to track an ongoing job. Returns
* null if the id does not correspond to any known job.
@@ -586,9 +611,10 @@ public RunningJob submitJob(JobConf conf) throws FileNotFoundException,
* jobid doesn't correspond to any known job.
* @throws IOException
*/
- public RunningJob getJob(JobID jobid) throws IOException {
+ public RunningJob getJob(final JobID jobid) throws IOException {
try {
- Job job = cluster.getJob(jobid);
+
+ Job job = getJobUsingCluster(jobid);
if (job != null) {
JobStatus status = JobStatus.downgrade(job.getStatus());
if (status != null) {
@@ -621,9 +647,10 @@ public TaskReport[] getMapTaskReports(JobID jobId) throws IOException {
return getTaskReports(jobId, TaskType.MAP);
}
- private TaskReport[] getTaskReports(JobID jobId, TaskType type) throws IOException {
+ private TaskReport[] getTaskReports(final JobID jobId, TaskType type) throws
+ IOException {
try {
- Job j = cluster.getJob(jobId);
+ Job j = getJobUsingCluster(jobId);
if(j == null) {
return EMPTY_TASK_REPORTS;
}
@@ -688,10 +715,11 @@ public TaskReport[] getReduceTaskReports(String jobId) throws IOException {
* @param state the state of the task
* (pending/running/completed/failed/killed)
*/
- public void displayTasks(JobID jobId, String type, String state)
+ public void displayTasks(final JobID jobId, String type, String state)
throws IOException {
try {
- super.displayTasks(cluster.getJob(jobId), type, state);
+ Job job = getJobUsingCluster(jobId);
+ super.displayTasks(job, type, state);
} catch (InterruptedException ie) {
throw new IOException(ie);
}
@@ -706,15 +734,20 @@ public void displayTasks(JobID jobId, String type, String state)
*/
public ClusterStatus getClusterStatus() throws IOException {
try {
- ClusterMetrics metrics = cluster.getClusterStatus();
- return new ClusterStatus(metrics.getTaskTrackerCount(),
- metrics.getBlackListedTaskTrackerCount(), cluster.getTaskTrackerExpiryInterval(),
- metrics.getOccupiedMapSlots(),
- metrics.getOccupiedReduceSlots(), metrics.getMapSlotCapacity(),
- metrics.getReduceSlotCapacity(),
- cluster.getJobTrackerStatus(),
- metrics.getDecommissionedTaskTrackerCount());
- } catch (InterruptedException ie) {
+ return clientUgi.doAs(new PrivilegedExceptionAction() {
+ public ClusterStatus run() throws IOException, InterruptedException {
+ ClusterMetrics metrics = cluster.getClusterStatus();
+ return new ClusterStatus(metrics.getTaskTrackerCount(),
+ metrics.getBlackListedTaskTrackerCount(), cluster.getTaskTrackerExpiryInterval(),
+ metrics.getOccupiedMapSlots(),
+ metrics.getOccupiedReduceSlots(), metrics.getMapSlotCapacity(),
+ metrics.getReduceSlotCapacity(),
+ cluster.getJobTrackerStatus(),
+ metrics.getDecommissionedTaskTrackerCount());
+ }
+ });
+ }
+ catch (InterruptedException ie) {
throw new IOException(ie);
}
}
@@ -750,13 +783,17 @@ private Collection arrayToBlackListInfo(TaskTrackerInfo[] objs)
*/
public ClusterStatus getClusterStatus(boolean detailed) throws IOException {
try {
- ClusterMetrics metrics = cluster.getClusterStatus();
- return new ClusterStatus(arrayToStringList(cluster.getActiveTaskTrackers()),
- arrayToBlackListInfo(cluster.getBlackListedTaskTrackers()),
- cluster.getTaskTrackerExpiryInterval(), metrics.getOccupiedMapSlots(),
- metrics.getOccupiedReduceSlots(), metrics.getMapSlotCapacity(),
- metrics.getReduceSlotCapacity(),
- cluster.getJobTrackerStatus());
+ return clientUgi.doAs(new PrivilegedExceptionAction() {
+ public ClusterStatus run() throws IOException, InterruptedException {
+ ClusterMetrics metrics = cluster.getClusterStatus();
+ return new ClusterStatus(arrayToStringList(cluster.getActiveTaskTrackers()),
+ arrayToBlackListInfo(cluster.getBlackListedTaskTrackers()),
+ cluster.getTaskTrackerExpiryInterval(), metrics.getOccupiedMapSlots(),
+ metrics.getOccupiedReduceSlots(), metrics.getMapSlotCapacity(),
+ metrics.getReduceSlotCapacity(),
+ cluster.getJobTrackerStatus());
+ }
+ });
} catch (InterruptedException ie) {
throw new IOException(ie);
}
@@ -787,7 +824,14 @@ public JobStatus[] jobsToComplete() throws IOException {
*/
public JobStatus[] getAllJobs() throws IOException {
try {
- org.apache.hadoop.mapreduce.JobStatus[] jobs = cluster.getAllJobStatuses();
+ org.apache.hadoop.mapreduce.JobStatus[] jobs =
+ clientUgi.doAs(new PrivilegedExceptionAction<
+ org.apache.hadoop.mapreduce.JobStatus[]> () {
+ public org.apache.hadoop.mapreduce.JobStatus[] run()
+ throws IOException, InterruptedException {
+ return cluster.getAllJobStatuses();
+ }
+ });
JobStatus[] stats = new JobStatus[jobs.length];
for (int i = 0; i < jobs.length; i++) {
stats[i] = JobStatus.downgrade(jobs[i]);
@@ -909,7 +953,12 @@ protected long getCounter(org.apache.hadoop.mapreduce.Counters cntrs,
*/
public int getDefaultMaps() throws IOException {
try {
- return cluster.getClusterStatus().getMapSlotCapacity();
+ return clientUgi.doAs(new PrivilegedExceptionAction() {
+ @Override
+ public Integer run() throws IOException, InterruptedException {
+ return cluster.getClusterStatus().getMapSlotCapacity();
+ }
+ });
} catch (InterruptedException ie) {
throw new IOException(ie);
}
@@ -923,7 +972,12 @@ public int getDefaultMaps() throws IOException {
*/
public int getDefaultReduces() throws IOException {
try {
- return cluster.getClusterStatus().getReduceSlotCapacity();
+ return clientUgi.doAs(new PrivilegedExceptionAction() {
+ @Override
+ public Integer run() throws IOException, InterruptedException {
+ return cluster.getClusterStatus().getReduceSlotCapacity();
+ }
+ });
} catch (InterruptedException ie) {
throw new IOException(ie);
}
@@ -936,8 +990,13 @@ public int getDefaultReduces() throws IOException {
*/
public Path getSystemDir() {
try {
- return cluster.getSystemDir();
- } catch (IOException ioe) {
+ return clientUgi.doAs(new PrivilegedExceptionAction() {
+ @Override
+ public Path run() throws IOException, InterruptedException {
+ return cluster.getSystemDir();
+ }
+ });
+ } catch (IOException ioe) {
return null;
} catch (InterruptedException ie) {
return null;
@@ -962,7 +1021,11 @@ private JobQueueInfo[] getJobQueueInfoArray(QueueInfo[] queues)
*/
public JobQueueInfo[] getRootQueues() throws IOException {
try {
- return getJobQueueInfoArray(cluster.getRootQueues());
+ return clientUgi.doAs(new PrivilegedExceptionAction() {
+ public JobQueueInfo[] run() throws IOException, InterruptedException {
+ return getJobQueueInfoArray(cluster.getRootQueues());
+ }
+ });
} catch (InterruptedException ie) {
throw new IOException(ie);
}
@@ -976,9 +1039,13 @@ public JobQueueInfo[] getRootQueues() throws IOException {
* @return the array of immediate children JobQueueInfo objects
* @throws IOException
*/
- public JobQueueInfo[] getChildQueues(String queueName) throws IOException {
+ public JobQueueInfo[] getChildQueues(final String queueName) throws IOException {
try {
- return getJobQueueInfoArray(cluster.getChildQueues(queueName));
+ return clientUgi.doAs(new PrivilegedExceptionAction() {
+ public JobQueueInfo[] run() throws IOException, InterruptedException {
+ return getJobQueueInfoArray(cluster.getChildQueues(queueName));
+ }
+ });
} catch (InterruptedException ie) {
throw new IOException(ie);
}
@@ -993,7 +1060,11 @@ public JobQueueInfo[] getChildQueues(String queueName) throws IOException {
*/
public JobQueueInfo[] getQueues() throws IOException {
try {
- return getJobQueueInfoArray(cluster.getQueues());
+ return clientUgi.doAs(new PrivilegedExceptionAction() {
+ public JobQueueInfo[] run() throws IOException, InterruptedException {
+ return getJobQueueInfoArray(cluster.getQueues());
+ }
+ });
} catch (InterruptedException ie) {
throw new IOException(ie);
}
@@ -1007,9 +1078,14 @@ public JobQueueInfo[] getQueues() throws IOException {
* @throws IOException
*/
- public JobStatus[] getJobsFromQueue(String queueName) throws IOException {
+ public JobStatus[] getJobsFromQueue(final String queueName) throws IOException {
try {
- QueueInfo queue = cluster.getQueue(queueName);
+ QueueInfo queue = clientUgi.doAs(new PrivilegedExceptionAction() {
+ @Override
+ public QueueInfo run() throws IOException, InterruptedException {
+ return cluster.getQueue(queueName);
+ }
+ });
if (queue == null) {
return null;
}
@@ -1032,9 +1108,14 @@ public JobStatus[] getJobsFromQueue(String queueName) throws IOException {
* @return Queue information associated to particular queue.
* @throws IOException
*/
- public JobQueueInfo getQueueInfo(String queueName) throws IOException {
+ public JobQueueInfo getQueueInfo(final String queueName) throws IOException {
try {
- QueueInfo queueInfo = cluster.getQueue(queueName);
+ QueueInfo queueInfo = clientUgi.doAs(new
+ PrivilegedExceptionAction() {
+ public QueueInfo run() throws IOException, InterruptedException {
+ return cluster.getQueue(queueName);
+ }
+ });
if (queueInfo != null) {
return new JobQueueInfo(queueInfo);
}
@@ -1052,7 +1133,14 @@ public JobQueueInfo getQueueInfo(String queueName) throws IOException {
public QueueAclsInfo[] getQueueAclsForCurrentUser() throws IOException {
try {
org.apache.hadoop.mapreduce.QueueAclsInfo[] acls =
- cluster.getQueueAclsForCurrentUser();
+ clientUgi.doAs(new
+ PrivilegedExceptionAction
+ () {
+ public org.apache.hadoop.mapreduce.QueueAclsInfo[] run()
+ throws IOException, InterruptedException {
+ return cluster.getQueueAclsForCurrentUser();
+ }
+ });
QueueAclsInfo[] ret = new QueueAclsInfo[acls.length];
for (int i = 0 ; i < acls.length; i++ ) {
ret[i] = QueueAclsInfo.downgrade(acls[i]);
@@ -1070,8 +1158,14 @@ public QueueAclsInfo[] getQueueAclsForCurrentUser() throws IOException {
* @throws IOException
*/
public Token
- getDelegationToken(Text renewer) throws IOException, InterruptedException {
- return cluster.getDelegationToken(renewer);
+ getDelegationToken(final Text renewer) throws IOException, InterruptedException {
+ return clientUgi.doAs(new
+ PrivilegedExceptionAction>() {
+ public Token run() throws IOException,
+ InterruptedException {
+ return cluster.getDelegationToken(renewer);
+ }
+ });
}
/**
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/Job.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/Job.java
index 530aee1999..5e92baa8b6 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/Job.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/Job.java
@@ -30,6 +30,7 @@
import java.net.URLConnection;
import java.net.URI;
import java.security.PrivilegedExceptionAction;
+import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -315,7 +316,12 @@ synchronized void ensureFreshStatus()
* @throws IOException
*/
synchronized void updateStatus() throws IOException, InterruptedException {
- this.status = cluster.getClient().getJobStatus(status.getJobID());
+ this.status = ugi.doAs(new PrivilegedExceptionAction() {
+ @Override
+ public JobStatus run() throws IOException, InterruptedException {
+ return cluster.getClient().getJobStatus(status.getJobID());
+ }
+ });
if (this.status == null) {
throw new IOException("Job status not available ");
}
@@ -430,6 +436,11 @@ public boolean isRetired() throws IOException, InterruptedException {
updateStatus();
return status.isRetired();
}
+
+ @Private
+ public Cluster getCluster() {
+ return cluster;
+ }
/** Only for mocks in unit tests. */
@Private
@@ -476,8 +487,16 @@ String getTaskFailureEventString() throws IOException,
InterruptedException {
int failCount = 1;
TaskCompletionEvent lastEvent = null;
- for (TaskCompletionEvent event : cluster.getClient().getTaskCompletionEvents(
- status.getJobID(), 0, 10)) {
+ TaskCompletionEvent[] events = ugi.doAs(new
+ PrivilegedExceptionAction() {
+ @Override
+ public TaskCompletionEvent[] run() throws IOException,
+ InterruptedException {
+ return cluster.getClient().getTaskCompletionEvents(
+ status.getJobID(), 0, 10);
+ }
+ });
+ for (TaskCompletionEvent event : events) {
if (event.getStatus().equals(TaskCompletionEvent.Status.FAILED)) {
failCount++;
lastEvent = event;
@@ -500,7 +519,12 @@ String getTaskFailureEventString() throws IOException,
public TaskReport[] getTaskReports(TaskType type)
throws IOException, InterruptedException {
ensureState(JobState.RUNNING);
- return cluster.getClient().getTaskReports(getJobID(), type);
+ final TaskType tmpType = type;
+ return ugi.doAs(new PrivilegedExceptionAction() {
+ public TaskReport[] run() throws IOException, InterruptedException {
+ return cluster.getClient().getTaskReports(getJobID(), tmpType);
+ }
+ });
}
/**
@@ -603,7 +627,14 @@ public void setPriority(JobPriority priority)
org.apache.hadoop.mapred.JobPriority.valueOf(priority.name()));
} else {
ensureState(JobState.RUNNING);
- cluster.getClient().setJobPriority(getJobID(), priority.toString());
+ final JobPriority tmpPriority = priority;
+ ugi.doAs(new PrivilegedExceptionAction