errose28 commented on code in PR #8914:
URL: https://github.com/apache/ozone/pull/8914#discussion_r2285923441
##########
hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/debug/replicas/chunk/ChunkKeyHandler.java:
##########
@@ -95,113 +90,155 @@ protected void execute(OzoneClient client, OzoneAddress
address)
}
ContainerLayoutVersion containerLayoutVersion = ContainerLayoutVersion
.getConfiguredVersion(getConf());
- ArrayNode responseArrayList = result.putArray("keyLocations");
- for (OmKeyLocationInfo keyLocation : locationInfos) {
- Pipeline keyPipeline = keyLocation.getPipeline();
- boolean isECKey =
- keyPipeline.getReplicationConfig().getReplicationType() ==
- HddsProtos.ReplicationType.EC;
- Pipeline pipeline;
- if (!isECKey && keyPipeline.getType() !=
HddsProtos.ReplicationType.STAND_ALONE) {
- pipeline = keyPipeline.copyForRead();
- } else {
- pipeline = keyPipeline;
- }
- XceiverClientSpi xceiverClient =
xceiverClientManager.acquireClientForReadData(pipeline);
- try {
- Map<DatanodeDetails, ContainerProtos.GetBlockResponseProto>
- responses =
- ContainerProtocolCalls.getBlockFromAllNodes(xceiverClient,
- keyLocation.getBlockID().getDatanodeBlockIDProtobuf(),
- keyLocation.getToken());
- Map<DatanodeDetails, ContainerProtos.ReadContainerResponseProto>
readContainerResponses =
- containerOperationClient.readContainerFromAllNodes(
- keyLocation.getContainerID(), pipeline);
- ArrayNode responseFromAllNodes = responseArrayList.addArray();
- for (Map.Entry<DatanodeDetails,
ContainerProtos.GetBlockResponseProto> entry : responses.entrySet()) {
- DatanodeDetails datanodeDetails = entry.getKey();
- GetBlockResponseProto blockResponse = entry.getValue();
-
- if (blockResponse == null || !blockResponse.hasBlockData()) {
- System.err.printf("GetBlock call failed on %s datanode and %s
block.%n",
- datanodeDetails.getHostName(), keyLocation.getBlockID());
- continue;
- }
- ContainerProtos.BlockData blockData = blockResponse.getBlockData();
- ContainerProtos.ChunkInfo chunkInfo = blockData.getChunksCount() >
0 ?
- blockData.getChunks(0) : null;
+ // Use Jackson streaming for all JSON generation
+ ObjectMapper mapper = new ObjectMapper();
+ JsonFactory jsonFactory = mapper.getFactory();
+
+ try (JsonGenerator jsonGen = jsonFactory.createGenerator(System.out)) {
+ jsonGen.useDefaultPrettyPrinter();
+
+ jsonGen.writeStartObject();
+ jsonGen.writeStringField("volumeName", volumeName);
+ jsonGen.writeStringField("bucketName", bucketName);
+ jsonGen.writeStringField("name", keyName);
+
+ // Start keyLocations array
+ jsonGen.writeArrayFieldStart("keyLocations");
+ for (OmKeyLocationInfo keyLocation : locationInfos) {
+ jsonGen.writeStartArray();
+
+ Pipeline keyPipeline = keyLocation.getPipeline();
+ boolean isECKey =
+ keyPipeline.getReplicationConfig().getReplicationType() ==
+ HddsProtos.ReplicationType.EC;
+ Pipeline pipeline;
+ if (!isECKey && keyPipeline.getType() !=
HddsProtos.ReplicationType.STAND_ALONE) {
+ pipeline = keyPipeline.copyForRead();
+ } else {
+ pipeline = keyPipeline;
+ }
+ XceiverClientSpi xceiverClient =
xceiverClientManager.acquireClientForReadData(pipeline);
+ try {
+ Map<DatanodeDetails, ContainerProtos.ReadContainerResponseProto>
readContainerResponses =
+
containerOperationClient.readContainerFromAllNodes(keyLocation.getContainerID(),
pipeline);
+
+ // Process each datanode individually
+ for (DatanodeDetails datanodeDetails : pipeline.getNodes()) {
+ try {
+ // Get block from THIS ONE datanode only
+ ContainerProtos.GetBlockResponseProto blockResponse =
+ ContainerProtocolCalls.getBlock(xceiverClient,
+ keyLocation.getBlockID(),
+ keyLocation.getToken(),
+ pipeline.getReplicaIndexes());
+
+ if (blockResponse == null || !blockResponse.hasBlockData()) {
+ System.err.printf("GetBlock call failed on %s datanode and
%s block.%n",
+ datanodeDetails.getHostName(), keyLocation.getBlockID());
+ continue;
+ }
+
+ ContainerProtos.BlockData blockData =
blockResponse.getBlockData();
+ ContainerProtos.ChunkInfo chunkInfo =
blockData.getChunksCount() > 0 ?
+ blockData.getChunks(0) : null;
- String fileName = "";
- if (chunkInfo != null) {
- ContainerProtos.ContainerDataProto containerData =
-
readContainerResponses.get(datanodeDetails).getContainerData();
- fileName = containerLayoutVersion.getChunkFile(new File(
+ String fileName = "";
+ if (chunkInfo != null) {
+ ContainerProtos.ContainerDataProto containerData =
+
readContainerResponses.get(datanodeDetails).getContainerData();
+ fileName = containerLayoutVersion.getChunkFile(new File(
getChunkLocationPath(containerData.getContainerPath())),
- keyLocation.getBlockID(),
- chunkInfo.getChunkName()).toString();
- }
-
- ObjectNode jsonObj = responseFromAllNodes.addObject();
- ObjectNode dnObj = jsonObj.putObject("datanode");
- dnObj.put("hostname", datanodeDetails.getHostName());
- dnObj.put("ip", datanodeDetails.getIpAddress());
- dnObj.put("uuid", datanodeDetails.getUuidString());
-
- jsonObj.put("file", fileName);
-
- ObjectNode blockDataNode = jsonObj.putObject("blockData");
- ObjectNode blockIdNode = blockDataNode.putObject("blockID");
- blockIdNode.put("containerID",
blockData.getBlockID().getContainerID());
- blockIdNode.put("localID", blockData.getBlockID().getLocalID());
- blockIdNode.put("blockCommitSequenceId",
blockData.getBlockID().getBlockCommitSequenceId());
- blockDataNode.put("size", blockData.getSize());
-
- ArrayNode chunkArray = blockDataNode.putArray("chunks");
- for (ContainerProtos.ChunkInfo chunk : blockData.getChunksList()) {
- ObjectNode chunkNode = chunkArray.addObject();
- chunkNode.put("offset", chunk.getOffset());
- chunkNode.put("len", chunk.getLen());
-
- if (chunk.hasChecksumData()) {
- ArrayNode checksums = chunkNode.putArray("checksums");
- for (ByteString bs :
chunk.getChecksumData().getChecksumsList()) {
- checksums.add(StringUtils.byteToHexString(bs.toByteArray()));
+ keyLocation.getBlockID(),
+ chunkInfo.getChunkName()).toString();
}
- chunkNode.put("checksumType",
chunk.getChecksumData().getType().name());
- chunkNode.put("bytesPerChecksum",
chunk.getChecksumData().getBytesPerChecksum());
- }
- if (chunk.hasStripeChecksum()) {
- byte[] stripeBytes = chunk.getStripeChecksum().toByteArray();
- int checksumLen =
chunk.getChecksumData().getChecksumsList().get(0).size();
+ // Start writing this datanode's response object
+ jsonGen.writeStartObject();
+
+ jsonGen.writeObjectFieldStart("datanode");
+ jsonGen.writeStringField("hostname",
datanodeDetails.getHostName());
+ jsonGen.writeStringField("ip", datanodeDetails.getIpAddress());
+ jsonGen.writeStringField("uuid",
datanodeDetails.getUuidString());
+ jsonGen.writeEndObject();
+
+ jsonGen.writeStringField("file", fileName);
+
+ // Write block data
+ jsonGen.writeObjectFieldStart("blockData");
+ jsonGen.writeObjectFieldStart("blockID");
+ jsonGen.writeNumberField("containerID",
blockData.getBlockID().getContainerID());
+ jsonGen.writeNumberField("localID",
blockData.getBlockID().getLocalID());
+ jsonGen.writeNumberField("blockCommitSequenceId",
blockData.getBlockID().getBlockCommitSequenceId());
+ jsonGen.writeEndObject();
+
+ jsonGen.writeNumberField("size", blockData.getSize());
+
+ // Write chunks array
+ jsonGen.writeArrayFieldStart("chunks");
+ for (ContainerProtos.ChunkInfo chunk :
blockData.getChunksList()) {
+ jsonGen.writeStartObject();
+ jsonGen.writeNumberField("offset", chunk.getOffset());
+ jsonGen.writeNumberField("len", chunk.getLen());
+
+ if (chunk.hasChecksumData()) {
+ jsonGen.writeArrayFieldStart("checksums");
+ for (ByteString bs :
chunk.getChecksumData().getChecksumsList()) {
+
jsonGen.writeString(StringUtils.byteToHexString(bs.toByteArray()));
+ }
+
+ jsonGen.writeEndArray();
+ jsonGen.writeStringField("checksumType",
chunk.getChecksumData().getType().name());
+ jsonGen.writeNumberField("bytesPerChecksum",
chunk.getChecksumData().getBytesPerChecksum());
+ }
+
+ if (chunk.hasStripeChecksum()) {
+ byte[] stripeBytes =
chunk.getStripeChecksum().toByteArray();
+ int checksumLen =
chunk.getChecksumData().getChecksumsList().get(0).size();
+
+ jsonGen.writeArrayFieldStart("stripeChecksum");
+ for (int i = 0; i <= stripeBytes.length - checksumLen; i
+= checksumLen) {
+ byte[] slice = Arrays.copyOfRange(stripeBytes, i, i +
checksumLen);
+ jsonGen.writeString(StringUtils.byteToHexString(slice));
+ }
+ jsonGen.writeEndArray();
+ }
+ jsonGen.writeEndObject();
+ }
- ArrayNode stripeChecksums =
chunkNode.putArray("stripeChecksum");
- for (int i = 0; i <= stripeBytes.length - checksumLen; i +=
checksumLen) {
- byte[] slice = Arrays.copyOfRange(stripeBytes, i, i +
checksumLen);
- stripeChecksums.add(StringUtils.byteToHexString(slice));
+ jsonGen.writeEndArray(); // End chunks array
+ jsonGen.writeEndObject(); // End blockData object
+
+ if (isECKey) {
+ int replicaIndex =
keyPipeline.getReplicaIndex(datanodeDetails);
+ int dataCount = ((ECReplicationConfig)
keyPipeline.getReplicationConfig()).getData();
+ // Index is 1-based,
+ // e.g. for RS-3-2 we will have data indexes 1,2,3 and
parity indexes 4,5
+ ChunkType chunkType = (replicaIndex > dataCount) ?
ChunkType.PARITY : ChunkType.DATA;
+ jsonGen.writeStringField("chunkType", chunkType.name());
+ jsonGen.writeNumberField("replicaIndex", replicaIndex);
}
- }
- }
+ jsonGen.writeEndObject(); // End this datanode's response
object
- if (isECKey) {
- int replicaIndex = keyPipeline.getReplicaIndex(entry.getKey());
- int dataCount = ((ECReplicationConfig)
keyPipeline.getReplicationConfig()).getData();
- // Index is 1-based,
- // e.g. for RS-3-2 we will have data indexes 1,2,3 and parity
indexes 4,5
- ChunkType chunkType = (replicaIndex > dataCount) ?
ChunkType.PARITY : ChunkType.DATA;
- jsonObj.put("chunkType", chunkType.name());
- jsonObj.put("replicaIndex", replicaIndex);
+ jsonGen.flush();
+ } catch (Exception e) {
+ System.err.printf("Error getting block from datanode %s: %s%n",
+ datanodeDetails.getHostName(), e.getMessage());
+ }
}
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ } finally {
+ xceiverClientManager.releaseClientForReadData(xceiverClient,
false);
}
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- } finally {
- xceiverClientManager.releaseClientForReadData(xceiverClient, false);
+
+ jsonGen.writeEndArray();
}
+
+ jsonGen.writeEndArray(); // End keyLocations array
+ jsonGen.writeEndObject(); // End root object
+ jsonGen.flush();
Review Comment:
We should print a newline here so the shell prompt does not end up on the
same line as the closing json bracket. Putting it inside the try-with-resources
should work. Once this block ends jackson will close stdout and we won't be
able to print anything. Double check this in your shell though.
```suggestion
jsonGen.flush();
System.out.println();
```
##########
hadoop-ozone/tools/pom.xml:
##########
@@ -38,6 +38,10 @@
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-annotations</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </dependency>
Review Comment:
Do we need to add this? The project builds fine for me without it. The
existing Jackson dependencies like `jackson-databind` should bring in
`jackson-core` transitively.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
