HDDS-11998. BlockDataStreamOutput should decrPendingContainerOpsMetrics (apache#7636)

Author: ivandika3

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java

@@ -253,7 +253,7 @@ public void write(ByteBuffer b, int off, int len) throws IOException {
     }
     while (len > 0) {
       allocateNewBufferIfNeeded();
-      int writeLen = Math.min(len, currentBuffer.length());
+      int writeLen = Math.min(len, currentBuffer.remaining());
       final StreamBuffer buf = new StreamBuffer(b, off, writeLen);
       currentBuffer.put(buf);
       writeChunkIfNeeded();
@@ -265,7 +265,7 @@ public void write(ByteBuffer b, int off, int len) throws IOException {
   }
 
   private void writeChunkIfNeeded() throws IOException {
-    if (currentBuffer.length() == 0) {
+    if (currentBuffer.remaining() == 0) {
       writeChunk(currentBuffer);
       currentBuffer = null;
     }
@@ -672,6 +672,7 @@ private void writeChunkToContainer(ByteBuffer buf)
             out.writeAsync(buf, StandardWriteOption.SYNC) :
             out.writeAsync(buf))
             .whenCompleteAsync((r, e) -> {
+              metrics.decrPendingContainerOpsMetrics(ContainerProtos.Type.WriteChunk);
               if (e != null || !r.isSuccess()) {
                 if (e == null) {
                   e = new IOException("result is not success");

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBuffer.java

@@ -39,8 +39,8 @@ public ByteBuffer duplicate() {
     return buffer.duplicate();
   }
 
-  public int length() {
-    return buffer.limit() - buffer.position();
+  public int remaining() {
+    return buffer.remaining();
   }
 
   public int position() {

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockDataStreamOutput.java

@@ -22,7 +22,6 @@
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.conf.StorageUnit;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
-import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.scm.OzoneClientConfig;
 import org.apache.hadoop.hdds.scm.XceiverClientManager;
 import org.apache.hadoop.hdds.scm.XceiverClientMetrics;
@@ -47,13 +46,14 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 
-import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type.PutBlock;
+import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type.WriteChunk;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_STALENODE_INTERVAL;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -77,13 +77,6 @@ public class TestBlockDataStreamOutput {
   private static String keyString;
   private static final DatanodeVersion DN_OLD_VERSION = DatanodeVersion.SEPARATE_RATIS_PORTS_AVAILABLE;
 
-  /**
-   * Create a MiniDFSCluster for testing.
-   * <p>
-   * Ozone is made active by setting OZONE_ENABLED = true
-   *
-   * @throws IOException
-   */
   @BeforeAll
   public static void init() throws Exception {
     chunkSize = 100;
@@ -120,7 +113,7 @@ public static void init() throws Exception {
     client = OzoneClientFactory.getRpcClient(conf);
     objectStore = client.getObjectStore();
     keyString = UUID.randomUUID().toString();
-    volumeName = "testblockoutputstream";
+    volumeName = "testblockdatastreamoutput";
     bucketName = volumeName;
     objectStore.createVolume(volumeName);
     objectStore.getVolume(volumeName).createBucket(bucketName);
@@ -130,9 +123,6 @@ static String getKeyName() {
     return UUID.randomUUID().toString();
   }
 
-  /**
-   * Shutdown MiniDFSCluster.
-   */
   @AfterAll
   public static void shutdown() {
     IOUtils.closeQuietly(client);
@@ -166,6 +156,11 @@ public void testMultiBlockWrite() throws Exception {
   }
 
   static void testWrite(int dataLength) throws Exception {
+    XceiverClientMetrics metrics =
+        XceiverClientManager.getXceiverClientMetrics();
+    long pendingWriteChunkCount = metrics.getPendingContainerOpCountMetrics(WriteChunk);
+    long pendingPutBlockCount = metrics.getPendingContainerOpCountMetrics(PutBlock);
+
     String keyName = getKeyName();
     OzoneDataStreamOutput key = createKey(
         keyName, ReplicationType.RATIS, dataLength);
@@ -174,9 +169,19 @@ static void testWrite(int dataLength) throws Exception {
     // now close the stream, It will update the key length.
     key.close();
     validateData(keyName, data);
+
+    assertEquals(pendingPutBlockCount,
+        metrics.getPendingContainerOpCountMetrics(PutBlock));
+    assertEquals(pendingWriteChunkCount,
+        metrics.getPendingContainerOpCountMetrics(WriteChunk));
   }
 
   private void testWriteWithFailure(int dataLength) throws Exception {
+    XceiverClientMetrics metrics =
+        XceiverClientManager.getXceiverClientMetrics();
+    long pendingWriteChunkCount = metrics.getPendingContainerOpCountMetrics(WriteChunk);
+    long pendingPutBlockCount = metrics.getPendingContainerOpCountMetrics(PutBlock);
+
     String keyName = getKeyName();
     OzoneDataStreamOutput key = createKey(
         keyName, ReplicationType.RATIS, dataLength);
@@ -195,32 +200,50 @@ private void testWriteWithFailure(int dataLength) throws Exception {
     key.close();
     String dataString = new String(data, UTF_8);
     validateData(keyName, dataString.concat(dataString).getBytes(UTF_8));
+
+    assertEquals(pendingPutBlockCount,
+        metrics.getPendingContainerOpCountMetrics(PutBlock));
+    assertEquals(pendingWriteChunkCount,
+        metrics.getPendingContainerOpCountMetrics(WriteChunk));
   }
 
   @Test
   public void testPutBlockAtBoundary() throws Exception {
-    int dataLength = 500;
+    int dataLength = maxFlushSize + 100;
     XceiverClientMetrics metrics =
         XceiverClientManager.getXceiverClientMetrics();
-    long putBlockCount = metrics.getContainerOpCountMetrics(
-        ContainerProtos.Type.PutBlock);
-    long pendingPutBlockCount = metrics.getPendingContainerOpCountMetrics(
-        ContainerProtos.Type.PutBlock);
+    long writeChunkCount = metrics.getContainerOpCountMetrics(WriteChunk);
+    long putBlockCount = metrics.getContainerOpCountMetrics(PutBlock);
+    long pendingWriteChunkCount = metrics.getPendingContainerOpCountMetrics(WriteChunk);
+    long pendingPutBlockCount = metrics.getPendingContainerOpCountMetrics(PutBlock);
+    long totalOpCount = metrics.getTotalOpCount();
+
     String keyName = getKeyName();
     OzoneDataStreamOutput key = createKey(
         keyName, ReplicationType.RATIS, 0);
     byte[] data =
         ContainerTestHelper.getFixedLengthString(keyString, dataLength)
             .getBytes(UTF_8);
     key.write(ByteBuffer.wrap(data));
-    assertThat(metrics.getPendingContainerOpCountMetrics(ContainerProtos.Type.PutBlock))
+    assertThat(metrics.getPendingContainerOpCountMetrics(PutBlock))
         .isLessThanOrEqualTo(pendingPutBlockCount + 1);
+    assertThat(metrics.getPendingContainerOpCountMetrics(WriteChunk))
+        .isLessThanOrEqualTo(pendingWriteChunkCount + 5);
     key.close();
     // Since data length is 500 , first putBlock will be at 400(flush boundary)
     // and the other at 500
-    assertEquals(
-        metrics.getContainerOpCountMetrics(ContainerProtos.Type.PutBlock),
-        putBlockCount + 2);
+    assertEquals(putBlockCount + 2,
+        metrics.getContainerOpCountMetrics(PutBlock));
+    // Each chunk is 100 so there will be 500 / 100 = 5 chunks.
+    assertEquals(writeChunkCount + 5,
+        metrics.getContainerOpCountMetrics(WriteChunk));
+    assertEquals(totalOpCount + 7,
+        metrics.getTotalOpCount());
+    assertEquals(pendingPutBlockCount,
+        metrics.getPendingContainerOpCountMetrics(PutBlock));
+    assertEquals(pendingWriteChunkCount,
+        metrics.getPendingContainerOpCountMetrics(WriteChunk));
+
     validateData(keyName, data);
   }
 
@@ -242,20 +265,22 @@ public void testMinPacketSize() throws Exception {
     XceiverClientMetrics metrics =
         XceiverClientManager.getXceiverClientMetrics();
     OzoneDataStreamOutput key = createKey(keyName, ReplicationType.RATIS, 0);
-    long writeChunkCount =
-        metrics.getContainerOpCountMetrics(ContainerProtos.Type.WriteChunk);
+    long writeChunkCount = metrics.getContainerOpCountMetrics(WriteChunk);
+    long pendingWriteChunkCount = metrics.getPendingContainerOpCountMetrics(WriteChunk);
     byte[] data =
         ContainerTestHelper.getFixedLengthString(keyString, chunkSize / 2)
             .getBytes(UTF_8);
     key.write(ByteBuffer.wrap(data));
     // minPacketSize= 100, so first write of 50 wont trigger a writeChunk
     assertEquals(writeChunkCount,
-        metrics.getContainerOpCountMetrics(ContainerProtos.Type.WriteChunk));
+        metrics.getContainerOpCountMetrics(WriteChunk));
     key.write(ByteBuffer.wrap(data));
     assertEquals(writeChunkCount + 1,
-        metrics.getContainerOpCountMetrics(ContainerProtos.Type.WriteChunk));
+        metrics.getContainerOpCountMetrics(WriteChunk));
     // now close the stream, It will update the key length.
     key.close();
+    assertEquals(pendingWriteChunkCount,
+        metrics.getPendingContainerOpCountMetrics(WriteChunk));
     String dataString = new String(data, UTF_8);
     validateData(keyName, dataString.concat(dataString).getBytes(UTF_8));
   }