[iotdb] branch ml_test_1_async updated: Revert "change offer to put do disable wal"

[xingtanzjr]

This is an automated email from the ASF dual-hosted git repository.

xingtanzjr pushed a commit to branch ml_test_1_async
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/ml_test_1_async by this push:
     new 14a3463b55 Revert "change offer to put do disable wal"
14a3463b55 is described below

commit 14a3463b55f98094af2f743501a094c950aa5086
Author: Jinrui.Zhang <xingtan...@gmail.com>
AuthorDate: Thu Aug 4 16:32:22 2022 +0800

    Revert "change offer to put do disable wal"
    
    This reverts commit b3413871ee2c6eac4bd0ab297e166961a3d18e1e.
---
 .../multileader/logdispatcher/LogDispatcher.java   | 40 ++++++++--------------
 1 file changed, 14 insertions(+), 26 deletions(-)

diff --git 
a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java
 
b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java
index faffa7bf60..da5ad41aaa 100644
--- 
a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java
+++ 
b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java
@@ -20,7 +20,6 @@
 package org.apache.iotdb.consensus.multileader.logdispatcher;
 
 import org.apache.iotdb.common.rpc.thrift.TEndPoint;
-import org.apache.iotdb.commons.StepTracker;
 import org.apache.iotdb.commons.client.IClientManager;
 import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
 import org.apache.iotdb.consensus.common.Peer;
@@ -115,32 +114,21 @@ public class LogDispatcher {
               "{}: Push a log to the queue, where the queue length is {}",
               impl.getThisNode().getGroupId(),
               thread.getPendingRequest().size());
-          long startTime = System.nanoTime();
-          try {
-            thread
-                .getPendingRequest()
-                .put(new IndexedConsensusRequest(serializedRequest, 
request.getSearchIndex()));
-          } catch (InterruptedException e) {
-            e.printStackTrace();
-          } finally {
-            StepTracker.trace("putToQueue", startTime, System.nanoTime());
+          if (!thread
+              .getPendingRequest()
+              .offer(new IndexedConsensusRequest(serializedRequest, 
request.getSearchIndex()))) {
+            logger.info(
+                "{}: Log queue to {} is full. skip current request: {}",
+                impl.getThisNode().getGroupId(),
+                thread.getPeer().getEndpoint().getIp(),
+                request.getSearchIndex());
+            logger.debug(
+                "{}: Log queue of {} is full, ignore the log to this node",
+                impl.getThisNode().getGroupId(),
+                thread.getPeer());
+          } else {
+            thread.countQueueUsage(request.getSearchIndex());
           }
-          //          if (!thread
-          //              .getPendingRequest()
-          //              .put(new IndexedConsensusRequest(serializedRequest,
-          // request.getSearchIndex()))) {
-          //            logger.info(
-          //                "{}: Log queue to {} is full. skip current 
request: {}",
-          //                impl.getThisNode().getGroupId(),
-          //                thread.getPeer().getEndpoint().getIp(),
-          //                request.getSearchIndex());
-          //            logger.debug(
-          //                "{}: Log queue of {} is full, ignore the log to 
this node",
-          //                impl.getThisNode().getGroupId(),
-          //                thread.getPeer());
-          //          } else {
-          //            thread.countQueueUsage(request.getSearchIndex());
-          //          }
         });
   }