Author: todd
Date: Mon Oct 31 21:52:52 2011
New Revision: 1195692
URL: http://svn.apache.org/viewvc?rev=1195692&view=rev
Log:
HDFS-2512. Add textual error message to data transfer protocol responses.
Contributed by Todd Lipcon.
Modified:
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/proto/DataTransferProtos.java
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/proto/datatransfer.proto
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDataTransferProtocol.java
Modified:
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt?rev=1195692&r1=1195691&r2=1195692&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
(original)
+++
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
Mon Oct 31 21:52:52 2011
@@ -758,6 +758,9 @@ Release 0.23.0 - Unreleased
HDFS-2436. Change FSNamesystem.setTimes(..) for allowing setting times on
directories. (Uma Maheswara Rao G via szetszwo)
+ HDFS-2512. Add textual error message to data transfer protocol responses
+ (todd)
+
OPTIMIZATIONS
HDFS-1458. Improve checkpoint performance by avoiding unnecessary image
Modified:
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/proto/DataTransferProtos.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/proto/DataTransferProtos.java?rev=1195692&r1=1195691&r2=1195692&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/proto/DataTransferProtos.java
(original)
+++
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/proto/DataTransferProtos.java
Mon Oct 31 21:52:52 2011
@@ -15,7 +15,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
// Generated by the protocol buffer compiler. DO NOT EDIT!
// source: datatransfer.proto
@@ -6936,6 +6935,10 @@ public final class DataTransferProtos {
boolean hasChecksumResponse();
org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpBlockChecksumResponseProto
getChecksumResponse();
org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpBlockChecksumResponseProtoOrBuilder
getChecksumResponseOrBuilder();
+
+ // optional string message = 4;
+ boolean hasMessage();
+ String getMessage();
}
public static final class BlockOpResponseProto extends
com.google.protobuf.GeneratedMessage
@@ -7021,10 +7024,43 @@ public final class DataTransferProtos {
return checksumResponse_;
}
+ // optional string message = 4;
+ public static final int MESSAGE_FIELD_NUMBER = 4;
+ private java.lang.Object message_;
+ public boolean hasMessage() {
+ return ((bitField0_ & 0x00000008) == 0x00000008);
+ }
+ public String getMessage() {
+ java.lang.Object ref = message_;
+ 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)) {
+ message_ = s;
+ }
+ return s;
+ }
+ }
+ private com.google.protobuf.ByteString getMessageBytes() {
+ java.lang.Object ref = message_;
+ if (ref instanceof String) {
+ com.google.protobuf.ByteString b =
+ com.google.protobuf.ByteString.copyFromUtf8((String) ref);
+ message_ = b;
+ return b;
+ } else {
+ return (com.google.protobuf.ByteString) ref;
+ }
+ }
+
private void initFields() {
status_ =
org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status.SUCCESS;
firstBadLink_ = "";
checksumResponse_ =
org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpBlockChecksumResponseProto.getDefaultInstance();
+ message_ = "";
}
private byte memoizedIsInitialized = -1;
public final boolean isInitialized() {
@@ -7057,6 +7093,9 @@ public final class DataTransferProtos {
if (((bitField0_ & 0x00000004) == 0x00000004)) {
output.writeMessage(3, checksumResponse_);
}
+ if (((bitField0_ & 0x00000008) == 0x00000008)) {
+ output.writeBytes(4, getMessageBytes());
+ }
getUnknownFields().writeTo(output);
}
@@ -7078,6 +7117,10 @@ public final class DataTransferProtos {
size += com.google.protobuf.CodedOutputStream
.computeMessageSize(3, checksumResponse_);
}
+ if (((bitField0_ & 0x00000008) == 0x00000008)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeBytesSize(4, getMessageBytes());
+ }
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
return size;
@@ -7116,6 +7159,11 @@ public final class DataTransferProtos {
result = result && getChecksumResponse()
.equals(other.getChecksumResponse());
}
+ result = result && (hasMessage() == other.hasMessage());
+ if (hasMessage()) {
+ result = result && getMessage()
+ .equals(other.getMessage());
+ }
result = result &&
getUnknownFields().equals(other.getUnknownFields());
return result;
@@ -7137,6 +7185,10 @@ public final class DataTransferProtos {
hash = (37 * hash) + CHECKSUMRESPONSE_FIELD_NUMBER;
hash = (53 * hash) + getChecksumResponse().hashCode();
}
+ if (hasMessage()) {
+ hash = (37 * hash) + MESSAGE_FIELD_NUMBER;
+ hash = (53 * hash) + getMessage().hashCode();
+ }
hash = (29 * hash) + getUnknownFields().hashCode();
return hash;
}
@@ -7264,6 +7316,8 @@ public final class DataTransferProtos {
checksumResponseBuilder_.clear();
}
bitField0_ = (bitField0_ & ~0x00000004);
+ message_ = "";
+ bitField0_ = (bitField0_ & ~0x00000008);
return this;
}
@@ -7318,6 +7372,10 @@ public final class DataTransferProtos {
} else {
result.checksumResponse_ = checksumResponseBuilder_.build();
}
+ if (((from_bitField0_ & 0x00000008) == 0x00000008)) {
+ to_bitField0_ |= 0x00000008;
+ }
+ result.message_ = message_;
result.bitField0_ = to_bitField0_;
onBuilt();
return result;
@@ -7343,6 +7401,9 @@ public final class DataTransferProtos {
if (other.hasChecksumResponse()) {
mergeChecksumResponse(other.getChecksumResponse());
}
+ if (other.hasMessage()) {
+ setMessage(other.getMessage());
+ }
this.mergeUnknownFields(other.getUnknownFields());
return this;
}
@@ -7409,6 +7470,11 @@ public final class DataTransferProtos {
setChecksumResponse(subBuilder.buildPartial());
break;
}
+ case 34: {
+ bitField0_ |= 0x00000008;
+ message_ = input.readBytes();
+ break;
+ }
}
}
}
@@ -7565,6 +7631,42 @@ public final class DataTransferProtos {
return checksumResponseBuilder_;
}
+ // optional string message = 4;
+ private java.lang.Object message_ = "";
+ public boolean hasMessage() {
+ return ((bitField0_ & 0x00000008) == 0x00000008);
+ }
+ public String getMessage() {
+ java.lang.Object ref = message_;
+ if (!(ref instanceof String)) {
+ String s = ((com.google.protobuf.ByteString) ref).toStringUtf8();
+ message_ = s;
+ return s;
+ } else {
+ return (String) ref;
+ }
+ }
+ public Builder setMessage(String value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ bitField0_ |= 0x00000008;
+ message_ = value;
+ onChanged();
+ return this;
+ }
+ public Builder clearMessage() {
+ bitField0_ = (bitField0_ & ~0x00000008);
+ message_ = getDefaultInstance().getMessage();
+ onChanged();
+ return this;
+ }
+ void setMessage(com.google.protobuf.ByteString value) {
+ bitField0_ |= 0x00000008;
+ message_ = value;
+ onChanged();
+ }
+
// @@protoc_insertion_point(builder_scope:BlockOpResponseProto)
}
@@ -8995,19 +9097,20 @@ public final class DataTransferProtos {
"\030\001 \002(\020\022\r\n\005seqno\030\002
\002(\020\022\031\n\021lastPacketInBlo" +
"ck\030\003 \002(\010\022\017\n\007dataLen\030\004
\002(\017\":\n\020PipelineAck" +
"Proto\022\r\n\005seqno\030\001 \002(\022\022\027\n\006status\030\002
\003(\0162\007.S" +
- "tatus\"~\n\024BlockOpResponseProto\022\027\n\006status\030" +
- "\001 \002(\0162\007.Status\022\024\n\014firstBadLink\030\002
\001(\t\0227\n\020" +
- "checksumResponse\030\003 \001(\0132\035.OpBlockChecksum" +
- "ResponseProto\"0\n\025ClientReadStatusProto\022\027" +
- "\n\006status\030\001 \002(\0162\007.Status\"-\n\022DNTransferAck" +
- "Proto\022\027\n\006status\030\001 \002(\0162\007.Status\"U\n\034OpBloc",
- "kChecksumResponseProto\022\023\n\013bytesPerCrc\030\001 " +
- "\002(\r\022\023\n\013crcPerBlock\030\002
\002(\004\022\013\n\003md5\030\003 \002(\014*\202\001" +
-
"\n\006Status\022\013\n\007SUCCESS\020\000\022\t\n\005ERROR\020\001\022\022\n\016ERRO"
+
- "R_CHECKSUM\020\002\022\021\n\rERROR_INVALID\020\003\022\020\n\014ERROR"
+
-
"_EXISTS\020\004\022\026\n\022ERROR_ACCESS_TOKEN\020\005\022\017\n\013CHE" +
- "CKSUM_OK\020\006B>\n%org.apache.hadoop.hdfs.pro" +
- "tocol.protoB\022DataTransferProtos\240\001\001"
+ "tatus\"\217\001\n\024BlockOpResponseProto\022\027\n\006status" +
+ "\030\001 \002(\0162\007.Status\022\024\n\014firstBadLink\030\002
\001(\t\0227\n" +
+ "\020checksumResponse\030\003 \001(\0132\035.OpBlockChecksu" +
+ "mResponseProto\022\017\n\007message\030\004 \001(\t\"0\n\025Clien" +
+ "tReadStatusProto\022\027\n\006status\030\001 \002(\0162\007.Statu" +
+ "s\"-\n\022DNTransferAckProto\022\027\n\006status\030\001 \002(\0162",
+ "\007.Status\"U\n\034OpBlockChecksumResponseProto" +
+ "\022\023\n\013bytesPerCrc\030\001
\002(\r\022\023\n\013crcPerBlock\030\002 \002" +
+ "(\004\022\013\n\003md5\030\003
\002(\014*\202\001\n\006Status\022\013\n\007SUCCESS\020\000\022" +
+
"\t\n\005ERROR\020\001\022\022\n\016ERROR_CHECKSUM\020\002\022\021\n\rERROR_" +
+
"INVALID\020\003\022\020\n\014ERROR_EXISTS\020\004\022\026\n\022ERROR_ACC" +
+ "ESS_TOKEN\020\005\022\017\n\013CHECKSUM_OK\020\006B>\n%org.apac" +
+ "he.hadoop.hdfs.protocol.protoB\022DataTrans" +
+ "ferProtos\240\001\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner
assigner =
new
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -9099,7 +9202,7 @@ public final class DataTransferProtos {
internal_static_BlockOpResponseProto_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_BlockOpResponseProto_descriptor,
- new java.lang.String[] { "Status", "FirstBadLink",
"ChecksumResponse", },
+ new java.lang.String[] { "Status", "FirstBadLink",
"ChecksumResponse", "Message", },
org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProto.class,
org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProto.Builder.class);
internal_static_ClientReadStatusProto_descriptor =
Modified:
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java?rev=1195692&r1=1195691&r2=1195692&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java
(original)
+++
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java
Mon Oct 31 21:52:52 2011
@@ -358,7 +358,8 @@ public class Balancer {
if (response.getStatus() != Status.SUCCESS) {
if (response.getStatus() == Status.ERROR_ACCESS_TOKEN)
throw new IOException("block move failed due to access token error");
- throw new IOException("block move is failed");
+ throw new IOException("block move is failed: " +
+ response.getMessage());
}
}
Modified:
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java?rev=1195692&r1=1195691&r2=1195692&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
(original)
+++
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
Mon Oct 31 21:52:52 2011
@@ -48,6 +48,8 @@ import org.apache.hadoop.hdfs.protocol.d
import org.apache.hadoop.hdfs.protocol.datatransfer.Receiver;
import org.apache.hadoop.hdfs.protocol.datatransfer.Sender;
import
org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProto;
+import
org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProto.Builder;
+import
org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProtoOrBuilder;
import
org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ClientReadStatusProto;
import
org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpBlockChecksumResponseProto;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status;
@@ -225,13 +227,14 @@ class DataXceiver extends Receiver imple
blockSender = new BlockSender(block, blockOffset, length,
true, true, false, datanode, clientTraceFmt);
} catch(IOException e) {
- LOG.info("opReadBlock " + block + " received exception " + e);
- sendResponse(s, ERROR, datanode.socketWriteTimeout);
+ String msg = "opReadBlock " + block + " received exception " + e;
+ LOG.info(msg);
+ sendResponse(s, ERROR, msg, datanode.socketWriteTimeout);
throw e;
}
// send op status
- sendResponse(s, SUCCESS, datanode.socketWriteTimeout);
+ sendResponse(s, SUCCESS, null, datanode.socketWriteTimeout);
long read = blockSender.sendBlock(out, baseStream, null); // send data
@@ -452,7 +455,7 @@ class DataXceiver extends Receiver imple
if (LOG.isTraceEnabled()) {
LOG.trace("TRANSFER: send close-ack");
}
- writeResponse(SUCCESS, replyOut);
+ writeResponse(SUCCESS, null, replyOut);
}
}
@@ -507,7 +510,7 @@ class DataXceiver extends Receiver imple
NetUtils.getOutputStream(s, datanode.socketWriteTimeout));
try {
datanode.transferReplicaForPipelineRecovery(blk, targets, clientName);
- writeResponse(Status.SUCCESS, out);
+ writeResponse(Status.SUCCESS, null, out);
} finally {
IOUtils.closeStream(out);
}
@@ -577,16 +580,17 @@ class DataXceiver extends Receiver imple
LOG.warn("Invalid access token in request from " + remoteAddress
+ " for OP_COPY_BLOCK for block " + block + " : "
+ e.getLocalizedMessage());
- sendResponse(s, ERROR_ACCESS_TOKEN, datanode.socketWriteTimeout);
+ sendResponse(s, ERROR_ACCESS_TOKEN, "Invalid access token",
datanode.socketWriteTimeout);
return;
}
}
if (!dataXceiverServer.balanceThrottler.acquire()) { // not able to start
- LOG.info("Not able to copy block " + block.getBlockId() + " to "
- + s.getRemoteSocketAddress() + " because threads quota is
exceeded.");
- sendResponse(s, ERROR, datanode.socketWriteTimeout);
+ String msg = "Not able to copy block " + block.getBlockId() + " to "
+ + s.getRemoteSocketAddress() + " because threads quota is exceeded.";
+ LOG.info(msg);
+ sendResponse(s, ERROR, msg, datanode.socketWriteTimeout);
return;
}
@@ -606,7 +610,7 @@ class DataXceiver extends Receiver imple
baseStream, HdfsConstants.SMALL_BUFFER_SIZE));
// send status first
- writeResponse(SUCCESS, reply);
+ writeResponse(SUCCESS, null, reply);
// send block content to the target
long read = blockSender.sendBlock(reply, baseStream,
dataXceiverServer.balanceThrottler);
@@ -653,21 +657,24 @@ class DataXceiver extends Receiver imple
LOG.warn("Invalid access token in request from " + remoteAddress
+ " for OP_REPLACE_BLOCK for block " + block + " : "
+ e.getLocalizedMessage());
- sendResponse(s, ERROR_ACCESS_TOKEN, datanode.socketWriteTimeout);
+ sendResponse(s, ERROR_ACCESS_TOKEN, "Invalid access token",
+ datanode.socketWriteTimeout);
return;
}
}
if (!dataXceiverServer.balanceThrottler.acquire()) { // not able to start
- LOG.warn("Not able to receive block " + block.getBlockId() + " from "
- + s.getRemoteSocketAddress() + " because threads quota is
exceeded.");
- sendResponse(s, ERROR, datanode.socketWriteTimeout);
+ String msg = "Not able to receive block " + block.getBlockId() + " from
"
+ + s.getRemoteSocketAddress() + " because threads quota is
exceeded.";
+ LOG.warn(msg);
+ sendResponse(s, ERROR, msg, datanode.socketWriteTimeout);
return;
}
Socket proxySock = null;
DataOutputStream proxyOut = null;
Status opStatus = SUCCESS;
+ String errMsg = null;
BlockReceiver blockReceiver = null;
DataInputStream proxyReply = null;
@@ -720,7 +727,8 @@ class DataXceiver extends Receiver imple
} catch (IOException ioe) {
opStatus = ERROR;
- LOG.info("opReplaceBlock " + block + " received exception " + ioe);
+ errMsg = "opReplaceBlock " + block + " received exception " + ioe;
+ LOG.info(errMsg);
throw ioe;
} finally {
// receive the last byte that indicates the proxy released its thread
resource
@@ -736,7 +744,7 @@ class DataXceiver extends Receiver imple
// send response back
try {
- sendResponse(s, opStatus, datanode.socketWriteTimeout);
+ sendResponse(s, opStatus, errMsg, datanode.socketWriteTimeout);
} catch (IOException ioe) {
LOG.warn("Error writing reply back to " + s.getRemoteSocketAddress());
}
@@ -759,21 +767,22 @@ class DataXceiver extends Receiver imple
* @param opStatus status message to write
* @param timeout send timeout
**/
- private void sendResponse(Socket s, Status status,
+ private void sendResponse(Socket s, Status status, String message,
long timeout) throws IOException {
DataOutputStream reply =
new DataOutputStream(NetUtils.getOutputStream(s, timeout));
- writeResponse(status, reply);
+ writeResponse(status, message, reply);
}
- private void writeResponse(Status status, OutputStream out)
+ private void writeResponse(Status status, String message, OutputStream out)
throws IOException {
- BlockOpResponseProto response = BlockOpResponseProto.newBuilder()
- .setStatus(status)
- .build();
-
- response.writeDelimitedTo(out);
+ BlockOpResponseProto.Builder response = BlockOpResponseProto.newBuilder()
+ .setStatus(status);
+ if (message != null) {
+ response.setMessage(message);
+ }
+ response.build().writeDelimitedTo(out);
out.flush();
}
Modified:
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/proto/datatransfer.proto
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/proto/datatransfer.proto?rev=1195692&r1=1195691&r2=1195692&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/proto/datatransfer.proto
(original)
+++
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/proto/datatransfer.proto
Mon Oct 31 21:52:52 2011
@@ -119,6 +119,9 @@ message BlockOpResponseProto {
optional string firstBadLink = 2;
optional OpBlockChecksumResponseProto checksumResponse = 3;
+
+ /** explanatory text which may be useful to log on the client side */
+ optional string message = 4;
}
/**
Modified:
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDataTransferProtocol.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDataTransferProtocol.java?rev=1195692&r1=1195691&r2=1195692&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDataTransferProtocol.java
(original)
+++
hadoop/common/branches/branch-0.23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDataTransferProtocol.java
Mon Oct 31 21:52:52 2011
@@ -117,10 +117,8 @@ public class TestDataTransferProtocol ex
throw eof;
}
- LOG.info("Received: " +
- StringUtils.byteToHexString(retBuf));
- LOG.info("Expected: " +
- StringUtils.byteToHexString(recvBuf.toByteArray()));
+ LOG.info("Received: " +new String(retBuf));
+ LOG.info("Expected: " +
StringUtils.byteToHexString(recvBuf.toByteArray()));
if (eofExpected) {
throw new IOException("Did not recieve IOException when an exception "
+
@@ -129,10 +127,8 @@ public class TestDataTransferProtocol ex
}
byte[] needed = recvBuf.toByteArray();
- for (int i=0; i<retBuf.length; i++) {
- System.out.print(retBuf[i]);
- assertEquals("checking byte[" + i + "]", needed[i], retBuf[i]);
- }
+ assertEquals(StringUtils.byteToHexString(needed),
+ StringUtils.byteToHexString(retBuf));
} finally {
IOUtils.closeSocket(sock);
}
@@ -166,18 +162,22 @@ public class TestDataTransferProtocol ex
sendOut.writeInt(0); // zero checksum
//ok finally write a block with 0 len
- sendResponse(Status.SUCCESS, "", recvOut);
+ sendResponse(Status.SUCCESS, "", null, recvOut);
new PipelineAck(100, new Status[]{Status.SUCCESS}).write(recvOut);
sendRecvData(description, false);
}
private void sendResponse(Status status, String firstBadLink,
+ String message,
DataOutputStream out)
throws IOException {
Builder builder = BlockOpResponseProto.newBuilder().setStatus(status);
if (firstBadLink != null) {
builder.setFirstBadLink(firstBadLink);
}
+ if (message != null) {
+ builder.setMessage(message);
+ }
builder.build()
.writeDelimitedTo(out);
}
@@ -190,11 +190,11 @@ public class TestDataTransferProtocol ex
new DatanodeInfo[1], null, stage,
0, block.getNumBytes(), block.getNumBytes(), newGS);
if (eofExcepted) {
- sendResponse(Status.ERROR, null, recvOut);
+ sendResponse(Status.ERROR, null, null, recvOut);
sendRecvData(description, true);
} else if (stage == BlockConstructionStage.PIPELINE_CLOSE_RECOVERY) {
//ok finally write a block with 0 len
- sendResponse(Status.SUCCESS, "", recvOut);
+ sendResponse(Status.SUCCESS, "", null, recvOut);
sendRecvData(description, false);
} else {
writeZeroLengthPacket(block, description);
@@ -383,7 +383,7 @@ public class TestDataTransferProtocol ex
// bad bytes per checksum
sendOut.writeInt(-1-random.nextInt(oneMil));
recvBuf.reset();
- sendResponse(Status.ERROR, null, recvOut);
+ sendResponse(Status.ERROR, null, null, recvOut);
sendRecvData("wrong bytesPerChecksum while writing", true);
sendBuf.reset();
@@ -403,7 +403,7 @@ public class TestDataTransferProtocol ex
-1 - random.nextInt(oneMil)); // bad datalen
hdr.write(sendOut);
- sendResponse(Status.SUCCESS, "", recvOut);
+ sendResponse(Status.SUCCESS, "", null, recvOut);
new PipelineAck(100, new Status[]{Status.ERROR}).write(recvOut);
sendRecvData("negative DATA_CHUNK len while writing block " + newBlockId,
true);
@@ -428,7 +428,7 @@ public class TestDataTransferProtocol ex
sendOut.writeInt(0); // zero checksum
sendOut.flush();
//ok finally write a block with 0 len
- sendResponse(Status.SUCCESS, "", recvOut);
+ sendResponse(Status.SUCCESS, "", null, recvOut);
new PipelineAck(100, new Status[]{Status.SUCCESS}).write(recvOut);
sendRecvData("Writing a zero len block blockid " + newBlockId, false);
@@ -462,7 +462,7 @@ public class TestDataTransferProtocol ex
// negative length is ok. Datanode assumes we want to read the whole block.
recvBuf.reset();
- sendResponse(Status.SUCCESS, null, recvOut);
+ sendResponse(Status.SUCCESS, null, null, recvOut);
sendBuf.reset();
sender.readBlock(blk, BlockTokenSecretManager.DUMMY_TOKEN, "cl",
0L, -1L-random.nextInt(oneMil));
@@ -471,7 +471,11 @@ public class TestDataTransferProtocol ex
// length is more than size of block.
recvBuf.reset();
- sendResponse(Status.ERROR, null, recvOut);
+ sendResponse(Status.ERROR, null,
+ "opReadBlock " + firstBlock +
+ " received exception java.io.IOException: " +
+ "Offset 0 and length 4097 don't match block " + firstBlock + " (
blockLen 4096 )",
+ recvOut);
sendBuf.reset();
sender.readBlock(blk, BlockTokenSecretManager.DUMMY_TOKEN, "cl",
0L, fileLen+1);
