diff --git a/application/src/main/java/org/thingsboard/server/service/script/AbstractJsInvokeService.java b/application/src/main/java/org/thingsboard/server/service/script/AbstractJsInvokeService.java index 2f0378ae9d..67ad4a8696 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/AbstractJsInvokeService.java +++ b/application/src/main/java/org/thingsboard/server/service/script/AbstractJsInvokeService.java @@ -30,6 +30,7 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; /** @@ -84,8 +85,10 @@ public abstract class AbstractJsInvokeService implements JsInvokeService { apiUsageClient.report(tenantId, customerId, ApiUsageRecordKey.JS_EXEC_COUNT, 1); return doInvokeFunction(scriptId, functionName, args); } else { - return Futures.immediateFailedFuture( - new RuntimeException("Script invocation is blocked due to maximum error count " + getMaxErrors() + "!")); + String message = "Script invocation is blocked due to maximum error count " + + getMaxErrors() + ", scriptId " + scriptId + "!"; + log.warn(message); + return Futures.immediateFailedFuture(new RuntimeException(message)); } } else { return Futures.immediateFailedFuture(new RuntimeException("JS Execution is disabled due to API limits!")); @@ -117,8 +120,11 @@ public abstract class AbstractJsInvokeService implements JsInvokeService { protected abstract long getMaxBlacklistDuration(); - protected void onScriptExecutionError(UUID scriptId) { - disabledFunctions.computeIfAbsent(scriptId, key -> new DisableListInfo()).incrementAndGet(); + protected void onScriptExecutionError(UUID scriptId, Throwable t, String scriptBody) { + DisableListInfo disableListInfo = disabledFunctions.computeIfAbsent(scriptId, key -> new DisableListInfo()); + log.warn("Script has exception and will increment counter {} on disabledFunctions for id {}, exception {}, cause {}, scriptBody {}", + disableListInfo.get(), scriptId, t, t.getCause(), scriptBody); + disableListInfo.incrementAndGet(); } private String generateJsScript(JsScriptType scriptType, String functionName, String scriptBody, String... argNames) { diff --git a/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsInvokeService.java b/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsInvokeService.java index 9985ac60a2..15a3cf1c15 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsInvokeService.java +++ b/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsInvokeService.java @@ -160,7 +160,7 @@ public abstract class AbstractNashornJsInvokeService extends AbstractJsInvokeSer return ((Invocable) engine).invokeFunction(functionName, args); } } catch (Exception e) { - onScriptExecutionError(scriptId); + onScriptExecutionError(scriptId, e, functionName); throw new ExecutionException(e); } }); diff --git a/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java b/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java index 334a471973..5b3b1f5e23 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java +++ b/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java @@ -18,7 +18,6 @@ package org.thingsboard.server.service.script; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.MoreExecutors; import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; @@ -161,7 +160,8 @@ public class RemoteJsInvokeService extends AbstractJsInvokeService { @Override protected ListenableFuture doInvokeFunction(UUID scriptId, String functionName, Object[] args) { - String scriptBody = scriptIdToBodysMap.get(scriptId); + log.trace("doInvokeFunction js-request for uuid {} with timeout {}ms", scriptId, maxRequestsTimeout); + final String scriptBody = scriptIdToBodysMap.get(scriptId); if (scriptBody == null) { return Futures.immediateFailedFuture(new RuntimeException("No script body found for scriptId: [" + scriptId + "]!")); } @@ -170,7 +170,7 @@ public class RemoteJsInvokeService extends AbstractJsInvokeService { .setScriptIdLSB(scriptId.getLeastSignificantBits()) .setFunctionName(functionName) .setTimeout((int) maxRequestsTimeout) - .setScriptBody(scriptIdToBodysMap.get(scriptId)); + .setScriptBody(scriptBody); for (Object arg : args) { jsRequestBuilder.addArgs(arg.toString()); @@ -193,7 +193,7 @@ public class RemoteJsInvokeService extends AbstractJsInvokeService { @Override public void onFailure(Throwable t) { - onScriptExecutionError(scriptId); + onScriptExecutionError(scriptId, t, scriptBody); if (t instanceof TimeoutException || (t.getCause() != null && t.getCause() instanceof TimeoutException)) { queueTimeoutMsgs.incrementAndGet(); } @@ -205,9 +205,10 @@ public class RemoteJsInvokeService extends AbstractJsInvokeService { if (invokeResult.getSuccess()) { return invokeResult.getResult(); } else { - onScriptExecutionError(scriptId); + final RuntimeException e = new RuntimeException(invokeResult.getErrorDetails()); + onScriptExecutionError(scriptId, e, scriptBody); log.debug("[{}] Failed to compile script due to [{}]: {}", scriptId, invokeResult.getErrorCode().name(), invokeResult.getErrorDetails()); - throw new RuntimeException(invokeResult.getErrorDetails()); + throw e; } }, callbackExecutor); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/TbQueueRequestTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/TbQueueRequestTemplate.java index 5dc89a9c26..192f8e1675 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/TbQueueRequestTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/TbQueueRequestTemplate.java @@ -24,6 +24,8 @@ public interface TbQueueRequestTemplate send(Request request); + ListenableFuture send(Request request, long timeoutNs); + void stop(); void setMessagesStats(MessagesStats messagesStats); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplate.java index b171f2d8d8..7463c11df5 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplate.java @@ -19,7 +19,9 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.SettableFuture; import lombok.Builder; +import lombok.Getter; import lombok.extern.slf4j.Slf4j; +import org.thingsboard.common.util.TbStopWatch; import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.queue.TbQueueAdmin; @@ -31,13 +33,17 @@ import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.common.stats.MessagesStats; +import javax.annotation.Nullable; import java.util.List; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.LockSupport; +import java.util.concurrent.locks.ReentrantLock; @Slf4j public class DefaultTbQueueRequestTemplate extends AbstractTbQueueTemplate @@ -46,15 +52,15 @@ public class DefaultTbQueueRequestTemplate requestTemplate; private final TbQueueConsumer responseTemplate; - private final ConcurrentMap> pendingRequests; - private final boolean internalExecutor; - private final ExecutorService executor; - private final long maxRequestTimeout; - private final long maxPendingRequests; - private final long pollInterval; - private volatile long tickTs = 0L; - private volatile long tickSize = 0L; - private volatile boolean stopped = false; + final ConcurrentHashMap> pendingRequests = new ConcurrentHashMap<>(); + final boolean internalExecutor; + final ExecutorService executor; + final long maxRequestTimeoutNs; + final long maxPendingRequests; + final long pollInterval; + volatile boolean stopped = false; + long nextCleanupNs = 0L; + private final Lock cleanerLock = new ReentrantLock(); private MessagesStats messagesStats; @@ -65,79 +71,113 @@ public class DefaultTbQueueRequestTemplate(); - this.maxRequestTimeout = maxRequestTimeout; + this.maxRequestTimeoutNs = TimeUnit.MILLISECONDS.toNanos(maxRequestTimeout); this.maxPendingRequests = maxPendingRequests; this.pollInterval = pollInterval; - if (executor != null) { - internalExecutor = false; - this.executor = executor; - } else { - internalExecutor = true; - this.executor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-queue-request-template-" + responseTemplate.getTopic())); - } + this.internalExecutor = (executor == null); + this.executor = internalExecutor ? createExecutor() : executor; + } + + ExecutorService createExecutor() { + return Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("tb-queue-request-template-" + responseTemplate.getTopic())); } @Override public void init() { queueAdmin.createTopicIfNotExists(responseTemplate.getTopic()); - this.requestTemplate.init(); - tickTs = System.currentTimeMillis(); + requestTemplate.init(); responseTemplate.subscribe(); - executor.submit(() -> { - long nextCleanupMs = 0L; - while (!stopped) { - try { - List responses = responseTemplate.poll(pollInterval); - if (responses.size() > 0) { - log.trace("Polling responses completed, consumer records count [{}]", responses.size()); - } - responses.forEach(response -> { - byte[] requestIdHeader = response.getHeaders().get(REQUEST_ID_HEADER); - UUID requestId; - if (requestIdHeader == null) { - log.error("[{}] Missing requestId in header and body", response); - } else { - requestId = bytesToUuid(requestIdHeader); - log.trace("[{}] Response received: {}", requestId, response); - ResponseMetaData expectedResponse = pendingRequests.remove(requestId); - if (expectedResponse == null) { - log.trace("[{}] Invalid or stale request", requestId); - } else { - expectedResponse.future.set(response); - } + executor.submit(this::mainLoop); + } + + void mainLoop() { + while (!stopped) { + TbStopWatch sw = TbStopWatch.startNew(); + try { + fetchAndProcessResponses(); + } catch (Throwable e) { + long sleepNanos = TimeUnit.MILLISECONDS.toNanos(this.pollInterval) - sw.stopAndGetTotalTimeNanos(); + log.warn("Failed to obtain and process responses from queue. Going to sleep " + sleepNanos + "ns", e); + sleep(sleepNanos); + } + } + } + + void fetchAndProcessResponses() { + final long pendingRequestsCount = pendingRequests.mappingCount(); + log.trace("Starting template pool topic {}, for pendingRequests {}", responseTemplate.getTopic(), pendingRequestsCount); + List responses = doPoll(); //poll js responses + log.trace("Completed template poll topic {}, for pendingRequests [{}], received [{}] responses", responseTemplate.getTopic(), pendingRequestsCount, responses.size()); + responses.forEach(this::processResponse); //this can take a long time + responseTemplate.commit(); + tryCleanStaleRequests(); + } + + private boolean tryCleanStaleRequests() { + if (!cleanerLock.tryLock()) { + return false; + } + try { + log.trace("tryCleanStaleRequest..."); + final long currentNs = getCurrentClockNs(); + if (nextCleanupNs < currentNs) { + pendingRequests.forEach((key, value) -> { + if (value.expTime < currentNs) { + ResponseMetaData staleRequest = pendingRequests.remove(key); + if (staleRequest != null) { + setTimeoutException(key, staleRequest, currentNs); } - }); - responseTemplate.commit(); - tickTs = System.currentTimeMillis(); - tickSize = pendingRequests.size(); - if (nextCleanupMs < tickTs) { - //cleanup; - pendingRequests.forEach((key, value) -> { - if (value.expTime < tickTs) { - ResponseMetaData staleRequest = pendingRequests.remove(key); - if (staleRequest != null) { - log.trace("[{}] Request timeout detected, expTime [{}], tickTs [{}]", key, staleRequest.expTime, tickTs); - staleRequest.future.setException(new TimeoutException()); - } - } - }); - nextCleanupMs = tickTs + maxRequestTimeout; - } - } catch (Throwable e) { - log.warn("Failed to obtain responses from queue.", e); - try { - Thread.sleep(pollInterval); - } catch (InterruptedException e2) { - log.trace("Failed to wait until the server has capacity to handle new responses", e2); } - } + }); + setupNextCleanup(); } - }); + } finally { + cleanerLock.unlock(); + } + return true; + } + + void setupNextCleanup() { + nextCleanupNs = getCurrentClockNs() + maxRequestTimeoutNs; + log.trace("setupNextCleanup {}", nextCleanupNs); + } + + List doPoll() { + return responseTemplate.poll(pollInterval); + } + + void sleep(long nanos) { + LockSupport.parkNanos(nanos); + } + + void setTimeoutException(UUID key, ResponseMetaData staleRequest, long currentNs) { + if (currentNs >= staleRequest.getSubmitTime() + staleRequest.getTimeout()) { + log.warn("Request timeout detected, currentNs [{}], {}, key [{}]", currentNs, staleRequest, key); + } else { + log.error("Request timeout detected, currentNs [{}], {}, key [{}]", currentNs, staleRequest, key); + } + staleRequest.future.setException(new TimeoutException()); + } + + void processResponse(Response response) { + byte[] requestIdHeader = response.getHeaders().get(REQUEST_ID_HEADER); + UUID requestId; + if (requestIdHeader == null) { + log.error("[{}] Missing requestId in header and body", response); + } else { + requestId = bytesToUuid(requestIdHeader); + log.trace("[{}] Response received: {}", requestId, String.valueOf(response).replace("\n", " ")); //TODO remove overhead + ResponseMetaData expectedResponse = pendingRequests.remove(requestId); + if (expectedResponse == null) { + log.warn("[{}] Invalid or stale request, response: {}", requestId, String.valueOf(response).replace("\n", " ")); + } else { + expectedResponse.future.set(response); + } + } } @Override @@ -164,17 +204,48 @@ public class DefaultTbQueueRequestTemplate send(Request request) { - if (tickSize > maxPendingRequests) { + return send(request, this.maxRequestTimeoutNs); + } + + @Override + public ListenableFuture send(Request request, long requestTimeoutNs) { + if (pendingRequests.mappingCount() >= maxPendingRequests) { + log.warn("Pending request map is full [{}]! Consider to increase maxPendingRequests or increase processing performance", maxPendingRequests); return Futures.immediateFailedFuture(new RuntimeException("Pending request map is full!")); } UUID requestId = UUID.randomUUID(); request.getHeaders().put(REQUEST_ID_HEADER, uuidToBytes(requestId)); request.getHeaders().put(RESPONSE_TOPIC_HEADER, stringToBytes(responseTemplate.getTopic())); - request.getHeaders().put(REQUEST_TIME, longToBytes(System.currentTimeMillis())); + request.getHeaders().put(REQUEST_TIME, longToBytes(getCurrentTimeMs())); + long currentClockNs = getCurrentClockNs(); SettableFuture future = SettableFuture.create(); - ResponseMetaData responseMetaData = new ResponseMetaData<>(tickTs + maxRequestTimeout, future); - pendingRequests.putIfAbsent(requestId, responseMetaData); - log.trace("[{}] Sending request, key [{}], expTime [{}]", requestId, request.getKey(), responseMetaData.expTime); + ResponseMetaData responseMetaData = new ResponseMetaData<>(currentClockNs + requestTimeoutNs, future, currentClockNs, requestTimeoutNs); + log.trace("pending {}", responseMetaData); + if (pendingRequests.putIfAbsent(requestId, responseMetaData) != null) { + log.warn("Pending request already exists [{}]!", maxPendingRequests); + return Futures.immediateFailedFuture(new RuntimeException("Pending request already exists !" + requestId)); + } + sendToRequestTemplate(request, requestId, future, responseMetaData); + return future; + } + + /** + * MONOTONIC clock instead jumping wall clock. + * Wrapped into the method for the test purposes to travel through the time + * */ + long getCurrentClockNs() { + return System.nanoTime(); + } + + /** + * Wall clock to send timestamp to an external service + * */ + long getCurrentTimeMs() { + return System.currentTimeMillis(); + } + + void sendToRequestTemplate(Request request, UUID requestId, SettableFuture future, ResponseMetaData responseMetaData) { + log.trace("[{}] Sending request, key [{}], expTime [{}], request {}", requestId, request.getKey(), responseMetaData.expTime, request); if (messagesStats != null) { messagesStats.incrementTotal(); } @@ -184,7 +255,7 @@ public class DefaultTbQueueRequestTemplate { + @Getter + static class ResponseMetaData { + private final long submitTime; + private final long timeout; private final long expTime; private final SettableFuture future; - ResponseMetaData(long ts, SettableFuture future) { + ResponseMetaData(long ts, SettableFuture future, long submitTime, long timeout) { + this.submitTime = submitTime; + this.timeout = timeout; this.expTime = ts; this.future = future; } + + @Override + public String toString() { + return "ResponseMetaData{" + + "submitTime=" + submitTime + + ", calculatedExpTime=" + (submitTime + timeout) + + ", deltaMs=" + (expTime - submitTime) + + ", expTime=" + expTime + + ", future=" + future + + '}'; + } } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java index bf694e00e6..3acabd5cf3 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaConsumerTemplate.java @@ -21,6 +21,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.springframework.util.StopWatch; import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.TbQueueMsg; import org.thingsboard.server.queue.common.AbstractTbQueueConsumerTemplate; @@ -82,7 +83,16 @@ public class TbKafkaConsumerTemplate extends AbstractTbQue @Override protected List> doPoll(long durationInMillis) { + StopWatch stopWatch = new StopWatch(); + stopWatch.start(); + + log.trace("poll topic {} maxDuration {}", getTopic(), durationInMillis); + ConsumerRecords records = consumer.poll(Duration.ofMillis(durationInMillis)); + + stopWatch.stop(); + log.trace("poll topic {} took {}ms", getTopic(), stopWatch.getTotalTimeMillis()); + if (records.isEmpty()) { return Collections.emptyList(); } else { diff --git a/common/queue/src/test/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplateTest.java b/common/queue/src/test/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplateTest.java new file mode 100644 index 0000000000..9979e9ac43 --- /dev/null +++ b/common/queue/src/test/java/org/thingsboard/server/queue/common/DefaultTbQueueRequestTemplateTest.java @@ -0,0 +1,211 @@ +/** + * Copyright © 2016-2021 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.queue.common; + +import lombok.extern.slf4j.Slf4j; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; +import org.thingsboard.server.queue.TbQueueAdmin; +import org.thingsboard.server.queue.TbQueueConsumer; +import org.thingsboard.server.queue.TbQueueMsg; +import org.thingsboard.server.queue.TbQueueProducer; + +import java.util.Collections; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.lessThan; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.BDDMockito.willDoNothing; +import static org.mockito.BDDMockito.willReturn; +import static org.mockito.Mockito.RETURNS_DEEP_STUBS; +import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.mockito.hamcrest.MockitoHamcrest.longThat; + +@Slf4j +@RunWith(MockitoJUnitRunner.class) +public class DefaultTbQueueRequestTemplateTest { + + @Mock + TbQueueAdmin queueAdmin; + @Mock + TbQueueProducer requestTemplate; + @Mock + TbQueueConsumer responseTemplate; + @Mock + ExecutorService executorMock; + + ExecutorService executor; + String topic = "js-responses-tb-node-0"; + long maxRequestTimeout = 10; + long maxPendingRequests = 32; + long pollInterval = 5; + + DefaultTbQueueRequestTemplate inst; + + @Before + public void setUp() throws Exception { + willReturn(topic).given(responseTemplate).getTopic(); + inst = spy(new DefaultTbQueueRequestTemplate( + queueAdmin, requestTemplate, responseTemplate, + maxRequestTimeout, maxPendingRequests, pollInterval, executorMock)); + + } + + @After + public void tearDown() throws Exception { + if (executor != null) { + executor.shutdownNow(); + } + } + + @Test + public void givenInstance_whenVerifyInitialParameters_thenOK() { + assertThat(inst.maxPendingRequests, equalTo(maxPendingRequests)); + assertThat(inst.maxRequestTimeoutNs, equalTo(TimeUnit.MILLISECONDS.toNanos(maxRequestTimeout))); + assertThat(inst.pollInterval, equalTo(pollInterval)); + assertThat(inst.executor, is(executorMock)); + assertThat(inst.stopped, is(false)); + assertThat(inst.internalExecutor, is(false)); + } + + @Test + public void givenExternalExecutor_whenInitStop_thenOK() { + inst.init(); + assertThat(inst.nextCleanupNs, equalTo(0L)); + verify(queueAdmin, times(1)).createTopicIfNotExists(topic); + verify(requestTemplate, times(1)).init(); + verify(responseTemplate, times(1)).subscribe(); + verify(executorMock, times(1)).submit(any(Runnable.class)); + + inst.stop(); + assertThat(inst.stopped, is(true)); + verify(responseTemplate, times(1)).unsubscribe(); + verify(requestTemplate, times(1)).stop(); + verify(executorMock, never()).shutdownNow(); + } + + @Test + public void givenMainLoop_whenLoopFewTimes_thenVerifyInvocationCount() throws InterruptedException { + executor = inst.createExecutor(); + CountDownLatch latch = new CountDownLatch(5); + willDoNothing().given(inst).sleep(anyLong()); + willAnswer(invocation -> { + if (latch.getCount() == 1) { + inst.stop(); //stop the loop in natural way + } + if (latch.getCount() == 3 || latch.getCount() == 4) { + latch.countDown(); + throw new RuntimeException("test catch block"); + } + latch.countDown(); + return null; + }).given(inst).fetchAndProcessResponses(); + + executor.submit(inst::mainLoop); + latch.await(10, TimeUnit.SECONDS); + + verify(inst, times(5)).fetchAndProcessResponses(); + verify(inst, times(2)).sleep(longThat(lessThan(TimeUnit.MILLISECONDS.toNanos(inst.pollInterval)))); + } + + @Test + public void givenMessages_whenSend_thenOK() { + willDoNothing().given(inst).sendToRequestTemplate(any(), any(), any(), any()); + inst.init(); + final int msgCount = 10; + for (int i = 0; i < msgCount; i++) { + inst.send(getRequestMsgMock()); + } + assertThat(inst.pendingRequests.mappingCount(), equalTo((long) msgCount)); + verify(inst, times(msgCount)).sendToRequestTemplate(any(), any(), any(), any()); + } + + @Test + public void givenMessagesOverMaxPendingRequests_whenSend_thenImmediateFailedFutureForTheOfRequests() { + willDoNothing().given(inst).sendToRequestTemplate(any(), any(), any(), any()); + inst.init(); + int msgOverflowCount = 10; + for (int i = 0; i < inst.maxPendingRequests; i++) { + assertThat(inst.send(getRequestMsgMock()).isDone(), is(false)); //SettableFuture future - pending only + } + for (int i = 0; i < msgOverflowCount; i++) { + assertThat("max pending requests overflow", inst.send(getRequestMsgMock()).isDone(), is(true)); //overflow, immediate failed future + } + assertThat(inst.pendingRequests.mappingCount(), equalTo(inst.maxPendingRequests)); + verify(inst, times((int) inst.maxPendingRequests)).sendToRequestTemplate(any(), any(), any(), any()); + } + + @Test + public void givenNothing_whenSendAndFetchAndProcessResponsesWithTimeout_thenFail() { + //given + AtomicLong currentTime = new AtomicLong(); + willAnswer(x -> { + log.info("currentTime={}", currentTime.get()); + return currentTime.get(); + }).given(inst).getCurrentClockNs(); + inst.init(); + inst.setupNextCleanup(); + willReturn(Collections.emptyList()).given(inst).doPoll(); + + //when + long stepNs = TimeUnit.MILLISECONDS.toNanos(1); + for (long i = 0; i <= inst.maxRequestTimeoutNs * 2; i = i + stepNs) { + currentTime.addAndGet(stepNs); + assertThat(inst.send(getRequestMsgMock()).isDone(), is(false)); //SettableFuture future - pending only + if (i % (inst.maxRequestTimeoutNs * 3 / 2) == 0) { + inst.fetchAndProcessResponses(); + } + } + + //then + ArgumentCaptor argumentCaptorResp = ArgumentCaptor.forClass(DefaultTbQueueRequestTemplate.ResponseMetaData.class); + ArgumentCaptor argumentCaptorUUID = ArgumentCaptor.forClass(UUID.class); + ArgumentCaptor argumentCaptorLong = ArgumentCaptor.forClass(Long.class); + verify(inst, atLeastOnce()).setTimeoutException(argumentCaptorUUID.capture(), argumentCaptorResp.capture(), argumentCaptorLong.capture()); + + List responseMetaDataList = argumentCaptorResp.getAllValues(); + List tickTsList = argumentCaptorLong.getAllValues(); + for (int i = 0; i < responseMetaDataList.size(); i++) { + assertThat("tickTs >= calculatedExpTime", tickTsList.get(i), greaterThanOrEqualTo(responseMetaDataList.get(i).getSubmitTime() + responseMetaDataList.get(i).getTimeout())); + } + } + + TbQueueMsg getRequestMsgMock() { + return mock(TbQueueMsg.class, RETURNS_DEEP_STUBS); + } +} \ No newline at end of file diff --git a/common/util/pom.xml b/common/util/pom.xml index 6f66da5790..8a0fa1705b 100644 --- a/common/util/pom.xml +++ b/common/util/pom.xml @@ -36,6 +36,10 @@ + + org.springframework + spring-core + com.google.guava guava diff --git a/common/util/src/main/java/org/thingsboard/common/util/TbStopWatch.java b/common/util/src/main/java/org/thingsboard/common/util/TbStopWatch.java new file mode 100644 index 0000000000..90f58ce7f2 --- /dev/null +++ b/common/util/src/main/java/org/thingsboard/common/util/TbStopWatch.java @@ -0,0 +1,57 @@ +/** + * Copyright © 2016-2021 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.common.util; + +import org.springframework.util.StopWatch; + +/** + * Utility method that extends Spring Framework StopWatch + * It is a MONOTONIC time stopwatch. + * It is a replacement for any measurements with a wall-clock like System.currentTimeMillis() + * It is not affected by leap second, day-light saving and wall-clock adjustments by manual or network time synchronization + * The main features is a single call for common use cases: + * - create and start: TbStopWatch sw = TbStopWatch.startNew() + * - stop and get: sw.stopAndGetTotalTimeMillis() or sw.stopAndGetLastTaskTimeMillis() + * */ +public class TbStopWatch extends StopWatch { + + public static TbStopWatch startNew(){ + TbStopWatch stopWatch = new TbStopWatch(); + stopWatch.start(); + return stopWatch; + } + + public long stopAndGetTotalTimeMillis(){ + stop(); + return getTotalTimeMillis(); + } + + public long stopAndGetTotalTimeNanos(){ + stop(); + return getLastTaskTimeNanos(); + } + + public long stopAndGetLastTaskTimeMillis(){ + stop(); + return getLastTaskTimeMillis(); + } + + public long stopAndGetLastTaskTimeNanos(){ + stop(); + return getLastTaskTimeNanos(); + } + +} diff --git a/pom.xml b/pom.xml index b7d0eb71f6..5f88903559 100755 --- a/pom.xml +++ b/pom.xml @@ -1009,6 +1009,11 @@ + + org.springframework + spring-core + ${spring.version} + org.springframework.boot spring-boot-starter-web