MASTER [ROCKETMQ-14] invoke callback shoule be invoked in an executor rather...

MASTER [ROCKETMQ-14] invoke callback shoule be invoked in an executor rather than in current thread, closes apache/incubator-rocketmq#2

rocketmq-remoting/src/main/java/com/alibaba/rocketmq/remoting/netty/NettyRemotingAbstract.java

@@ -198,6 +198,18 @@ public abstract class NettyRemotingAbstract {
             responseTable.remove(opaque);
 
             if (responseFuture.getInvokeCallback() != null) {
+                executeInvokeCallback(responseFuture);
+            } else {
+                responseFuture.putResponse(cmd);
+            }
+        } else {
+            PLOG.warn("receive response, but not matched any request, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
+            PLOG.warn(cmd.toString());
+        }
+    }
+
+    //execute callback in callback executor. If callback executor is null, run directly in current thread
+    private void executeInvokeCallback(final ResponseFuture responseFuture) {
         boolean runInThisThread = false;
         ExecutorService executor = this.getCallbackExecutor();
         if (executor != null) {
@@ -227,13 +239,6 @@ public abstract class NettyRemotingAbstract {
                 PLOG.warn("executeInvokeCallback Exception", e);
             }
         }
-            } else {
-                responseFuture.putResponse(cmd);
-            }
-        } else {
-            PLOG.warn("receive response, but not matched any request, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
-            PLOG.warn(cmd.toString());
-        }
     }
 
     public abstract RPCHook getRPCHook();
@@ -257,7 +262,7 @@ public abstract class NettyRemotingAbstract {
 
         for (ResponseFuture rf : rfList) {
             try {
-                rf.executeInvokeCallback();
+                executeInvokeCallback(rf);
             } catch (Throwable e) {
                 PLOG.warn("scanResponseTable, operationComplete Exception", e);
             }
@@ -329,7 +334,7 @@ public abstract class NettyRemotingAbstract {
                         responseFuture.putResponse(null);
                         responseTable.remove(opaque);
                         try {
-                            responseFuture.executeInvokeCallback();
+                            executeInvokeCallback(responseFuture);
                         } catch (Throwable e) {
                             PLOG.warn("excute callback in writeAndFlush addListener, and callback throw", e);
                         } finally {

rocketmq-remoting/src/test/java/com/alibaba/rocketmq/remoting/NettyConnectionTest.java

@@ -22,9 +22,15 @@ import com.alibaba.rocketmq.remoting.exception.RemotingSendRequestException;
 import com.alibaba.rocketmq.remoting.exception.RemotingTimeoutException;
 import com.alibaba.rocketmq.remoting.netty.NettyClientConfig;
 import com.alibaba.rocketmq.remoting.netty.NettyRemotingClient;
+import com.alibaba.rocketmq.remoting.netty.ResponseFuture;
 import com.alibaba.rocketmq.remoting.protocol.RemotingCommand;
+import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
 
 /**
 
@@ -51,6 +57,52 @@ public class NettyConnectionTest {
         System.out.println("-----------------------------------------------------------------");
     }
 
+
+    @Test
+    public void test_async_timeout() throws InterruptedException, RemotingConnectException,
+            RemotingSendRequestException, RemotingTimeoutException {
+        RemotingClient client = createRemotingClient();
+        final AtomicInteger ai = new AtomicInteger(0);
+        final CountDownLatch latch = new CountDownLatch(100);
+        for(int i=0;i<100;i++) {
+            try {
+                RemotingCommand request = RemotingCommand.createRequestCommand(0, null);
+                client.invokeAsync("localhost:8888", request, 5, new InvokeCallback() {//very easy to timeout
+                    @Override
+                    public void operationComplete(ResponseFuture responseFuture) {
+                        if (responseFuture.isTimeout()) {
+                            if(ai.getAndIncrement()==4) {
+                                try {
+                                    System.out.println("First try timeout,  blocking 10s" + Thread.currentThread().getName());
+                                    Thread.sleep(10 * 1000);
+                                } catch (InterruptedException e) {
+                                    e.printStackTrace();
+                                }
+                            }
+                            else{
+                                System.out.println("Timeout callback execute,very short."+Thread.currentThread().getName());
+                            }
+                        }
+                        else{
+                            System.out.println("Success."+Thread.currentThread().getName());
+                        }
+                        latch.countDown();
+
+                    }
+                });
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+        }
+
+
+
+        latch.await(1000, TimeUnit.MILLISECONDS);
+        Assert.assertEquals(1, latch.getCount());//only one should be blocked
+        client.shutdown();
+        System.out.println("-----------------------------------------------------------------");
+    }
+
     public static RemotingClient createRemotingClient() {
         NettyClientConfig config = new NettyClientConfig();
         config.setClientChannelMaxIdleTimeSeconds(15);