Revert "HDDS-1373. KeyOutputStream, close after write request fails a…

…fter retries, runs into IllegalArgumentException..(#729)"

This reverts commit df2ae27.

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/common/helpers/ExcludeList.java

@@ -102,10 +102,4 @@ public static ExcludeList getFromProtoBuf(
     });
     return excludeList;
   }
-
-  public void clear() {
-    datanodes.clear();
-    containerIds.clear();
-    pipelineIds.clear();
-  }
 }

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyOutputStream.java

@@ -295,66 +295,60 @@ private void handleWrite(byte[] b, int off, long len, boolean retry)
       throws IOException {
     int succeededAllocates = 0;
     while (len > 0) {
-      try {
-        if (streamEntries.size() <= currentStreamIndex) {
-          Preconditions.checkNotNull(omClient);
-          // allocate a new block, if a exception happens, log an error and
-          // throw exception to the caller directly, and the write fails.
-          try {
-            allocateNewBlock(currentStreamIndex);
-            succeededAllocates += 1;
-          } catch (IOException ioe) {
-            LOG.error("Try to allocate more blocks for write failed, already "
-                + "allocated " + succeededAllocates
-                + " blocks for this write.");
-            throw ioe;
-          }
-        }
-        // in theory, this condition should never violate due the check above
-        // still do a sanity check.
-        Preconditions.checkArgument(currentStreamIndex < streamEntries.size());
-        BlockOutputStreamEntry current = streamEntries.get(currentStreamIndex);
-
-        // length(len) will be in int range if the call is happening through
-        // write API of blockOutputStream. Length can be in long range if it comes
-        // via Exception path.
-        int writeLen = Math.min((int) len, (int) current.getRemaining());
-        long currentPos = current.getWrittenDataLength();
+      if (streamEntries.size() <= currentStreamIndex) {
+        Preconditions.checkNotNull(omClient);
+        // allocate a new block, if a exception happens, log an error and
+        // throw exception to the caller directly, and the write fails.
         try {
-          if (retry) {
-            current.writeOnRetry(len);
-          } else {
-            current.write(b, off, writeLen);
-            offset += writeLen;
-          }
+          allocateNewBlock(currentStreamIndex);
+          succeededAllocates += 1;
         } catch (IOException ioe) {
-          // for the current iteration, totalDataWritten - currentPos gives the
-          // amount of data already written to the buffer
-
-          // In the retryPath, the total data to be written will always be equal
-          // to or less than the max length of the buffer allocated.
-          // The len specified here is the combined sum of the data length of
-          // the buffers
-          Preconditions.checkState(!retry || len <= streamBufferMaxSize);
-          int dataWritten = (int) (current.getWrittenDataLength() - currentPos);
-          writeLen = retry ? (int) len : dataWritten;
-          // In retry path, the data written is already accounted in offset.
-          if (!retry) {
-            offset += writeLen;
-          }
-          LOG.debug("writeLen {}, total len {}", writeLen, len);
-          handleException(current, currentStreamIndex, ioe);
+          LOG.error("Try to allocate more blocks for write failed, already "
+              + "allocated " + succeededAllocates + " blocks for this write.");
+          throw ioe;
         }
-        if (current.getRemaining() <= 0) {
-          // since the current block is already written close the stream.
-          handleFlushOrClose(StreamAction.FULL);
+      }
+      // in theory, this condition should never violate due the check above
+      // still do a sanity check.
+      Preconditions.checkArgument(currentStreamIndex < streamEntries.size());
+      BlockOutputStreamEntry current = streamEntries.get(currentStreamIndex);
+
+      // length(len) will be in int range if the call is happening through
+      // write API of blockOutputStream. Length can be in long range if it comes
+      // via Exception path.
+      int writeLen = Math.min((int)len, (int) current.getRemaining());
+      long currentPos = current.getWrittenDataLength();
+      try {
+        if (retry) {
+          current.writeOnRetry(len);
+        } else {
+          current.write(b, off, writeLen);
+          offset += writeLen;
+        }
+      } catch (IOException ioe) {
+        // for the current iteration, totalDataWritten - currentPos gives the
+        // amount of data already written to the buffer
+
+        // In the retryPath, the total data to be written will always be equal
+        // to or less than the max length of the buffer allocated.
+        // The len specified here is the combined sum of the data length of
+        // the buffers
+        Preconditions.checkState(!retry || len <= streamBufferMaxSize);
+        int dataWritten  = (int) (current.getWrittenDataLength() - currentPos);
+        writeLen = retry ? (int) len : dataWritten;
+        // In retry path, the data written is already accounted in offset.
+        if (!retry) {
+          offset += writeLen;
         }
-        len -= writeLen;
-        off += writeLen;
-      } catch (Exception e) {
-        markStreamClosed();
-        throw e;
+        LOG.debug("writeLen {}, total len {}", writeLen, len);
+        handleException(current, currentStreamIndex, ioe);
+      }
+      if (current.getRemaining() <= 0) {
+        // since the current block is already written close the stream.
+        handleFlushOrClose(StreamAction.FULL);
       }
+      len -= writeLen;
+      off += writeLen;
     }
   }
 
@@ -371,7 +365,7 @@ private void discardPreallocatedBlocks(long containerID,
     // pre allocated blocks available.
 
     // This will be called only to discard the next subsequent unused blocks
-    // in the streamEntryList.
+    // in the sreamEntryList.
     if (streamIndex < streamEntries.size()) {
       ListIterator<BlockOutputStreamEntry> streamEntryIterator =
           streamEntries.listIterator(streamIndex);
@@ -404,20 +398,6 @@ private void removeEmptyBlocks() {
       }
     }
   }
-
-  private void cleanup() {
-    if (excludeList != null) {
-      excludeList.clear();
-      excludeList = null;
-    }
-    if (bufferPool != null) {
-      bufferPool.clearBufferPool();
-    }
-
-    if (streamEntries != null) {
-      streamEntries.clear();
-    }
-  }
   /**
    * It performs following actions :
    * a. Updates the committed length at datanode for the current stream in
@@ -438,7 +418,8 @@ private void handleException(BlockOutputStreamEntry streamEntry,
       closedContainerException = checkIfContainerIsClosed(t);
     }
     PipelineID pipelineId = null;
-    long totalSuccessfulFlushedData = streamEntry.getTotalAckDataLength();
+    long totalSuccessfulFlushedData =
+        streamEntry.getTotalAckDataLength();
     //set the correct length for the current stream
     streamEntry.setCurrentPosition(totalSuccessfulFlushedData);
     long bufferedDataLen = computeBufferData();
@@ -469,8 +450,8 @@ private void handleException(BlockOutputStreamEntry streamEntry,
     if (closedContainerException) {
       // discard subsequent pre allocated blocks from the streamEntries list
       // from the closed container
-      discardPreallocatedBlocks(streamEntry.getBlockID().getContainerID(), null,
-          streamIndex + 1);
+      discardPreallocatedBlocks(streamEntry.getBlockID().getContainerID(),
+          null, streamIndex + 1);
     } else {
       // In case there is timeoutException or Watch for commit happening over
       // majority or the client connection failure to the leader in the
@@ -494,11 +475,6 @@ private void handleException(BlockOutputStreamEntry streamEntry,
     }
   }
 
-  private void markStreamClosed() {
-    cleanup();
-    closed = true;
-  }
-
   private void handleRetry(IOException exception, long len) throws IOException {
     RetryPolicy.RetryAction action;
     try {
@@ -610,46 +586,40 @@ private void handleFlushOrClose(StreamAction op) throws IOException {
       return;
     }
     while (true) {
-      try {
-        int size = streamEntries.size();
-        int streamIndex =
-            currentStreamIndex >= size ? size - 1 : currentStreamIndex;
-        BlockOutputStreamEntry entry = streamEntries.get(streamIndex);
-        if (entry != null) {
-          try {
-            Collection<DatanodeDetails> failedServers =
-                entry.getFailedServers();
-            // failed servers can be null in case there is no data written in the
-            // stream
-            if (failedServers != null && !failedServers.isEmpty()) {
-              excludeList.addDatanodes(failedServers);
-            }
-            switch (op) {
-            case CLOSE:
+      int size = streamEntries.size();
+      int streamIndex =
+          currentStreamIndex >= size ? size - 1 : currentStreamIndex;
+      BlockOutputStreamEntry entry = streamEntries.get(streamIndex);
+      if (entry != null) {
+        try {
+          Collection<DatanodeDetails> failedServers = entry.getFailedServers();
+          // failed servers can be null in case there is no data written in the
+          // stream
+          if (failedServers != null && !failedServers.isEmpty()) {
+            excludeList.addDatanodes(failedServers);
+          }
+          switch (op) {
+          case CLOSE:
+            entry.close();
+            break;
+          case FULL:
+            if (entry.getRemaining() == 0) {
               entry.close();
-              break;
-            case FULL:
-              if (entry.getRemaining() == 0) {
-                entry.close();
-                currentStreamIndex++;
-              }
-              break;
-            case FLUSH:
-              entry.flush();
-              break;
-            default:
-              throw new IOException("Invalid Operation");
+              currentStreamIndex++;
             }
-          } catch (IOException ioe) {
-            handleException(entry, streamIndex, ioe);
-            continue;
+            break;
+          case FLUSH:
+            entry.flush();
+            break;
+          default:
+            throw new IOException("Invalid Operation");
           }
+        } catch (IOException ioe) {
+          handleException(entry, streamIndex, ioe);
+          continue;
         }
-        break;
-      } catch (Exception e) {
-        markStreamClosed();
-        throw e;
       }
+      break;
     }
   }
 
@@ -688,7 +658,7 @@ public void close() throws IOException {
     } catch (IOException ioe) {
       throw ioe;
     } finally {
-      cleanup();
+      bufferPool.clearBufferPool();
     }
   }
 

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

@@ -189,7 +189,6 @@ public void testBufferCaching() throws Exception {
     // flush ensures watchForCommit updates the total length acknowledged
     Assert.assertEquals(dataLength, blockOutputStream.getTotalAckDataLength());
 
-    Assert.assertEquals(1, keyOutputStream.getStreamEntries().size());
     // now close the stream, It will update the ack length after watchForCommit
     key.close();
 
@@ -209,7 +208,7 @@ public void testBufferCaching() throws Exception {
         .assertEquals(0, blockOutputStream.getBufferPool().computeBufferData());
     Assert.assertEquals(dataLength, blockOutputStream.getTotalAckDataLength());
     Assert.assertNull(blockOutputStream.getCommitIndex2flushedDataMap());
-    Assert.assertEquals(0, keyOutputStream.getStreamEntries().size());
+    Assert.assertEquals(1, keyOutputStream.getStreamEntries().size());
     validateData(keyName, data1);
   }
 
@@ -264,7 +263,6 @@ public void testFlushChunk() throws Exception {
     // Now do a flush. This will flush the data and update the flush length and
     // the map.
     key.flush();
-    Assert.assertEquals(1, keyOutputStream.getStreamEntries().size());
     // flush is a sync call, all pending operations will complete
     Assert.assertEquals(pendingWriteChunkCount,
         metrics.getContainerOpsMetrics(ContainerProtos.Type.WriteChunk));
@@ -304,7 +302,7 @@ public void testFlushChunk() throws Exception {
         metrics.getContainerOpCountMetrics(ContainerProtos.Type.PutBlock));
     Assert.assertEquals(totalOpCount + 3,
         metrics.getTotalOpCount());
-    Assert.assertEquals(0, keyOutputStream.getStreamEntries().size());
+    Assert.assertEquals(1, keyOutputStream.getStreamEntries().size());
     validateData(keyName, data1);
   }
 
@@ -399,7 +397,6 @@ public void testMultiChunkWrite() throws Exception {
         metrics.getContainerOpCountMetrics(ContainerProtos.Type.PutBlock));
     Assert.assertEquals(totalOpCount + 3,
         metrics.getTotalOpCount());
-    Assert.assertEquals(0, keyOutputStream.getStreamEntries().size());
     validateData(keyName, data1);
   }
 
@@ -457,7 +454,6 @@ public void testMultiChunkWrite2() throws Exception {
         blockOutputStream.getCommitIndex2flushedDataMap().size());
 
     Assert.assertEquals(0, blockOutputStream.getTotalAckDataLength());
-    Assert.assertEquals(1, keyOutputStream.getStreamEntries().size());
     key.close();
     Assert.assertEquals(pendingWriteChunkCount,
         metrics.getContainerOpsMetrics(ContainerProtos.Type.WriteChunk));
@@ -475,7 +471,7 @@ public void testMultiChunkWrite2() throws Exception {
         .assertEquals(0, blockOutputStream.getBufferPool().computeBufferData());
     Assert.assertEquals(dataLength, blockOutputStream.getTotalAckDataLength());
     Assert.assertNull(blockOutputStream.getCommitIndex2flushedDataMap());
-    Assert.assertEquals(0, keyOutputStream.getStreamEntries().size());
+    Assert.assertEquals(1, keyOutputStream.getStreamEntries().size());
     validateData(keyName, data1);
   }
 
@@ -540,7 +536,6 @@ public void testFullBufferCondition() throws Exception {
     // Now do a flush. This will flush the data and update the flush length and
     // the map.
     key.flush();
-    Assert.assertEquals(1, keyOutputStream.getStreamEntries().size());
     Assert.assertEquals(pendingWriteChunkCount,
         metrics.getContainerOpsMetrics(ContainerProtos.Type.WriteChunk));
     Assert.assertEquals(pendingPutBlockCount,
@@ -575,7 +570,7 @@ public void testFullBufferCondition() throws Exception {
         .assertEquals(0, blockOutputStream.getBufferPool().computeBufferData());
     Assert.assertEquals(dataLength, blockOutputStream.getTotalAckDataLength());
     Assert.assertNull(blockOutputStream.getCommitIndex2flushedDataMap());
-    Assert.assertEquals(0, keyOutputStream.getStreamEntries().size());
+    Assert.assertEquals(1, keyOutputStream.getStreamEntries().size());
     validateData(keyName, data1);
   }
 
@@ -643,7 +638,6 @@ public void testWriteWithExceedingMaxBufferLimit() throws Exception {
     // Now do a flush. This will flush the data and update the flush length and
     // the map.
     key.flush();
-    Assert.assertEquals(1, keyOutputStream.getStreamEntries().size());
     Assert.assertEquals(pendingWriteChunkCount,
         metrics.getContainerOpsMetrics(ContainerProtos.Type.WriteChunk));
     Assert.assertEquals(pendingPutBlockCount,
@@ -679,7 +673,7 @@ public void testWriteWithExceedingMaxBufferLimit() throws Exception {
         metrics.getTotalOpCount());
     Assert.assertEquals(dataLength, blockOutputStream.getTotalAckDataLength());
     Assert.assertNull(blockOutputStream.getCommitIndex2flushedDataMap());
-    Assert.assertEquals(0, keyOutputStream.getStreamEntries().size());
+    Assert.assertEquals(1, keyOutputStream.getStreamEntries().size());
     validateData(keyName, data1);
   }