Repository: alluxio/alluxio
Issue: workerFuse stressbench write Input/output error

**Alluxio Version:**
Current master branch

**Describe the bug**
Fuse stress bench with workerFuse writing data errors out. 
Not tuning MaxDirectMemorySize, all threads fail. Logs from `worker.out`:
```
io.netty.util.internal.OutOfDirectMemoryError: failed to allocate 16777216 byte(s) of direct memory (used: 4278190359, max: 4294967296)
        at io.netty.util.internal.PlatformDependent.incrementMemoryCounter(PlatformDependent.java:754)
        at io.netty.util.internal.PlatformDependent.allocateDirectNoCleaner(PlatformDependent.java:709)
        at io.netty.buffer.PoolArena$DirectArena.allocateDirect(PoolArena.java:645)
        at io.netty.buffer.PoolArena$DirectArena.newChunk(PoolArena.java:621)
        at io.netty.buffer.PoolArena.allocateNormal(PoolArena.java:204)
        at io.netty.buffer.PoolArena.tcacheAllocateNormal(PoolArena.java:188)
        at io.netty.buffer.PoolArena.allocate(PoolArena.java:138)
        at io.netty.buffer.PoolArena.allocate(PoolArena.java:128)
        at io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:378)
        at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:187)
        at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:178)
        at io.netty.buffer.AbstractByteBufAllocator.buffer(AbstractByteBufAllocator.java:115)
        at alluxio.client.block.stream.BlockOutStream.allocateBuffer(BlockOutStream.java:294)
        at alluxio.client.block.stream.BlockOutStream.updateCurrentChunk(BlockOutStream.java:252)
        at alluxio.client.block.stream.BlockOutStream.write(BlockOutStream.java:141)
        at alluxio.client.file.AlluxioFileOutStream.writeInternal(AlluxioFileOutStream.java:271)
        at alluxio.client.file.AlluxioFileOutStream.write(AlluxioFileOutStream.java:229)
        at alluxio.fuse.AlluxioJniFuseFileSystem.writeInternal(AlluxioJniFuseFileSystem.java:438)
        at alluxio.fuse.AlluxioJniFuseFileSystem.lambda$write$6(AlluxioJniFuseFileSystem.java:413)
        at alluxio.fuse.AlluxioFuseUtils.call(AlluxioFuseUtils.java:278)
        at alluxio.fuse.AlluxioJniFuseFileSystem.write(AlluxioJniFuseFileSystem.java:413)
        at alluxio.jnifuse.AbstractFuseFileSystem.writeCallback(AbstractFuseFileSystem.java:316)
```
With following java options in `alluxio-env.sh`: `ALLUXIO_WORKER_JAVA_OPTS+=" -Xmx24G -Xms24G -XX:MaxDirectMemorySize=18g" `, only some threads report failure, and no logs in `worker.out`.

**To Reproduce**
Using ec2 instance r5.2xlarge which has 64GB memory.
Using the following properties in `alluxio-site.properties`:
```
alluxio.master.hostname=localhost
alluxio.master.mount.table.root.ufs=s3://xxxxx
aws.accessKeyId=xxxxxxxxx
aws.secretKey=xxxxxxxxxxxxxxx

alluxio.worker.ramdisk.size=15GB
alluxio.worker.fuse.enabled=true
alluxio.worker.fuse.mount.alluxio.path=/fuseTest
alluxio.worker.fuse.mount.point=/home/centos/fuseTest

# User properties
alluxio.user.file.writetype.default=MUST_CACHE
alluxio.user.block.size.bytes.default=200k
alluxio.client.cache.async.write.enabled=false
alluxio.user.file.passive.cache.enabled=false
```
Run Fuse stress bench with: 
```
bin/alluxio runClass alluxio.stress.cli.fuse.FuseIOBench --local-path /home/centos/fuseTest --operation Write --threads 32 --num-dirs 100 --num-files-per-dir 1000 --file-size 100k
```

**Additional information**
Standalone Fuse has no such issue.

Changed files:
core/client/fs/src/main/java/alluxio/client/block/stream/BlockWorkerDataWriter.java

diff --git a/core/client/fs/src/main/java/alluxio/client/block/stream/BlockWorkerDataWriter.java b/core/client/fs/src/main/java/alluxio/client/block/stream/BlockWorkerDataWriter.java
index 0076f9b427..ffe84c9b91 100644
--- a/core/client/fs/src/main/java/alluxio/client/block/stream/BlockWorkerDataWriter.java
+++ b/core/client/fs/src/main/java/alluxio/client/block/stream/BlockWorkerDataWriter.java
@@ -92,18 +92,23 @@ public final class BlockWorkerDataWriter implements DataWriter {
 
   @Override
   public void writeChunk(final ByteBuf buf) throws IOException {
-    if (mReservedBytes < pos() + buf.readableBytes()) {
-      try {
-        long bytesToReserve = Math.max(mBufferSize, pos() + buf.readableBytes() - mReservedBytes);
-        // Allocate enough space in the existing temporary block for the write.
-        mBlockWorker.requestSpace(mSessionId, mBlockId, bytesToReserve);
-      } catch (Exception e) {
-        throw new IOException(e);
+    try {
+      if (mReservedBytes < pos() + buf.readableBytes()) {
+        try {
+          long bytesToReserve = Math.max(mBufferSize, pos() + buf.readableBytes()
+              - mReservedBytes);
+          // Allocate enough space in the existing temporary block for the write.
+          mBlockWorker.requestSpace(mSessionId, mBlockId, bytesToReserve);
+        } catch (Exception e) {
+          throw new IOException(e);
+        }
       }
+      long append = mBlockWriter.append(buf);
+      MetricsSystem.counter(MetricKey.WORKER_BYTES_WRITTEN_DIRECT.getName()).inc(append);
+      MetricsSystem.meter(MetricKey.WORKER_BYTES_WRITTEN_DIRECT_THROUGHPUT.getName()).mark(append);   
+    } finally {
+      buf.release();
     }
-    long append = mBlockWriter.append(buf);
-    MetricsSystem.counter(MetricKey.WORKER_BYTES_WRITTEN_DIRECT.getName()).inc(append);
-    MetricsSystem.meter(MetricKey.WORKER_BYTES_WRITTEN_DIRECT_THROUGHPUT.getName()).mark(append);
   }
 
   @Override