From 923e41c0291a0574919be69dc9368c403aacdc20 Mon Sep 17 00:00:00 2001 From: Sergey Matvienko Date: Mon, 28 Aug 2023 16:56:59 +0200 Subject: [PATCH] MathNode: tellFailure and process the next message in the queue in case we are the last consumer --- .../rule/engine/math/TbMathNode.java | 15 ++++-- .../rule/engine/math/TbMathNodeTest.java | 54 +++++++++++++++++++ 2 files changed, 65 insertions(+), 4 deletions(-) diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java index 5a0c717445..a2b809037b 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/math/TbMathNode.java @@ -154,10 +154,17 @@ public class TbMathNode implements TbNode { tryProcessQueue(lockAndQueue); } }, ctx.getDbCallbackExecutor()); - } catch (Throwable e) { + } catch (Throwable t) { semaphore.release(); - log.warn("[{}] Failed to process message: {}", lockAndQueue.getEntityId(), tbMsgTbContext == null ? null : tbMsgTbContext.getMsg(), e); - throw e; + if (tbMsgTbContext == null) { // if no message polled, the loop become infinite, will throw exception + log.error("[{}] Failed to process TbMsgTbContext queue", lockAndQueue.getEntityId(), t); + throw t; + } + TbMsg msg = tbMsgTbContext.getMsg(); + TbContext ctx = tbMsgTbContext.getCtx(); + log.warn("[{}] Failed to process message: {}", lockAndQueue.getEntityId(), msg, t); + ctx.tellFailure(msg, t); // you are not allowed to throw here, because queue will remain unprocessed + continue; // We are probably the last who process the queue. We have to continue poll until get successful callback or queue is empty } break; //submitted async exact one task. next poll will try on callback } @@ -364,7 +371,7 @@ public class TbMathNode implements TbNode { return function.apply(arg1.getValue(), arg2.getValue()); } - private ListenableFuture resolveArguments(TbContext ctx, TbMsg msg, Optional msgBodyOpt, TbMathArgument arg) { + ListenableFuture resolveArguments(TbContext ctx, TbMsg msg, Optional msgBodyOpt, TbMathArgument arg) { switch (arg.getType()) { case CONSTANT: return Futures.immediateFuture(TbMathArgumentValue.constant(arg)); diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java index 50424a1521..4a5ec8b209 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/math/TbMathNodeTest.java @@ -64,6 +64,7 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.BDDMockito.willThrow; import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; @@ -577,6 +578,59 @@ public class TbMathNodeTest { verify(ctx, never()).tellFailure(any(), any()); } + @Test + public void testExp4j_concurrentBySingleOriginator_processMsgAsyncException() { + TbMathNode node = spy(initNodeWithCustomFunction("2a+3b", + new TbMathResult(TbMathArgumentType.MESSAGE_BODY, "result", 2, false, false, null), + new TbMathArgument(TbMathArgumentType.MESSAGE_BODY, "a"), + new TbMathArgument(TbMathArgumentType.MESSAGE_BODY, "b") + )); + + willThrow(new RuntimeException("Message body has no 'delta'")).given(node).resolveArguments(any(), any(), any(), any()); + + EntityId originatorSlow = DeviceId.fromString("7f01170d-6bba-419c-b95c-2b4c3ba32f30"); + CountDownLatch slowProcessingLatch = new CountDownLatch(1); + + List slowMsgList = IntStream.range(0, 5) + .mapToObj(x -> TbMsg.newMsg("TEST", originatorSlow, new TbMsgMetaData(), JacksonUtil.newObjectNode().put("a", 2).put("b", 2).toString())) + .collect(Collectors.toList()); + + assertThat(slowMsgList.size()).as("slow msgs >= rule-dispatcher pool size").isGreaterThanOrEqualTo(RULE_DISPATCHER_POOL_SIZE); + + log.debug("rule-dispatcher [{}], db-callback [{}], slowMsg [{}]", RULE_DISPATCHER_POOL_SIZE, DB_CALLBACK_POOL_SIZE, slowMsgList.size()); + + willAnswer(invocation -> { + TbMsg msg = invocation.getArgument(1); + if (slowProcessingLatch.getCount() > 0) { + log.debug("Await on slowProcessingLatch before processMsgAsync"); + try { + assertThat(slowProcessingLatch.await(30, TimeUnit.SECONDS)).as("await on slowProcessingLatch").isTrue(); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + log.debug("\uD83D\uDC0C processMsgAsync with exception [{}][{}]", msg.getOriginator(), msg); + return invocation.callRealMethod(); + }).given(node).processMsgAsync(eq(ctx), argThat(slowMsgList::contains)); + + willAnswer(invocation -> { + TbMsg msg = invocation.getArgument(1); + log.debug("submit slow originator onMsg [{}][{}]", msg.getOriginator(), msg); + return invocation.callRealMethod(); + }).given(node).onMsg(eq(ctx), argThat(slowMsgList::contains)); + + // submit slow msg may block all rule engine dispatcher threads + slowMsgList.forEach(msg -> ruleEngineDispatcherExecutor.executeAsync(() -> node.onMsg(ctx, msg))); + // wait until dispatcher threads started with all slowMsg + verify(node, new Timeout(TimeUnit.SECONDS.toMillis(5), times(slowMsgList.size()))).onMsg(eq(ctx), argThat(slowMsgList::contains)); + + slowProcessingLatch.countDown(); + + verify(ctx, new Timeout(TimeUnit.SECONDS.toMillis(5), times(slowMsgList.size()))).tellFailure(any(), any()); + verify(ctx, never()).tellSuccess(any()); + + } + static class RuleDispatcherExecutor extends AbstractListeningExecutor { @Override protected int getThreadPollSize() {