|
@@ -834,15 +834,15 @@ public abstract class Server {
|
|
|
final Writable rpcRequest; // Serialized Rpc request from client
|
|
|
ByteBuffer rpcResponse; // the response for this call
|
|
|
|
|
|
- private RpcResponseHeaderProto bufferedHeader; // the response header
|
|
|
- private Writable bufferedRv; // the byte response
|
|
|
+ private ResponseParams responseParams; // the response params
|
|
|
+ private Writable rv; // the byte response
|
|
|
|
|
|
RpcCall(RpcCall call) {
|
|
|
super(call);
|
|
|
this.connection = call.connection;
|
|
|
this.rpcRequest = call.rpcRequest;
|
|
|
- this.bufferedRv = call.bufferedRv;
|
|
|
- this.bufferedHeader = call.bufferedHeader;
|
|
|
+ this.rv = call.rv;
|
|
|
+ this.responseParams = call.responseParams;
|
|
|
}
|
|
|
|
|
|
RpcCall(Connection connection, int id) {
|
|
@@ -863,12 +863,10 @@ public abstract class Server {
|
|
|
this.rpcRequest = param;
|
|
|
}
|
|
|
|
|
|
- public void setBufferedHeader(RpcResponseHeaderProto header) {
|
|
|
- this.bufferedHeader = header;
|
|
|
- }
|
|
|
-
|
|
|
- public void setBufferedRv(Writable rv) {
|
|
|
- this.bufferedRv = rv;
|
|
|
+ void setResponseFields(Writable returnValue,
|
|
|
+ ResponseParams responseParams) {
|
|
|
+ this.rv = returnValue;
|
|
|
+ this.responseParams = responseParams;
|
|
|
}
|
|
|
|
|
|
@Override
|
|
@@ -902,9 +900,7 @@ public abstract class Server {
|
|
|
populateResponseParamsOnError(e, responseParams);
|
|
|
}
|
|
|
if (!isResponseDeferred()) {
|
|
|
- setupResponse(this, responseParams.returnStatus,
|
|
|
- responseParams.detailedErr,
|
|
|
- value, responseParams.errorClass, responseParams.error);
|
|
|
+ setResponseFields(value, responseParams);
|
|
|
sendResponse();
|
|
|
} else {
|
|
|
if (LOG.isDebugEnabled()) {
|
|
@@ -959,13 +955,11 @@ public abstract class Server {
|
|
|
setupResponse(call,
|
|
|
RpcStatusProto.FATAL, RpcErrorCodeProto.ERROR_RPC_SERVER,
|
|
|
null, t.getClass().getName(), StringUtils.stringifyException(t));
|
|
|
- } else if (alignmentContext != null) {
|
|
|
- // rebuild response with state context in header
|
|
|
- RpcResponseHeaderProto.Builder responseHeader =
|
|
|
- call.bufferedHeader.toBuilder();
|
|
|
- alignmentContext.updateResponseState(responseHeader);
|
|
|
- RpcResponseHeaderProto builtHeader = responseHeader.build();
|
|
|
- setupResponse(call, builtHeader, call.bufferedRv);
|
|
|
+ } else {
|
|
|
+ setupResponse(call, call.responseParams.returnStatus,
|
|
|
+ call.responseParams.detailedErr, call.rv,
|
|
|
+ call.responseParams.errorClass,
|
|
|
+ call.responseParams.error);
|
|
|
}
|
|
|
connection.sendResponse(call);
|
|
|
}
|
|
@@ -2879,6 +2873,9 @@ public abstract class Server {
|
|
|
headerBuilder.setRetryCount(call.retryCount);
|
|
|
headerBuilder.setStatus(status);
|
|
|
headerBuilder.setServerIpcVersionNum(CURRENT_VERSION);
|
|
|
+ if (alignmentContext != null) {
|
|
|
+ alignmentContext.updateResponseState(headerBuilder);
|
|
|
+ }
|
|
|
|
|
|
if (status == RpcStatusProto.SUCCESS) {
|
|
|
RpcResponseHeaderProto header = headerBuilder.build();
|
|
@@ -2905,12 +2902,6 @@ public abstract class Server {
|
|
|
|
|
|
private void setupResponse(RpcCall call,
|
|
|
RpcResponseHeaderProto header, Writable rv) throws IOException {
|
|
|
- if (alignmentContext != null && call.bufferedHeader == null
|
|
|
- && call.bufferedRv == null) {
|
|
|
- call.setBufferedHeader(header);
|
|
|
- call.setBufferedRv(rv);
|
|
|
- }
|
|
|
-
|
|
|
final byte[] response;
|
|
|
if (rv == null || (rv instanceof RpcWritable.ProtobufWrapper)) {
|
|
|
response = setupResponseForProtobuf(header, rv);
|