HDDS-10377. Allow datanodes to do chunk level modifications to closed containers.

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java

@@ -478,6 +478,45 @@ public static XceiverClientReply writeChunkAsync(
     return xceiverClient.sendCommandAsync(request);
   }
 
+  /**
+   * Calls the container protocol to write a chunk for a closed container.
+   *
+   * @param xceiverClient client to perform call
+   * @param chunk information about chunk to write
+   * @param blockID ID of the block
+   * @param data the data of the chunk to write
+   * @param tokenString serialized block token
+   * @throws IOException if there is an I/O error while performing the call
+   */
+  public static XceiverClientReply writeChunkForClosedContainer(
+      XceiverClientSpi xceiverClient, ChunkInfo chunk, BlockID blockID,
+      ByteString data, String tokenString, int replicationIndex)
+      throws IOException, ExecutionException, InterruptedException {
+    WriteChunkRequestProto.Builder writeChunkRequest =
+        WriteChunkRequestProto.newBuilder()
+            .setBlockID(DatanodeBlockID.newBuilder()
+                .setContainerID(blockID.getContainerID())
+                .setLocalID(blockID.getLocalID())
+                .setBlockCommitSequenceId(blockID.getBlockCommitSequenceId())
+                .setReplicaIndex(replicationIndex)
+                .build())
+            .setChunkData(chunk)
+            .setData(data);
+    String id = xceiverClient.getPipeline().getFirstNode().getUuidString();
+    ContainerCommandRequestProto.Builder builder =
+        ContainerCommandRequestProto.newBuilder()
+            .setCmdType(Type.WriteChunkForClosedContainer)
+            .setContainerID(blockID.getContainerID())
+            .setDatanodeUuid(id)
+            .setWriteChunk(writeChunkRequest);
+
+    if (tokenString != null) {
+      builder.setEncodedToken(tokenString);
+    }
+    ContainerCommandRequestProto request = builder.build();
+    return xceiverClient.sendCommandAsync(request);
+  }
+
   /**
    * Allows writing a small file using single RPC. This takes the container
    * name, block name and data to write sends all that data to the container

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/audit/DNAction.java

@@ -42,7 +42,8 @@ public enum DNAction implements AuditAction {
   STREAM_INIT,
   FINALIZE_BLOCK,
   ECHO,
-  GET_CONTAINER_MERKLE_TREE;
+  GET_CONTAINER_MERKLE_TREE,
+  WRITE_CHUNK_FOR_CLOSED_CONTAINER;
 
   @Override
   public String getAction() {

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/checksum/DNContainerOperationClient.java

@@ -18,12 +18,14 @@
 package org.apache.hadoop.ozone.container.checksum;
 
 import com.google.common.collect.ImmutableList;
+import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.client.StandaloneReplicationConfig;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.scm.XceiverClientManager;
+import org.apache.hadoop.hdds.scm.XceiverClientReply;
 import org.apache.hadoop.hdds.scm.XceiverClientSpi;
 import org.apache.hadoop.hdds.scm.client.ClientTrustManager;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
@@ -44,6 +46,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.concurrent.ExecutionException;
 
 import static org.apache.hadoop.ozone.container.common.helpers.TokenHelper.encode;
 
@@ -104,6 +107,21 @@ public ByteString getContainerMerkleTree(long containerId, DatanodeDetails dn)
     }
   }
 
+  public XceiverClientReply writeChunkForClosedContainer(ContainerProtos.ChunkInfo chunkInfo, BlockID blockID,
+                                                         ByteString data, int replicationIndex, DatanodeDetails dn)
+      throws IOException, ExecutionException, InterruptedException {
+    XceiverClientSpi xceiverClient = this.xceiverClientManager.acquireClient(createSingleNodePipeline(dn));
+    // Set OzoneConsts.CHUNK_OVERWRITE
+    try {
+      String containerToken = encode(tokenHelper.getContainerToken(
+          ContainerID.valueOf(blockID.getContainerID())));
+      return ContainerProtocolCalls.writeChunkForClosedContainer(
+          xceiverClient, chunkInfo, blockID, data, containerToken, replicationIndex);
+    } finally {
+      this.xceiverClientManager.releaseClient(xceiverClient, false);
+    }
+  }
+
   public static Pipeline createSingleNodePipeline(DatanodeDetails dn) {
     return Pipeline.newBuilder()
         .setNodes(ImmutableList.of(dn))

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java

@@ -833,6 +833,7 @@ private static DNAction getAuditAction(Type cmdType) {
     case FinalizeBlock    : return DNAction.FINALIZE_BLOCK;
     case Echo             : return DNAction.ECHO;
     case GetContainerMerkleTree : return DNAction.GET_CONTAINER_MERKLE_TREE;
+    case WriteChunkForClosedContainer: return DNAction.WRITE_CHUNK_FOR_CLOSED_CONTAINER;
     default :
       LOG.debug("Invalid command type - {}", cmdType);
       return null;

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java

@@ -109,6 +109,7 @@
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.IO_EXCEPTION;
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.PUT_SMALL_FILE_ERROR;
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.UNSUPPORTED_REQUEST;
+import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.UNCLOSED_CONTAINER_IO;
 import static org.apache.hadoop.hdds.scm.protocolPB.ContainerCommandResponseBuilders.getBlockDataResponse;
 import static org.apache.hadoop.hdds.scm.protocolPB.ContainerCommandResponseBuilders.getBlockLengthResponse;
 import static org.apache.hadoop.hdds.scm.protocolPB.ContainerCommandResponseBuilders.getEchoResponse;
@@ -299,6 +300,8 @@ static ContainerCommandResponseProto dispatchRequest(KeyValueHandler handler,
       return handler.handleEcho(request, kvContainer);
     case GetContainerMerkleTree:
       return handler.handleGetContainerMerkleTree(request, kvContainer);
+    case WriteChunkForClosedContainer:
+      return handler.handleWriteChunkForClosedContainer(request, kvContainer);
     default:
       return null;
     }
@@ -970,6 +973,47 @@ ContainerCommandResponseProto handleWriteChunk(
     return getWriteChunkResponseSuccess(request, blockDataProto);
   }
 
+  /**
+   * Handle Write Chunk operation for closed container. Calls ChunkManager to process the request.
+   */
+  ContainerCommandResponseProto handleWriteChunkForClosedContainer(
+      ContainerCommandRequestProto request, KeyValueContainer kvContainer) {
+
+    if (!request.hasWriteChunk()) {
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Malformed Write Chunk request. trace ID: {}",
+            request.getTraceID());
+      }
+      return malformedRequest(request);
+    }
+
+    try {
+      checkContainerClose(kvContainer);
+
+      WriteChunkRequestProto writeChunk = request.getWriteChunk();
+      BlockID blockID = BlockID.getFromProtobuf(writeChunk.getBlockID());
+      ContainerProtos.ChunkInfo chunkInfoProto = writeChunk.getChunkData();
+
+      ChunkInfo chunkInfo = ChunkInfo.getFromProtoBuf(chunkInfoProto);
+      Preconditions.checkNotNull(chunkInfo);
+
+      DispatcherContext dispatcherContext =
+          DispatcherContext.getHandleWriteChunk();
+      ChunkBuffer data =
+          ChunkBuffer.wrap(writeChunk.getData().asReadOnlyByteBufferList());
+
+      chunkManager.writeChunk(kvContainer, blockID, chunkInfo, data,
+          dispatcherContext);
+    } catch (StorageContainerException ex) {
+      return ContainerUtils.logAndReturnError(LOG, ex, request);
+    } catch (IOException ex) {
+      return ContainerUtils.logAndReturnError(LOG,
+          new StorageContainerException("Write Chunk failed", ex, IO_EXCEPTION),
+          request);
+    }
+    return getSuccessResponse(request);
+  }
+
   /**
    * Handle Put Small File operation. Writes the chunk and associated key
    * using a single RPC. Calls BlockManager and ChunkManager to process the
@@ -1146,6 +1190,38 @@ private void checkContainerOpen(KeyValueContainer kvContainer)
     throw new StorageContainerException(msg, result);
   }
 
+  /**
+   * Check if container is Closed. Throw exception otherwise.
+   * @param kvContainer
+   * @throws StorageContainerException
+   */
+  private void checkContainerClose(KeyValueContainer kvContainer)
+      throws StorageContainerException {
+
+    final State containerState = kvContainer.getContainerState();
+    if (containerState == State.QUASI_CLOSED || containerState == State.CLOSED
+        || containerState == State.UNHEALTHY) {
+      return;
+    }
+
+    final ContainerProtos.Result result;
+    switch (containerState) {
+    case OPEN:
+    case CLOSING:
+    case RECOVERING:
+      result = UNCLOSED_CONTAINER_IO;
+      break;
+    case INVALID:
+      result = INVALID_CONTAINER_STATE;
+      break;
+    default:
+      result = CONTAINER_INTERNAL_ERROR;
+    }
+    String msg = "Requested operation not allowed as ContainerState is " +
+        containerState;
+    throw new StorageContainerException(msg, result);
+  }
+
   @Override
   public Container importContainer(ContainerData originalContainerData,
       final InputStream rawContainerStream,

hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto

@@ -108,6 +108,7 @@ enum Type {
   FinalizeBlock = 21;
   Echo = 22;
   GetContainerMerkleTree = 23;
+  WriteChunkForClosedContainer = 24;
 }