AbstractRuleEngineLifecycleIntegrationTest added await getLagTotal == 0

application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java

@@ -42,11 +42,14 @@ import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg;
 import org.thingsboard.server.common.msg.queue.TbMsgCallback;
 import org.thingsboard.server.controller.AbstractRuleEngineControllerTest;
 import org.thingsboard.server.dao.attributes.AttributesService;
+import org.thingsboard.server.queue.memory.InMemoryStorage;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
+import static org.awaitility.Awaitility.await;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
 /**
@@ -132,6 +135,8 @@ public abstract class AbstractRuleEngineLifecycleIntegrationTest extends Abstrac
         attributesService.save(device.getTenantId(), device.getId(), DataConstants.SERVER_SCOPE,
                 Collections.singletonList(new BaseAttributeKvEntry(new StringDataEntry("serverAttributeKey", "serverAttributeValue"), System.currentTimeMillis())));
 
+        await("total inMemory queue lag is empty").atMost(30, TimeUnit.SECONDS)
+                .until(() -> InMemoryStorage.getInstance().getLagTotal() == 0);
         Thread.sleep(1000);
 
         TbMsgCallback tbMsgCallback = Mockito.mock(TbMsgCallback.class);
@@ -139,7 +144,7 @@ public abstract class AbstractRuleEngineLifecycleIntegrationTest extends Abstrac
         QueueToRuleEngineMsg qMsg = new QueueToRuleEngineMsg(savedTenant.getId(), tbMsg, null, null);
         // Pushing Message to the system
         actorSystem.tell(qMsg);
-        Mockito.verify(tbMsgCallback, Mockito.timeout(3000)).onSuccess();
+        Mockito.verify(tbMsgCallback, Mockito.timeout(10000)).onSuccess();
 
 
         PageData<Event> eventsPage = getDebugEvents(savedTenant.getId(), ruleChain.getFirstRuleNodeId(), 1000);