Using Exectutor in Kafka Node. NEVER use Fork-Join pool with parallelism 1

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  * You may obtain a copy of the License at
  *
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  *
  * Unless required by applicable law or agreed to in writing, software
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * distributed under the License is distributed on an "AS IS" BASIS,

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  * You may obtain a copy of the License at
  *
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  *
  * Unless required by applicable law or agreed to in writing, software
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -21,6 +21,7 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.context.event.ApplicationReadyEvent;
 import org.springframework.boot.context.event.ApplicationReadyEvent;
 import org.springframework.context.event.EventListener;
 import org.springframework.context.event.EventListener;
 import org.springframework.stereotype.Service;
 import org.springframework.stereotype.Service;
+import org.thingsboard.common.util.ThingsBoardThreadFactory;
 import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.actors.ActorSystemContext;
 import org.thingsboard.server.actors.DefaultTbActorSystem;
 import org.thingsboard.server.actors.DefaultTbActorSystem;
 import org.thingsboard.server.actors.TbActorId;
 import org.thingsboard.server.actors.TbActorId;
@@ -83,10 +84,10 @@ public class DefaultActorService implements ActorService {
         TbActorSystemSettings settings = new TbActorSystemSettings(actorThroughput, schedulerPoolSize, maxActorInitAttempts);
         TbActorSystemSettings settings = new TbActorSystemSettings(actorThroughput, schedulerPoolSize, maxActorInitAttempts);
         system = new DefaultTbActorSystem(settings);
         system = new DefaultTbActorSystem(settings);
 
 
-        system.createDispatcher(APP_DISPATCHER_NAME, initDispatcherExecutor(appDispatcherSize));
-        system.createDispatcher(TENANT_DISPATCHER_NAME, initDispatcherExecutor(tenantDispatcherSize));
-        system.createDispatcher(DEVICE_DISPATCHER_NAME, initDispatcherExecutor(deviceDispatcherSize));
-        system.createDispatcher(RULE_DISPATCHER_NAME, initDispatcherExecutor(ruleDispatcherSize));
+        system.createDispatcher(APP_DISPATCHER_NAME, initDispatcherExecutor(APP_DISPATCHER_NAME, appDispatcherSize));
+        system.createDispatcher(TENANT_DISPATCHER_NAME, initDispatcherExecutor(TENANT_DISPATCHER_NAME, tenantDispatcherSize));
+        system.createDispatcher(DEVICE_DISPATCHER_NAME, initDispatcherExecutor(DEVICE_DISPATCHER_NAME, deviceDispatcherSize));
+        system.createDispatcher(RULE_DISPATCHER_NAME, initDispatcherExecutor(RULE_DISPATCHER_NAME, ruleDispatcherSize));
 
 
         actorContext.setActorSystem(system);
         actorContext.setActorSystem(system);
 
 
@@ -99,13 +100,13 @@ public class DefaultActorService implements ActorService {
         log.info("Actor system initialized.");
         log.info("Actor system initialized.");
     }
     }
 
 
