fix 4M in writeWithoutMmap

Change-Id: I6ea680fd0db6657af8442dfc78a5f40a3e588803

Author: guyinyou

store/src/main/java/org/apache/rocketmq/store/logfile/DefaultMappedFile.java

@@ -159,7 +159,7 @@ public ByteBuffer acquire() {
 
         SHARED_BYTE_BUFFER = new SharedByteBuffer[maxSharedNum];
         for (int i = 0; i < maxSharedNum; i++) {
-            SHARED_BYTE_BUFFER[i] = new SharedByteBuffer(4 * 1024 * 1024);
+            SHARED_BYTE_BUFFER[i] = new SharedByteBuffer(5 * 1024 * 1024);
         }
     }
 
@@ -336,24 +336,26 @@ public AppendMessageResult appendMessage(final ByteBuffer byteBufferMsg, final C
                 byteBuffer.position(currentPos);
             }
 
-            AppendMessageResult result = cb.doAppend(byteBuffer, fileFromOffset, this.fileSize - currentPos, byteBufferMsg);
-
-            if (sharedByteBuffer != null) {
-                try {
-                    this.fileChannel.position(currentPos);
-                    byteBuffer.position(0).limit(result.getWroteBytes());
-                    this.fileChannel.write(byteBuffer);
-                } catch (Throwable t) {
-                    log.error("Failed to write to mappedFile {}", this.fileName, t);
-                    return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
-                } finally {
+            try {
+                AppendMessageResult result = cb.doAppend(byteBuffer, fileFromOffset, this.fileSize - currentPos, byteBufferMsg);
+                if (sharedByteBuffer != null) {
+                    try {
+                        this.fileChannel.position(currentPos);
+                        byteBuffer.position(0).limit(result.getWroteBytes());
+                        this.fileChannel.write(byteBuffer);
+                    } catch (Throwable t) {
+                        log.error("Failed to write to mappedFile {}", this.fileName, t);
+                        return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
+                    }
+                }
+                WROTE_POSITION_UPDATER.addAndGet(this, result.getWroteBytes());
+                this.storeTimestamp = result.getStoreTimestamp();
+                return result;
+            } finally {
+                if (sharedByteBuffer != null) {
                     sharedByteBuffer.release();
                 }
             }
-
-            WROTE_POSITION_UPDATER.addAndGet(this, result.getWroteBytes());
-            this.storeTimestamp = result.getStoreTimestamp();
-            return result;
         }
         log.error("MappedFile.appendMessage return null, wrotePosition: {} fileSize: {}", currentPos, this.fileSize);
         return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
@@ -393,30 +395,31 @@ public AppendMessageResult appendMessagesInner(final MessageExt messageExt, fina
             }
 
             AppendMessageResult result;
-            if (messageExt instanceof MessageExtBatch && !((MessageExtBatch) messageExt).isInnerBatch()) {
-                // traditional batch message
-                result = cb.doAppend(fileFromOffset, byteBuffer, this.fileSize - currentPos,
-                    (MessageExtBatch) messageExt, putMessageContext);
-            } else if (messageExt instanceof MessageExtBrokerInner) {
-                // traditional single message or newly introduced inner-batch message
-                result = cb.doAppend(fileFromOffset, byteBuffer, this.fileSize - currentPos,
-                    (MessageExtBrokerInner) messageExt, putMessageContext);
-            } else {
-                if (sharedByteBuffer != null) {
-                    sharedByteBuffer.release();
+            try {
+                if (messageExt instanceof MessageExtBatch && !((MessageExtBatch) messageExt).isInnerBatch()) {
+                    // traditional batch message
+                    result = cb.doAppend(fileFromOffset, byteBuffer, this.fileSize - currentPos,
+                        (MessageExtBatch) messageExt, putMessageContext);
+                } else if (messageExt instanceof MessageExtBrokerInner) {
+                    // traditional single message or newly introduced inner-batch message
+                    result = cb.doAppend(fileFromOffset, byteBuffer, this.fileSize - currentPos,
+                        (MessageExtBrokerInner) messageExt, putMessageContext);
+                } else {
+                    return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
                 }
-                return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
-            }
 
-            if (sharedByteBuffer != null) {
-                try {
-                    this.fileChannel.position(currentPos);
-                    byteBuffer.position(0).limit(result.getWroteBytes());
-                    this.fileChannel.write(byteBuffer);
-                } catch (Throwable t) {
-                    log.error("Failed to write to mappedFile {}", this.fileName, t);
-                    return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
-                } finally {
+                if (sharedByteBuffer != null) {
+                    try {
+                        this.fileChannel.position(currentPos);
+                        byteBuffer.position(0).limit(result.getWroteBytes());
+                        this.fileChannel.write(byteBuffer);
+                    } catch (Throwable t) {
+                        log.error("Failed to write to mappedFile {}", this.fileName, t);
+                        return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
+                    }
+                }
+            } finally {
+                if (sharedByteBuffer != null) {
                     sharedByteBuffer.release();
                 }
             }
@@ -428,6 +431,7 @@ public AppendMessageResult appendMessagesInner(final MessageExt messageExt, fina
         log.error("MappedFile.appendMessage return null, wrotePosition: {} fileSize: {}", currentPos, this.fileSize);
         return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
     }
+
     protected ByteBuffer appendMessageBuffer() {
         this.mappedByteBufferAccessCountSinceLastSwap++;
         return writeBuffer != null ? writeBuffer : this.mappedByteBuffer;