[IOTDB-1583] Raft log failed to be committed in cluster version

cluster/src/main/java/org/apache/iotdb/cluster/server/member/RaftMember.java

@@ -1032,6 +1032,15 @@ TSStatus processPlanLocally(PhysicalPlan plan) {
       log.setCurrLogTerm(getTerm().get());
       log.setCurrLogIndex(logManager.getLastLogIndex() + 1);
 
+      // if a single log exceeds the threshold
+      // we need to return error code to the client as in server mode
+      if (log.serialize().capacity() + Integer.BYTES
+          >= ClusterDescriptor.getInstance().getConfig().getRaftLogBufferSize()) {
+        logger.error(
+            "Log cannot fit into buffer, please increase raft_log_buffer_size;"
+                + "or reduce the size of requests you send.");
+        return StatusUtils.INTERNAL_ERROR;
+      }
       logManager.append(log);
     }
     Timer.Statistic.RAFT_SENDER_APPEND_LOG.calOperationCostTimeFromStart(startTime);
@@ -1077,6 +1086,14 @@ private TSStatus processPlanLocallyV2(PhysicalPlan plan) {
       log.setCurrLogIndex(logManager.getLastLogIndex() + 1);
 
       startTime = Timer.Statistic.RAFT_SENDER_APPEND_LOG_V2.getOperationStartTime();
+      // just like processPlanLocally,we need to check the size of log
+      if (log.serialize().capacity() + Integer.BYTES
+          >= ClusterDescriptor.getInstance().getConfig().getRaftLogBufferSize()) {
+        logger.error(
+            "Log cannot fit into buffer, please increase raft_log_buffer_size;"
+                + "or reduce the size of requests you send.");
+        return StatusUtils.INTERNAL_ERROR;
+      }
       // logDispatcher will serialize log, and set log size, and we will use the size after it
       logManager.append(log);
       Timer.Statistic.RAFT_SENDER_APPEND_LOG_V2.calOperationCostTimeFromStart(startTime);

server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java

@@ -113,6 +113,9 @@ public void write(PhysicalPlan plan) throws IOException {
         sync();
       }
     } catch (BufferOverflowException e) {
+      // if the size of a single plan bigger than logBufferWorking
+      // we need to clear the buffer to drop something wrong that has written.
+      logBufferWorking.clear();
       throw new IOException("Log cannot fit into the buffer, please increase wal_buffer_size", e);
     } finally {
       lock.unlock();

server/src/test/java/org/apache/iotdb/db/writelog/WriteLogNodeTest.java

@@ -316,6 +316,32 @@ public void testOverSizedWAL() throws IOException, IllegalPathException {
     }
     assertTrue(caught);
 
+    // if last insertplan failed by overflow,it can not take affect the next insertplan
+    InsertRowPlan bwInsertPlan2 =
+        new InsertRowPlan(
+            new PartialPath("root.logTestDevice.oversize"),
+            100,
+            new String[] {"s1", "s2", "s3", "s4"},
+            new TSDataType[] {
+              TSDataType.DOUBLE, TSDataType.INT64, TSDataType.TEXT, TSDataType.BOOLEAN
+            },
+            // try to apply a insertplan whose size will fill the entire logBufferWorking
+            new String[] {
+              "1.0",
+              "15",
+              new String(
+                  new char
+                      [(IoTDBDescriptor.getInstance().getConfig().getWalBufferSize() / 2) - 109]),
+              "false"
+            });
+    caught = false;
+    try {
+      logNode.write(bwInsertPlan2);
+    } catch (IOException e) {
+      caught = true;
+    }
+    assertFalse(caught);
+
     ByteBuffer[] array = logNode.delete();
     for (ByteBuffer byteBuffer : array) {
       MmapUtil.clean((MappedByteBuffer) byteBuffer);