-    private ExecutorService initDispatcherExecutor(int poolSize) {
+    private ExecutorService initDispatcherExecutor(String dispatcherName, int poolSize) {
         if (poolSize == 0) {
         if (poolSize == 0) {
             int cores = Runtime.getRuntime().availableProcessors();
             int cores = Runtime.getRuntime().availableProcessors();
             poolSize = Math.max(1, cores / 2);
             poolSize = Math.max(1, cores / 2);
         }
         }
         if (poolSize == 1) {
         if (poolSize == 1) {
-            return Executors.newFixedThreadPool(1);
+            return Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName(dispatcherName));
         } else {
         } else {
             return Executors.newWorkStealingPool(poolSize);
             return Executors.newWorkStealingPool(poolSize);
         }
         }

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  * You may obtain a copy of the License at
  *
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  *
  * Unless required by applicable law or agreed to in writing, software
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * distributed under the License is distributed on an "AS IS" BASIS,

@@ -164,7 +164,7 @@ public class DefaultTbRuleEngineRpcService implements TbRuleEngineDeviceRpcServi
     }
     }
 
 
     private void scheduleTimeout(ToDeviceRpcRequest request, UUID requestId) {
     private void scheduleTimeout(ToDeviceRpcRequest request, UUID requestId) {
-        long timeout = Math.max(0, request.getExpirationTime() - System.currentTimeMillis());
+        long timeout = Math.max(0, request.getExpirationTime() - System.currentTimeMillis()) + TimeUnit.SECONDS.toMillis(1);
         log.trace("[{}] processing the request: [{}]", this.hashCode(), requestId);
         log.trace("[{}] processing the request: [{}]", this.hashCode(), requestId);
         scheduler.schedule(() -> {
         scheduler.schedule(() -> {
             log.trace("[{}] timeout the request: [{}]", this.hashCode(), requestId);
             log.trace("[{}] timeout the request: [{}]", this.hashCode(), requestId);

@@ -26,7 +26,9 @@ import java.util.Arrays;
 
 
 @RunWith(ClasspathSuite.class)
 @RunWith(ClasspathSuite.class)
 @ClasspathSuite.ClassnameFilters({
 @ClasspathSuite.ClassnameFilters({
-        "org.thingsboard.server.mqtt.rpc.sql.*Test", "org.thingsboard.server.mqtt.telemetry.sql.*Test"})
+        "org.thingsboard.server.mqtt.rpc.sql.*Test",
+        "org.thingsboard.server.mqtt.telemetry.sql.*Test"
+})
 public class MqttSqlTestSuite {
 public class MqttSqlTestSuite {
 
 
     @ClassRule
     @ClassRule

@@ -136,7 +136,7 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractC
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"24\",\"value\": 1},\"timeout\": 6000}";
         String setGpioRequest = "{\"method\":\"setGpio\",\"params\":{\"pin\": \"24\",\"value\": 1},\"timeout\": 6000}";
         String deviceId = savedDevice.getId().getId().toString();
         String deviceId = savedDevice.getId().getId().toString();
 
 
-        doPostAsync("/api/plugins/rpc/oneway/" + deviceId, setGpioRequest, String.class, status().isRequestTimeout(),
+        doPostAsync("/api/plugins/rpc/oneway/" + deviceId, setGpioRequest, String.class, status().is(409),
                 asyncContextTimeoutToUseRpcPlugin);
                 asyncContextTimeoutToUseRpcPlugin);
     }
     }
 
 

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  * You may obtain a copy of the License at
  *
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  *
  * Unless required by applicable law or agreed to in writing, software
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * distributed under the License is distributed on an "AS IS" BASIS,

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  * You may obtain a copy of the License at
  *
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  *
  * Unless required by applicable law or agreed to in writing, software
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -54,7 +54,6 @@ public final class TbActorMailbox implements TbActorCtx {
         dispatcher.getExecutor().execute(() -> tryInit(1));
         dispatcher.getExecutor().execute(() -> tryInit(1));
     }
     }
 
 
-
     private void tryInit(int attempt) {
     private void tryInit(int attempt) {
         try {
         try {
             log.debug("[{}] Trying to init actor, attempt: {}", selfId, attempt);
             log.debug("[{}] Trying to init actor, attempt: {}", selfId, attempt);

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  * You may obtain a copy of the License at
  *
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  *
  * Unless required by applicable law or agreed to in writing, software
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * distributed under the License is distributed on an "AS IS" BASIS,

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  * You may obtain a copy of the License at
  *
  *
- * http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  *
  * Unless required by applicable law or agreed to in writing, software
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -40,23 +40,19 @@ import java.util.concurrent.atomic.AtomicLong;
 public class ActorSystemTest {
 public class ActorSystemTest {
 
 
     public static final String ROOT_DISPATCHER = "root-dispatcher";
     public static final String ROOT_DISPATCHER = "root-dispatcher";
-    private static final int _1M = 1024 * 1024;
+    private static final int _100K = 100 * 1024;
 
 
     private volatile TbActorSystem actorSystem;
     private volatile TbActorSystem actorSystem;
     private volatile ExecutorService submitPool;
     private volatile ExecutorService submitPool;
+    private int parallelism;
 
 
     @Before
     @Before
     public void initActorSystem() {
     public void initActorSystem() {
         int cores = Runtime.getRuntime().availableProcessors();
         int cores = Runtime.getRuntime().availableProcessors();
-        int parallelism = Math.max(1, cores / 2);
+        parallelism = Math.max(2, cores / 2);
         TbActorSystemSettings settings = new TbActorSystemSettings(5, parallelism, 42);
         TbActorSystemSettings settings = new TbActorSystemSettings(5, parallelism, 42);
         actorSystem = new DefaultTbActorSystem(settings);
         actorSystem = new DefaultTbActorSystem(settings);
         submitPool = Executors.newWorkStealingPool(parallelism);
         submitPool = Executors.newWorkStealingPool(parallelism);
-//        actorSystem.createDispatcher(ROOT_DISPATCHER, Executors.newCachedThreadPool());
-//        actorSystem.createDispatcher(ROOT_DISPATCHER, Executors.newFixedThreadPool(parallelism));
-        actorSystem.createDispatcher(ROOT_DISPATCHER, Executors.newWorkStealingPool(parallelism));
-//        actorSystem.createDispatcher(ROOT_DISPATCHER, Executors.newWorkStealingPool(1));
-//        actorSystem.createDispatcher(ROOT_DISPATCHER, Executors.newFixedThreadPool(1));
     }
     }
 
 
     @After
     @After
@@ -66,32 +62,44 @@ public class ActorSystemTest {
     }
     }
 
 
     @Test
     @Test
-    public void test1actorsAnd1MMessages() throws InterruptedException {
-        testActorsAndMessages(1, _1M, 5);
+    public void test1actorsAnd100KMessages() throws InterruptedException {
+        actorSystem.createDispatcher(ROOT_DISPATCHER, Executors.newWorkStealingPool(parallelism));
+        testActorsAndMessages(1, _100K, 1);
+    }
+
+    @Test
+    public void test10actorsAnd100KMessages() throws InterruptedException {
+        actorSystem.createDispatcher(ROOT_DISPATCHER, Executors.newWorkStealingPool(parallelism));
+        testActorsAndMessages(10, _100K, 1);
     }
     }
 
 
     @Test
     @Test
-    public void test10actorsAnd1MMessages() throws InterruptedException {
-        testActorsAndMessages(10, _1M, 5);
+    public void test100KActorsAnd1Messages5timesSingleThread() throws InterruptedException {
+        actorSystem.createDispatcher(ROOT_DISPATCHER, Executors.newSingleThreadExecutor());
+        testActorsAndMessages(_100K, 1, 5);
     }
     }
 
 
     @Test
     @Test
-    public void test1MActorsAnd1Messages5times() throws InterruptedException {
-        testActorsAndMessages(_1M, 1, 5);
+    public void test100KActorsAnd1Messages5times() throws InterruptedException {
+        actorSystem.createDispatcher(ROOT_DISPATCHER, Executors.newWorkStealingPool(parallelism));
+        testActorsAndMessages(_100K, 1, 5);
     }
     }
 
 
     @Test
     @Test
-    public void test1MActorsAnd10Messages() throws InterruptedException {
-        testActorsAndMessages(_1M, 10, 1);
+    public void test100KActorsAnd10Messages() throws InterruptedException {
+        actorSystem.createDispatcher(ROOT_DISPATCHER, Executors.newWorkStealingPool(parallelism));
+        testActorsAndMessages(_100K, 10, 1);
     }
     }
 
 
     @Test
     @Test
     public void test1KActorsAnd1KMessages() throws InterruptedException {
     public void test1KActorsAnd1KMessages() throws InterruptedException {
+        actorSystem.createDispatcher(ROOT_DISPATCHER, Executors.newWorkStealingPool(parallelism));
         testActorsAndMessages(1000, 1000, 10);
         testActorsAndMessages(1000, 1000, 10);
     }
     }
 
 
     @Test
     @Test
     public void testNoMessagesAfterDestroy() throws InterruptedException {
     public void testNoMessagesAfterDestroy() throws InterruptedException {
+        actorSystem.createDispatcher(ROOT_DISPATCHER, Executors.newWorkStealingPool(parallelism));
         ActorTestCtx testCtx1 = getActorTestCtx(1);
         ActorTestCtx testCtx1 = getActorTestCtx(1);
         ActorTestCtx testCtx2 = getActorTestCtx(1);
         ActorTestCtx testCtx2 = getActorTestCtx(1);
 
 
@@ -105,11 +113,12 @@ public class ActorSystemTest {
         actorSystem.stop(actorId1);
         actorSystem.stop(actorId1);
 
 
         Assert.assertTrue(testCtx2.getLatch().await(1, TimeUnit.SECONDS));
         Assert.assertTrue(testCtx2.getLatch().await(1, TimeUnit.SECONDS));
-        Assert.assertFalse(testCtx1.getLatch().await(2, TimeUnit.SECONDS));
+        Assert.assertFalse(testCtx1.getLatch().await(1, TimeUnit.SECONDS));
     }
     }
 
 
     @Test
     @Test
     public void testOneActorCreated() throws InterruptedException {
     public void testOneActorCreated() throws InterruptedException {
+        actorSystem.createDispatcher(ROOT_DISPATCHER, Executors.newWorkStealingPool(parallelism));
         ActorTestCtx testCtx1 = getActorTestCtx(1);
         ActorTestCtx testCtx1 = getActorTestCtx(1);
         ActorTestCtx testCtx2 = getActorTestCtx(1);
         ActorTestCtx testCtx2 = getActorTestCtx(1);
         TbActorId actorId = new TbEntityActorId(new DeviceId(UUID.randomUUID()));
         TbActorId actorId = new TbEntityActorId(new DeviceId(UUID.randomUUID()));
@@ -119,12 +128,13 @@ public class ActorSystemTest {
         Thread.sleep(1000);
         Thread.sleep(1000);
         actorSystem.tell(actorId, new IntTbActorMsg(42));
         actorSystem.tell(actorId, new IntTbActorMsg(42));
 
 
-        Assert.assertTrue(testCtx1.getLatch().await(3, TimeUnit.SECONDS));
-        Assert.assertFalse(testCtx2.getLatch().await(3, TimeUnit.SECONDS));
+        Assert.assertTrue(testCtx1.getLatch().await(1, TimeUnit.SECONDS));
+        Assert.assertFalse(testCtx2.getLatch().await(1, TimeUnit.SECONDS));
     }
     }
 
 
     @Test
     @Test
     public void testActorCreatorCalledOnce() throws InterruptedException {
     public void testActorCreatorCalledOnce() throws InterruptedException {
+        actorSystem.createDispatcher(ROOT_DISPATCHER, Executors.newWorkStealingPool(parallelism));
         ActorTestCtx testCtx = getActorTestCtx(1);
         ActorTestCtx testCtx = getActorTestCtx(1);
         TbActorId actorId = new TbEntityActorId(new DeviceId(UUID.randomUUID()));
         TbActorId actorId = new TbEntityActorId(new DeviceId(UUID.randomUUID()));
         for (int i = 0; i < 1000; i++) {
         for (int i = 0; i < 1000; i++) {
@@ -170,7 +180,7 @@ public class ActorSystemTest {
             testCtxes.forEach(ctx -> {
             testCtxes.forEach(ctx -> {
                 try {
                 try {
                     boolean success = ctx.getLatch().await(1, TimeUnit.MINUTES);
                     boolean success = ctx.getLatch().await(1, TimeUnit.MINUTES);
-                    if(!success){
+                    if (!success) {
                         log.warn("Failed: {}, {}", ctx.getActual().get(), ctx.getInvocationCount().get());
                         log.warn("Failed: {}, {}", ctx.getActual().get(), ctx.getInvocationCount().get());
                     }
                     }
                     Assert.assertTrue(success);
                     Assert.assertTrue(success);

@@ -96,7 +96,19 @@ public class TbKafkaNode implements TbNode {
     public void onMsg(TbContext ctx, TbMsg msg) {
     public void onMsg(TbContext ctx, TbMsg msg) {
         String topic = TbNodeUtils.processPattern(config.getTopicPattern(), msg.getMetaData());
         String topic = TbNodeUtils.processPattern(config.getTopicPattern(), msg.getMetaData());
         try {
         try {
+            ctx.getExternalCallExecutor().executeAsync(() -> {
+                publish(ctx, msg, topic);
+                return null;
+            });
+        } catch (Exception e) {
+            ctx.tellFailure(msg, e);
+        }
+    }
+
+    protected void publish(TbContext ctx, TbMsg msg, String topic) {
+        try {
             if (!addMetadataKeyValuesAsKafkaHeaders) {
             if (!addMetadataKeyValuesAsKafkaHeaders) {
+                //TODO: external system executor
                 producer.send(new ProducerRecord<>(topic, msg.getData()),
                 producer.send(new ProducerRecord<>(topic, msg.getData()),
                         (metadata, e) -> processRecord(ctx, msg, metadata, e));
                         (metadata, e) -> processRecord(ctx, msg, metadata, e));
             } else {
             } else {
@@ -105,9 +117,8 @@ public class TbKafkaNode implements TbNode {
                 producer.send(new ProducerRecord<>(topic, null, null, null, msg.getData(), headers),
                 producer.send(new ProducerRecord<>(topic, null, null, null, msg.getData(), headers),
                         (metadata, e) -> processRecord(ctx, msg, metadata, e));
                         (metadata, e) -> processRecord(ctx, msg, metadata, e));
             }
             }
-
         } catch (Exception e) {
         } catch (Exception e) {
-            ctx.tellFailure(msg, e);
+            log.debug("[{}] Failed to process message: {}", ctx.getSelfId(), msg, e);
         }
         }
     }
     }