ashishkumar50 commented on code in PR #5663:
URL: https://github.com/apache/ozone/pull/5663#discussion_r1474480506
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java:
##########
@@ -758,6 +790,156 @@ CompletableFuture<ContainerCommandResponseProto>
writeChunkToContainer(
return null;
}
+ /**
+ * Update container block data, which is later sent to DataNodes via
PutBlock,
+ * using the new chunks sent out via WriteChunk.
+ *
+ * This method is only used when incremental chunk list is enabled.
+ * @param chunk the chunk buffer to be sent out by WriteChunk.
+ * @throws OzoneChecksumException
+ */
+ private void updateBlockDataForWriteChunk(ChunkBuffer chunk)
+ throws OzoneChecksumException {
+ // Update lastChunkBuffer using the new chunk data.
+ // This is used to calculate checksum for the last partial chunk in
+ // containerBlockData which will used by PutBlock.
+
+ // the last partial chunk in containerBlockData will be replaced.
+ // So remove it.
+ removeLastPartialChunk();
+ chunk.rewind();
+ LOG.debug("Adding chunk pos {} limit {} remaining {}." +
+ "lastChunkBuffer pos {} limit {} remaining {} lastChunkOffset =
{}",
+ chunk.position(), chunk.limit(), chunk.remaining(),
+ lastChunkBuffer.position(), lastChunkBuffer.limit(),
+ lastChunkBuffer.remaining(), lastChunkOffset);
+
+ // Append the chunk to the last chunk buffer.
+ // if the resulting size exceeds limit (4MB),
+ // drop the full chunk and keep the rest.
+ if (lastChunkBuffer.position() + chunk.remaining() <=
+ lastChunkBuffer.capacity()) {
+ appendLastChunkBuffer(chunk, 0, chunk.remaining());
+ } else {
+ int remainingBufferSize =
+ lastChunkBuffer.capacity() - lastChunkBuffer.position();
+ appendLastChunkBuffer(chunk, 0, remainingBufferSize);
+ updateBlockDataWithLastChunkBuffer();
+ appendLastChunkBuffer(chunk, remainingBufferSize,
+ chunk.remaining() - remainingBufferSize);
+ }
+ LOG.debug("after append, lastChunkBuffer={} lastChunkOffset={}",
+ lastChunkBuffer, lastChunkOffset);
+
+ updateBlockDataWithLastChunkBuffer();
+ }
+
+ private void updateBlockDataWithLastChunkBuffer()
+ throws OzoneChecksumException {
+ // create chunk info for lastChunkBuffer
+ ChunkInfo lastChunkInfo = createChunkInfo(lastChunkOffset);
+ LOG.debug("lastChunkInfo = {}", lastChunkInfo);
+ long lastChunkSize = lastChunkInfo.getLen();
+ addToBlockData(lastChunkInfo);
+
+ lastChunkBuffer.clear();
+ if (lastChunkSize == config.getStreamBufferSize()) {
+ lastChunkOffset += config.getStreamBufferSize();
+ } else {
+ lastChunkBuffer.position((int) lastChunkSize);
+ }
+ }
+
+ private void appendLastChunkBuffer(ChunkBuffer chunkBuffer, int offset,
+ int length) {
+ LOG.debug("copying to last chunk buffer offset={} length={}",
+ offset, length);
+ int pos = 0;
+ int uncopied = length;
+ for (ByteBuffer bb : chunkBuffer.asByteBufferList()) {
+ if (pos + bb.remaining() > offset) {
+ int copyStart = offset < pos ? 0 : offset - pos;
+ int copyLen = Math.min(uncopied, bb.remaining());
+ try {
+ LOG.debug("put into last chunk buffer start = {} len = {}",
+ copyStart, copyLen);
+ lastChunkBuffer.put(bb.array(), copyStart, copyLen);
+ } catch (BufferOverflowException e) {
+ LOG.error("appending from " + copyStart + " for len=" + copyLen +
+ ". lastChunkBuffer remaining=" + lastChunkBuffer.remaining() +
+ " pos=" + lastChunkBuffer.position() +
+ " limit=" + lastChunkBuffer.limit() +
+ " capacity=" + lastChunkBuffer.capacity());
+ throw e;
+ }
+
+ uncopied -= copyLen;
+ }
+
+ pos += bb.remaining();
+ if (pos > offset + length) {
Review Comment:
Here it should be ">=" as already copied till length?
--
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]
