You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by at...@apache.org on 2011/11/15 03:39:21 UTC
svn commit: r1202013 - in
/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common: ./
src/main/bin/ src/main/docs/ src/main/java/
src/main/java/org/apache/hadoop/http/ src/main/java/org/apache/hadoop/ipc/
src/main/packages/ src/main/packa...
Author: atm
Date: Tue Nov 15 02:39:13 2011
New Revision: 1202013
URL: http://svn.apache.org/viewvc?rev=1202013&view=rev
Log:
Merge trunk into HA branch
Added:
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcPayloadHeader.java
- copied unchanged from r1202009, hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcPayloadHeader.java
Modified:
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/CHANGES.txt (contents, props changed)
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/bin/hadoop-config.sh
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/docs/ (props changed)
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/ (props changed)
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/http/HttpServer.java
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-datanode
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-jobtracker
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-namenode
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-tasktracker
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/hadoop-setup-conf.sh
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/rpm/init.d/hadoop-jobtracker
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/rpm/init.d/hadoop-namenode
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/rpm/init.d/hadoop-tasktracker
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/test/core/ (props changed)
hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/http/TestServletFilter.java
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/CHANGES.txt?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/CHANGES.txt (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/CHANGES.txt Tue Nov 15 02:39:13 2011
@@ -55,6 +55,11 @@ Trunk (unreleased changes)
HADOOP-7792. Add verifyToken method to AbstractDelegationTokenSecretManager.
(jitendra)
+ HADOOP-7776 Make the Ipc-Header in a RPC-Payload an explicit header (sanjay)
+
+ HADOOP-7688. Add servlet handler check in HttpServer.start().
+ (Uma Maheswara Rao G via szetszwo)
+
BUGS
HADOOP-7606. Upgrade Jackson to version 1.7.1 to match the version required
@@ -95,6 +100,20 @@ Trunk (unreleased changes)
HADOOP-7773. Add support for protocol buffer based RPC engine.
(suresh)
+Release 0.23.1 - Unreleased
+
+ INCOMPATIBLE CHANGES
+
+ NEW FEATURES
+
+ IMPROVEMENTS
+
+ HADOOP-7801. HADOOP_PREFIX cannot be overriden. (Bruno Mahé via tomwhite)
+
+ OPTIMIZATIONS
+
+ BUG FIXES
+
Release 0.23.0 - 2011-11-01
INCOMPATIBLE CHANGES
@@ -798,6 +817,12 @@ Release 0.23.0 - 2011-11-01
HADOOP-7778. FindBugs warning in Token.getKind(). (tomwhite)
+ HADOOP-7798. Add support gpg signatures for maven release artifacts.
+ (cutting via acmurthy)
+
+ HADOOP-7797. Fix top-level pom.xml to refer to correct staging maven
+ repository. (omalley via acmurthy)
+
Release 0.22.0 - Unreleased
INCOMPATIBLE CHANGES
Propchange: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/CHANGES.txt
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Nov 15 02:39:13 2011
@@ -1,5 +1,5 @@
/hadoop/common/branches/yahoo-merge/CHANGES.txt:1079157,1079163-1079164,1079167
-/hadoop/common/trunk/hadoop-common-project/hadoop-common/CHANGES.txt:1161333-1196451
+/hadoop/common/trunk/hadoop-common-project/hadoop-common/CHANGES.txt:1161333-1202009
/hadoop/core/branches/branch-0.18/CHANGES.txt:727226
/hadoop/core/branches/branch-0.19/CHANGES.txt:713112
/hadoop/core/trunk/CHANGES.txt:776175-785643,785929-786278
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/bin/hadoop-config.sh
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/bin/hadoop-config.sh?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/bin/hadoop-config.sh (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/bin/hadoop-config.sh Tue Nov 15 02:39:13 2011
@@ -27,7 +27,9 @@ this="$common_bin/$script"
# the root of the Hadoop installation
# See HADOOP-6255 for directory structure layout
-export HADOOP_PREFIX=`dirname "$this"`/..
+HADOOP_DEFAULT_PREFIX=`dirname "$this"`/..
+HADOOP_PREFIX=${HADOOP_PREFIX:-$HADOOP_DEFAULT_PREFIX}
+export HADOOP_PREFIX
#check to see if the conf dir is given as an optional argument
if [ $# -gt 1 ]
Propchange: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/docs/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Nov 15 02:39:13 2011
@@ -1,2 +1,2 @@
-/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/docs:1152502-1196451
+/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/docs:1152502-1202009
/hadoop/core/branches/branch-0.19/src/docs:713112
Propchange: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Nov 15 02:39:13 2011
@@ -1,3 +1,3 @@
-/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java:1152502-1196451
+/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java:1152502-1202009
/hadoop/core/branches/branch-0.19/core/src/java:713112
/hadoop/core/trunk/src/core:776175-785643,785929-786278
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/http/HttpServer.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/http/HttpServer.java?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/http/HttpServer.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/http/HttpServer.java Tue Nov 15 02:39:13 2011
@@ -707,6 +707,14 @@ public class HttpServer implements Filte
listener.setPort((oriPort += 1));
}
}
+ // Make sure there is no handler failures.
+ Handler[] handlers = webServer.getHandlers();
+ for (int i = 0; i < handlers.length; i++) {
+ if (handlers[i].isFailed()) {
+ throw new IOException(
+ "Problem in starting http server. Server handlers failed");
+ }
+ }
} catch (IOException e) {
throw e;
} catch (InterruptedException e) {
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java Tue Nov 15 02:39:13 2011
@@ -48,6 +48,7 @@ import org.apache.commons.logging.*;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.RpcPayloadHeader.*;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
@@ -152,16 +153,20 @@ public class Client {
return refCount==0;
}
- /** A call waiting for a value. */
+ /**
+ * Class that represents an RPC call
+ */
private class Call {
- int id; // call id
- Writable param; // parameter
- Writable value; // value, null if error
- IOException error; // exception, null if value
- boolean done; // true when call is done
-
- protected Call(Writable param) {
- this.param = param;
+ final int id; // call id
+ final Writable rpcRequest; // the serialized rpc request - RpcPayload
+ Writable rpcResponse; // null if rpc has error
+ IOException error; // exception, null if success
+ final RpcKind rpcKind; // Rpc EngineKind
+ boolean done; // true when call is done
+
+ protected Call(RpcKind rpcKind, Writable param) {
+ this.rpcKind = rpcKind;
+ this.rpcRequest = param;
synchronized (Client.this) {
this.id = counter++;
}
@@ -187,15 +192,15 @@ public class Client {
/** Set the return value when there is no error.
* Notify the caller the call is done.
*
- * @param value return value of the call.
+ * @param rpcResponse return value of the rpc call.
*/
- public synchronized void setValue(Writable value) {
- this.value = value;
+ public synchronized void setRpcResponse(Writable rpcResponse) {
+ this.rpcResponse = rpcResponse;
callComplete();
}
- public synchronized Writable getValue() {
- return value;
+ public synchronized Writable getRpcResult() {
+ return rpcResponse;
}
}
@@ -727,6 +732,7 @@ public class Client {
}
}
+ @SuppressWarnings("unused")
public InetSocketAddress getRemoteAddress() {
return server;
}
@@ -787,8 +793,10 @@ public class Client {
//for serializing the
//data to be written
d = new DataOutputBuffer();
- d.writeInt(call.id);
- call.param.write(d);
+ RpcPayloadHeader header = new RpcPayloadHeader(
+ call.rpcKind, RpcPayloadOperation.RPC_FINAL_PAYLOAD, call.id);
+ header.write(d);
+ call.rpcRequest.write(d);
byte[] data = d.getData();
int dataLength = d.getLength();
out.writeInt(dataLength); //first put the data length
@@ -825,7 +833,7 @@ public class Client {
if (state == Status.SUCCESS.state) {
Writable value = ReflectionUtils.newInstance(valueClass, conf);
value.readFields(in); // read value
- call.setValue(value);
+ call.setRpcResponse(value);
calls.remove(id);
} else if (state == Status.ERROR.state) {
call.setException(new RemoteException(WritableUtils.readString(in),
@@ -909,7 +917,7 @@ public class Client {
private int index;
public ParallelCall(Writable param, ParallelResults results, int index) {
- super(param);
+ super(RpcKind.RPC_WRITABLE, param);
this.results = results;
this.index = index;
}
@@ -933,7 +941,7 @@ public class Client {
/** Collect a result. */
public synchronized void callComplete(ParallelCall call) {
- values[call.index] = call.getValue(); // store the value
+ values[call.index] = call.getRpcResult(); // store the value
count++; // count it
if (count == size) // if all values are in
notify(); // then notify waiting caller
@@ -993,15 +1001,23 @@ public class Client {
}
}
+ /**
+ * Same as {@link #call(RpcKind, Writable, ConnectionId)} for Writable
+ */
+ public Writable call(Writable param, InetSocketAddress address)
+ throws InterruptedException, IOException {
+ return call(RpcKind.RPC_WRITABLE, param, address);
+
+ }
/** Make a call, passing <code>param</code>, to the IPC server running at
* <code>address</code>, returning the value. Throws exceptions if there are
* network problems or if the remote code threw an exception.
- * @deprecated Use {@link #call(Writable, ConnectionId)} instead
+ * @deprecated Use {@link #call(RpcKind, Writable, ConnectionId)} instead
*/
@Deprecated
- public Writable call(Writable param, InetSocketAddress address)
+ public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress address)
throws InterruptedException, IOException {
- return call(param, address, null);
+ return call(rpcKind, param, address, null);
}
/** Make a call, passing <code>param</code>, to the IPC server running at
@@ -1009,15 +1025,15 @@ public class Client {
* the value.
* Throws exceptions if there are network problems or if the remote code
* threw an exception.
- * @deprecated Use {@link #call(Writable, ConnectionId)} instead
+ * @deprecated Use {@link #call(RpcKind, Writable, ConnectionId)} instead
*/
@Deprecated
- public Writable call(Writable param, InetSocketAddress addr,
+ public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress addr,
UserGroupInformation ticket)
throws InterruptedException, IOException {
ConnectionId remoteId = ConnectionId.getConnectionId(addr, null, ticket, 0,
conf);
- return call(param, remoteId);
+ return call(rpcKind, param, remoteId);
}
/** Make a call, passing <code>param</code>, to the IPC server running at
@@ -1026,18 +1042,33 @@ public class Client {
* timeout, returning the value.
* Throws exceptions if there are network problems or if the remote code
* threw an exception.
- * @deprecated Use {@link #call(Writable, ConnectionId)} instead
+ * @deprecated Use {@link #call(RpcKind, Writable, ConnectionId)} instead
*/
@Deprecated
- public Writable call(Writable param, InetSocketAddress addr,
+ public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress addr,
Class<?> protocol, UserGroupInformation ticket,
int rpcTimeout)
throws InterruptedException, IOException {
ConnectionId remoteId = ConnectionId.getConnectionId(addr, protocol,
ticket, rpcTimeout, conf);
- return call(param, remoteId);
+ return call(rpcKind, param, remoteId);
}
+
+ /**
+ * Same as {@link #call(RpcKind, Writable, InetSocketAddress,
+ * Class, UserGroupInformation, int, Configuration)}
+ * except that rpcKind is writable.
+ */
+ public Writable call(Writable param, InetSocketAddress addr,
+ Class<?> protocol, UserGroupInformation ticket,
+ int rpcTimeout, Configuration conf)
+ throws InterruptedException, IOException {
+ ConnectionId remoteId = ConnectionId.getConnectionId(addr, protocol,
+ ticket, rpcTimeout, conf);
+ return call(RpcKind.RPC_WRITABLE, param, remoteId);
+ }
+
/**
* Make a call, passing <code>param</code>, to the IPC server running at
* <code>address</code> which is servicing the <code>protocol</code> protocol,
@@ -1046,22 +1077,31 @@ public class Client {
* value. Throws exceptions if there are network problems or if the remote
* code threw an exception.
*/
- public Writable call(Writable param, InetSocketAddress addr,
+ public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress addr,
Class<?> protocol, UserGroupInformation ticket,
int rpcTimeout, Configuration conf)
throws InterruptedException, IOException {
ConnectionId remoteId = ConnectionId.getConnectionId(addr, protocol,
ticket, rpcTimeout, conf);
- return call(param, remoteId);
+ return call(rpcKind, param, remoteId);
+ }
+
+ /**
+ * Same as {link {@link #call(RpcKind, Writable, ConnectionId)}
+ * except the rpcKind is RPC_WRITABLE
+ */
+ public Writable call(Writable param, ConnectionId remoteId)
+ throws InterruptedException, IOException {
+ return call(RpcKind.RPC_WRITABLE, param, remoteId);
}
/** Make a call, passing <code>param</code>, to the IPC server defined by
* <code>remoteId</code>, returning the value.
* Throws exceptions if there are network problems or if the remote code
* threw an exception. */
- public Writable call(Writable param, ConnectionId remoteId)
+ public Writable call(RpcKind rpcKind, Writable param, ConnectionId remoteId)
throws InterruptedException, IOException {
- Call call = new Call(param);
+ Call call = new Call(rpcKind, param);
Connection connection = getConnection(remoteId, call);
connection.sendParam(call); // send the parameter
boolean interrupted = false;
@@ -1093,7 +1133,7 @@ public class Client {
call.error);
}
} else {
- return call.value;
+ return call.rpcResponse;
}
}
}
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java Tue Nov 15 02:39:13 2011
@@ -37,6 +37,7 @@ import org.apache.hadoop.classification.
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRpcExceptionProto;
import org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRpcRequestProto;
import org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRpcResponseProto;
@@ -139,7 +140,7 @@ public class ProtobufRpcEngine implement
HadoopRpcRequestProto rpcRequest = constructRpcRequest(method, args);
RpcResponseWritable val = null;
try {
- val = (RpcResponseWritable) client.call(
+ val = (RpcResponseWritable) client.call(RpcKind.RPC_PROTOCOL_BUFFER,
new RpcRequestWritable(rpcRequest), remoteId);
} catch (Exception e) {
RpcClientException ce = new RpcClientException("Client exception", e);
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java Tue Nov 15 02:39:13 2011
@@ -61,11 +61,13 @@ import org.apache.commons.logging.LogFac
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.hadoop.ipc.RpcPayloadHeader.*;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.ipc.RPC.VersionMismatch;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcPayloadOperation;
import org.apache.hadoop.ipc.metrics.RpcDetailedMetrics;
import org.apache.hadoop.ipc.metrics.RpcMetrics;
import org.apache.hadoop.net.NetUtils;
@@ -124,7 +126,8 @@ public abstract class Server {
// 4 : Introduced SASL security layer
// 5 : Introduced use of {@link ArrayPrimitiveWritable$Internal}
// in ObjectWritable to efficiently transmit arrays of primitives
- public static final byte CURRENT_VERSION = 5;
+ // 6 : Made RPC payload header explicit
+ public static final byte CURRENT_VERSION = 6;
/**
* Initial and max size of response buffer
@@ -277,28 +280,33 @@ public abstract class Server {
/** A call queued for handling. */
private static class Call {
- private int id; // the client's call id
- private Writable param; // the parameter passed
- private Connection connection; // connection to client
- private long timestamp; // the time received when response is null
- // the time served when response is not null
- private ByteBuffer response; // the response for this call
-
- public Call(int id, Writable param, Connection connection) {
- this.id = id;
- this.param = param;
+ private final int callId; // the client's call id
+ private final Writable rpcRequest; // Serialized Rpc request from client
+ private final Connection connection; // connection to client
+ private long timestamp; // time received when response is null
+ // time served when response is not null
+ private ByteBuffer rpcResponse; // the response for this call
+ private final RpcKind rpcKind;
+
+ public Call(int id, Writable param, Connection connection) {
+ this( id, param, connection, RpcKind.RPC_BUILTIN );
+ }
+ public Call(int id, Writable param, Connection connection, RpcKind kind) {
+ this.callId = id;
+ this.rpcRequest = param;
this.connection = connection;
this.timestamp = System.currentTimeMillis();
- this.response = null;
+ this.rpcResponse = null;
+ this.rpcKind = kind;
}
@Override
public String toString() {
- return param.toString() + " from " + connection.toString();
+ return rpcRequest.toString() + " from " + connection.toString();
}
public void setResponse(ByteBuffer response) {
- this.response = response;
+ this.rpcResponse = response;
}
}
@@ -795,17 +803,17 @@ public abstract class Server {
call = responseQueue.removeFirst();
SocketChannel channel = call.connection.channel;
if (LOG.isDebugEnabled()) {
- LOG.debug(getName() + ": responding to #" + call.id + " from " +
+ LOG.debug(getName() + ": responding to #" + call.callId + " from " +
call.connection);
}
//
// Send as much data as we can in the non-blocking fashion
//
- int numBytes = channelWrite(channel, call.response);
+ int numBytes = channelWrite(channel, call.rpcResponse);
if (numBytes < 0) {
return true;
}
- if (!call.response.hasRemaining()) {
+ if (!call.rpcResponse.hasRemaining()) {
call.connection.decRpcCount();
if (numElements == 1) { // last call fully processes.
done = true; // no more data for this channel.
@@ -813,7 +821,7 @@ public abstract class Server {
done = false; // more calls pending to be sent.
}
if (LOG.isDebugEnabled()) {
- LOG.debug(getName() + ": responding to #" + call.id + " from " +
+ LOG.debug(getName() + ": responding to #" + call.callId + " from " +
call.connection + " Wrote " + numBytes + " bytes.");
}
} else {
@@ -841,7 +849,7 @@ public abstract class Server {
}
}
if (LOG.isDebugEnabled()) {
- LOG.debug(getName() + ": responding to #" + call.id + " from " +
+ LOG.debug(getName() + ": responding to #" + call.callId + " from " +
call.connection + " Wrote partial " + numBytes +
" bytes.");
}
@@ -1408,18 +1416,24 @@ public abstract class Server {
private void processData(byte[] buf) throws IOException, InterruptedException {
DataInputStream dis =
new DataInputStream(new ByteArrayInputStream(buf));
- int id = dis.readInt(); // try to read an id
+ RpcPayloadHeader header = new RpcPayloadHeader();
+ header.readFields(dis); // Read the RpcPayload header
if (LOG.isDebugEnabled())
- LOG.debug(" got #" + id);
- Writable param;
- try {
- param = ReflectionUtils.newInstance(paramClass, conf);//read param
- param.readFields(dis);
+ LOG.debug(" got #" + header.getCallId());
+ if (header.getOperation() != RpcPayloadOperation.RPC_FINAL_PAYLOAD) {
+ throw new IOException("IPC Server does not implement operation" +
+ header.getOperation());
+ }
+ Writable rpcRequest;
+ try { //Read the rpc request
+ rpcRequest = ReflectionUtils.newInstance(paramClass, conf);
+ rpcRequest.readFields(dis);
} catch (Throwable t) {
LOG.warn("Unable to read call parameters for client " +
getHostAddress(), t);
- final Call readParamsFailedCall = new Call(id, null, this);
+ final Call readParamsFailedCall =
+ new Call(header.getCallId(), null, this);
ByteArrayOutputStream responseBuffer = new ByteArrayOutputStream();
setupResponse(responseBuffer, readParamsFailedCall, Status.FATAL, null,
@@ -1429,7 +1443,7 @@ public abstract class Server {
return;
}
- Call call = new Call(id, param, this);
+ Call call = new Call(header.getCallId(), rpcRequest, this, header.getkind());
callQueue.put(call); // queue the call; maybe blocked here
incRpcCount(); // Increment the rpc count
}
@@ -1493,8 +1507,8 @@ public abstract class Server {
final Call call = callQueue.take(); // pop the queue; maybe blocked here
if (LOG.isDebugEnabled())
- LOG.debug(getName() + ": has #" + call.id + " from " +
- call.connection);
+ LOG.debug(getName() + ": has Call#" + call.callId +
+ "for RpcKind " + call.rpcKind + " from " + call.connection);
String errorClass = null;
String error = null;
@@ -1505,7 +1519,7 @@ public abstract class Server {
// Make the call as the user via Subject.doAs, thus associating
// the call with the Subject
if (call.connection.user == null) {
- value = call(call.connection.protocolName, call.param,
+ value = call(call.connection.protocolName, call.rpcRequest,
call.timestamp);
} else {
value =
@@ -1515,7 +1529,7 @@ public abstract class Server {
public Writable run() throws Exception {
// make the call
return call(call.connection.protocolName,
- call.param, call.timestamp);
+ call.rpcRequest, call.timestamp);
}
}
@@ -1657,7 +1671,7 @@ public abstract class Server {
throws IOException {
response.reset();
DataOutputStream out = new DataOutputStream(response);
- out.writeInt(call.id); // write call id
+ out.writeInt(call.callId); // write call id
out.writeInt(status.state); // write status
if (status == Status.SUCCESS) {
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java Tue Nov 15 02:39:13 2011
@@ -39,6 +39,7 @@ import javax.net.SocketFactory;
import org.apache.commons.logging.*;
import org.apache.hadoop.io.*;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.ipc.VersionedProtocol;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.SecretManager;
@@ -242,7 +243,7 @@ public class WritableRpcEngine implement
}
ObjectWritable value = (ObjectWritable)
- client.call(new Invocation(method, args), remoteId);
+ client.call(RpcKind.RPC_WRITABLE, new Invocation(method, args), remoteId);
if (LOG.isDebugEnabled()) {
long callTime = System.currentTimeMillis() - startTime;
LOG.debug("Call: " + method.getName() + " " + callTime);
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-datanode
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-datanode?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-datanode (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-datanode Tue Nov 15 02:39:13 2011
@@ -75,7 +75,7 @@ check_privsep_dir() {
}
export PATH="${PATH:+$PATH:}/usr/sbin:/usr/bin"
-export HADOOP_PREFIX="/usr"
+export HADOOP_PREFIX=${HADOOP_PREFIX:-/usr}
case "$1" in
start)
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-jobtracker
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-jobtracker?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-jobtracker (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-jobtracker Tue Nov 15 02:39:13 2011
@@ -67,7 +67,7 @@ check_privsep_dir() {
}
export PATH="${PATH:+$PATH:}/usr/sbin:/usr/bin"
-export HADOOP_PREFIX="/usr"
+export HADOOP_PREFIX=${HADOOP_PREFIX:-/usr}
case "$1" in
start)
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-namenode
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-namenode?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-namenode (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-namenode Tue Nov 15 02:39:13 2011
@@ -71,7 +71,7 @@ format() {
}
export PATH="${PATH:+$PATH:}/usr/sbin:/usr/bin"
-export HADOOP_PREFIX="/usr"
+export HADOOP_PREFIX=${HADOOP_PREFIX:-/usr}
case "$1" in
start)
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-tasktracker
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-tasktracker?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-tasktracker (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/deb/init.d/hadoop-tasktracker Tue Nov 15 02:39:13 2011
@@ -67,7 +67,7 @@ check_privsep_dir() {
}
export PATH="${PATH:+$PATH:}/usr/sbin:/usr/bin"
-export HADOOP_PREFIX="/usr"
+export HADOOP_PREFIX=${HADOOP_PREFIX:-/usr}
case "$1" in
start)
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/hadoop-setup-conf.sh
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/hadoop-setup-conf.sh?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/hadoop-setup-conf.sh (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/hadoop-setup-conf.sh Tue Nov 15 02:39:13 2011
@@ -19,7 +19,9 @@ bin=`dirname "$0"`
bin=`cd "$bin"; pwd`
this="${BASH_SOURCE-$0}"
-export HADOOP_PREFIX=`dirname "$this"`/..
+HADOOP_DEFAULT_PREFIX=`dirname "$this"`/..
+HADOOP_PREFIX=${HADOOP_PREFIX:-$HADOOP_DEFAULT_PREFIX}
+export HADOOP_PREFIX
usage() {
echo "
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/rpm/init.d/hadoop-jobtracker
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/rpm/init.d/hadoop-jobtracker?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/rpm/init.d/hadoop-jobtracker (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/rpm/init.d/hadoop-jobtracker Tue Nov 15 02:39:13 2011
@@ -27,7 +27,7 @@ source /etc/default/hadoop-env.sh
RETVAL=0
PIDFILE="${HADOOP_PID_DIR}/hadoop-mapred-jobtracker.pid"
desc="Hadoop jobtracker daemon"
-export HADOOP_PREFIX="/usr"
+export HADOOP_PREFIX=${HADOOP_PREFIX:-/usr}
start() {
echo -n $"Starting $desc (hadoop-jobtracker): "
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/rpm/init.d/hadoop-namenode
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/rpm/init.d/hadoop-namenode?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/rpm/init.d/hadoop-namenode (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/rpm/init.d/hadoop-namenode Tue Nov 15 02:39:13 2011
@@ -27,7 +27,7 @@ source /etc/default/hadoop-env.sh
RETVAL=0
PIDFILE="${HADOOP_PID_DIR}/hadoop-hdfs-namenode.pid"
desc="Hadoop namenode daemon"
-export HADOOP_PREFIX="/usr"
+export HADOOP_PREFIX=${HADOOP_PREFIX:-/usr}
start() {
echo -n $"Starting $desc (hadoop-namenode): "
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/rpm/init.d/hadoop-tasktracker
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/rpm/init.d/hadoop-tasktracker?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/rpm/init.d/hadoop-tasktracker (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/main/packages/rpm/init.d/hadoop-tasktracker Tue Nov 15 02:39:13 2011
@@ -27,7 +27,7 @@ source /etc/default/hadoop-env.sh
RETVAL=0
PIDFILE="${HADOOP_PID_DIR}/hadoop-mapred-tasktracker.pid"
desc="Hadoop tasktracker daemon"
-export HADOOP_PREFIX="/usr"
+export HADOOP_PREFIX=${HADOOP_PREFIX:-/usr}
start() {
echo -n $"Starting $desc (hadoop-tasktracker): "
Propchange: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/test/core/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Nov 15 02:39:13 2011
@@ -1,3 +1,3 @@
-/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/core:1152502-1196451
+/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/core:1152502-1202009
/hadoop/core/branches/branch-0.19/core/src/test/core:713112
/hadoop/core/trunk/src/test/core:776175-785643,785929-786278
Modified: hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/http/TestServletFilter.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/http/TestServletFilter.java?rev=1202013&r1=1202012&r2=1202013&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/http/TestServletFilter.java (original)
+++ hadoop/common/branches/HDFS-1623/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/http/TestServletFilter.java Tue Nov 15 02:39:13 2011
@@ -45,7 +45,7 @@ public class TestServletFilter extends H
static public class SimpleFilter implements Filter {
private FilterConfig filterConfig = null;
- public void init(FilterConfig filterConfig) {
+ public void init(FilterConfig filterConfig) throws ServletException {
this.filterConfig = filterConfig;
}
@@ -137,4 +137,36 @@ public class TestServletFilter extends H
http.stop();
}
}
+
+ static public class ErrorFilter extends SimpleFilter {
+ @Override
+ public void init(FilterConfig arg0) throws ServletException {
+ throw new ServletException("Throwing the exception from Filter init");
+ }
+
+ /** Configuration for the filter */
+ static public class Initializer extends FilterInitializer {
+ public Initializer() {
+ }
+
+ public void initFilter(FilterContainer container, Configuration conf) {
+ container.addFilter("simple", ErrorFilter.class.getName(), null);
+ }
+ }
+ }
+
+ @Test
+ public void testServletFilterWhenInitThrowsException() throws Exception {
+ Configuration conf = new Configuration();
+ // start a http server with CountingFilter
+ conf.set(HttpServer.FILTER_INITIALIZER_PROPERTY,
+ ErrorFilter.Initializer.class.getName());
+ HttpServer http = createTestServer(conf);
+ try {
+ http.start();
+ fail("expecting exception");
+ } catch (IOException e) {
+ assertTrue( e.getMessage().contains("Problem in starting http server. Server handlers failed"));
+ }
+ }
}