HDFS-17289. Considering the size of non-lastBlocks equals to complete block size can cause append failure. (#6357). Contributed by farmmamba.

Reviewed-by: Haiyang Hu <haiyang.hu@shopee.com>
Reviewed-by: huangzhaobo <huangzhaobo99@126.com>
Signed-off-by:  Shuyan Zhang <zhangshuyan@apache.org>

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java

@@ -356,7 +356,7 @@ public class DFSOutputStream extends FSOutputSummer
       streamer = new DataStreamer(lastBlock, stat, dfsClient, src, progress,
           checksum, cachingStrategy, byteArrayManager);
       getStreamer().setBytesCurBlock(lastBlock.getBlockSize());
-      adjustPacketChunkSize(stat);
+      adjustPacketChunkSize(lastBlock);
       getStreamer().setPipelineInConstruction(lastBlock);
     } else {
       computePacketChunkSize(dfsClient.getConf().getWritePacketSize(),
@@ -368,14 +368,14 @@ public class DFSOutputStream extends FSOutputSummer
     }
   }
 
-  private void adjustPacketChunkSize(HdfsFileStatus stat) throws IOException{
+  private void adjustPacketChunkSize(LocatedBlock lastBlock) throws IOException{
 
-    long usedInLastBlock = stat.getLen() % blockSize;
+    long usedInLastBlock = lastBlock.getBlockSize();
     int freeInLastBlock = (int)(blockSize - usedInLastBlock);
 
     // calculate the amount of free space in the pre-existing
     // last crc chunk
-    int usedInCksum = (int)(stat.getLen() % bytesPerChecksum);
+    int usedInCksum = (int)(lastBlock.getBlockSize() % bytesPerChecksum);
     int freeInCksum = bytesPerChecksum - usedInCksum;
 
     // if there is space in the last block, then we have to

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestFileAppend3.java

@@ -590,4 +590,40 @@ public class TestFileAppend3  {
   public void testAppendToPartialChunkforAppend2() throws IOException {
     testAppendToPartialChunk(true);
   }
+
+  @Test
+  public void testApppendToPartialChunkWithMiddleBlockNotComplete() throws IOException {
+    final Path p = new Path("/TC8/foo");
+
+    //a. Create file and write one block of data. Close file.
+    final int len1 = (int) BLOCK_SIZE;
+    try (FSDataOutputStream out = fs.create(p, false, buffersize, REPLICATION,
+        BLOCK_SIZE)) {
+      AppendTestUtil.write(out, 0, len1);
+    }
+
+    // Reopen file to append. This length is not the multiple of 512.
+    final int len2 = (int) BLOCK_SIZE / 2 + 68;
+    try (FSDataOutputStream out = fs.append(p,
+        EnumSet.of(CreateFlag.APPEND, CreateFlag.NEW_BLOCK), 4096, null)) {
+      AppendTestUtil.write(out, len1, len2);
+    }
+
+    // Reopen file to append with NEW_BLOCK flag again, this will make middle block
+    // in file /TC8/foo not full.
+    final int len3 = (int) BLOCK_SIZE / 2;
+    try (FSDataOutputStream out = fs.append(p,
+        EnumSet.of(CreateFlag.APPEND, CreateFlag.NEW_BLOCK), 4096, null)) {
+      AppendTestUtil.write(out, len1 + len2, len3);
+    }
+
+    // Not use NEW_BLOCK flag.
+    final int len4 = 600;
+    try(FSDataOutputStream out = fs.append(p,
+        EnumSet.of(CreateFlag.APPEND), 4096, null)) {
+      AppendTestUtil.write(out, len1 + len2 + len3, len4);
+    }
+
+    AppendTestUtil.check(fs, p, len1 + len2 + len3 + len4);
+  }
 }