tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Cancelling consumer loop task on stop

pull/9453/head
ViacheslavKlimov 3 years ago
parent
921b2ae4b7
commit
ea4eadb47f
6 changed files with 78 additions and 49 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 1
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java
  2. 2
      application/src/main/java/org/thingsboard/server/service/queue/ruleengine/QueueEvent.java
  3. 34
      application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerTask.java
  4. 10
      application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineConsumerContext.java
  5. 70
      application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java
  6. 10
      application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java

1
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java
View File

@ -126,6 +126,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
@Override
protected void stopConsumers() {
consumers.values().forEach(TbRuleEngineQueueConsumerManager::stop);
consumers.values().forEach(TbRuleEngineQueueConsumerManager::awaitStop);
ctx.stop();
}

2
application/src/main/java/org/thingsboard/server/service/queue/ruleengine/QueueEvent.java
View File

@ -19,6 +19,6 @@ import java.io.Serializable;
public enum QueueEvent implements Serializable {
PARTITION_CHANGE, CONFIG_UPDATE, STOP, DELETE
PARTITION_CHANGE, CONFIG_UPDATE, DELETE
}

34
application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbQueueConsumerTask.java
View File

@ -15,7 +15,8 @@
*/
package org.thingsboard.server.service.queue.ruleengine;
import lombok.Data;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.gen.transport.TransportProtos;
@ -23,16 +24,26 @@ import org.thingsboard.server.queue.TbQueueConsumer;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@Data
@RequiredArgsConstructor
@Slf4j
public class TbQueueConsumerTask {
@Getter
private final Object key;
@Getter
private final TbQueueConsumer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> consumer;
private volatile Future<?> task;
private Future<?> task;
private CountDownLatch completionLatch;
public void setTask(Future<?> task) {
this.task = task;
this.completionLatch = new CountDownLatch(1);
}
public void subscribe(Set<TopicPartitionInfo> partitions) {
log.trace("[{}] Subscribing to partitions: {}", key, partitions);
@ -42,23 +53,32 @@ public class TbQueueConsumerTask {
public void initiateStop() {
log.debug("[{}] Initiating stop", key);
consumer.stop();
if (isRunning()) {
task.cancel(true);
}
}
public void awaitFinish() {
public void awaitCompletion() {
log.trace("[{}] Awaiting finish", key);
if (isRunning()) {
try {
task.get(60, TimeUnit.SECONDS);
task = null;
if (!completionLatch.await(30, TimeUnit.SECONDS)) {
throw new IllegalStateException("timeout of 30 seconds expired");
}
} catch (Exception e) {
log.warn("[{}] Failed to await for consumer to stop", key, e);
}
task = null;
}
log.trace("[{}] Awaited finish", key);
}
public boolean isRunning() {
return task != null && !task.isDone();
return task != null;
}
public void finished() {
completionLatch.countDown();
}
}

10
application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineConsumerContext.java
View File

@ -36,7 +36,6 @@ import javax.annotation.PostConstruct;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@Component
@TbRuleEngineComponent
@ -83,12 +82,7 @@ public class TbRuleEngineConsumerContext {
public void stop() {
scheduler.shutdownNow();
consumersExecutor.shutdown();
mgmtExecutor.shutdown();
try {
mgmtExecutor.awaitTermination(15, TimeUnit.SECONDS);
} catch (InterruptedException e) {
log.warn("Failed to await mgmtExecutor termination");
}
consumersExecutor.shutdownNow();
mgmtExecutor.shutdownNow();
}
}

70
application/src/main/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManager.java
View File

@ -85,7 +85,13 @@ public class TbRuleEngineQueueConsumerManager {
}
public void init(Queue queue) {
doInit(queue);
this.queue = queue;
if (queue.isConsumerPerPartition()) {
this.consumerWrapper = new ConsumerPerPartitionWrapper();
} else {
this.consumerWrapper = new SingleConsumerWrapper();
}
log.debug("[{}] Initialized consumer for queue: {}", queueKey, queue);
}
public void update(Queue queue) {
@ -96,10 +102,6 @@ public class TbRuleEngineQueueConsumerManager {
addTask(new TbQueueConsumerManagerTask(QueueEvent.PARTITION_CHANGE, partitions));
}
public void stop() {
addTask(new TbQueueConsumerManagerTask(QueueEvent.STOP));
}
public void delete() {
addTask(new TbQueueConsumerManagerTask(QueueEvent.DELETE));
}
@ -135,20 +137,22 @@ public class TbRuleEngineQueueConsumerManager {
newPartitions = task.getPartitions();
} else if (task.getEvent() == QueueEvent.CONFIG_UPDATE) {
newConfiguration = task.getQueue();
} else if (task.getEvent() == QueueEvent.STOP) {
doStop();
return;
} else if (task.getEvent() == QueueEvent.DELETE) {
doDelete();
return;
}
}
if (stopped) {
return;
}
if (newConfiguration != null) {
doUpdate(newConfiguration);
}
if (newPartitions != null) {
doUpdate(newPartitions);
}
} catch (Exception e) {
log.error("[{}] Failed to process tasks", queueKey, e);
} finally {
lock.unlock();
}
@ -159,16 +163,6 @@ public class TbRuleEngineQueueConsumerManager {
});
}
private void doInit(Queue queue) {
this.queue = queue;
if (queue.isConsumerPerPartition()) {
consumerWrapper = new ConsumerPerPartitionWrapper();
} else {
consumerWrapper = new SingleConsumerWrapper();
}
log.debug("[{}] Initialized consumer for queue: {}", queueKey, queue);
}
private void doUpdate(Queue newQueue) {
log.info("[{}] Processing queue update: {}", queueKey, newQueue);
var oldQueue = this.queue;
@ -179,12 +173,12 @@ public class TbRuleEngineQueueConsumerManager {
}
if (oldQueue == null) {
doInit(queue);
init(queue);
} else if (newQueue.isConsumerPerPartition() != oldQueue.isConsumerPerPartition()) {
consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::initiateStop);
consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitFinish);
consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitCompletion);
doInit(queue);
init(queue);
if (partitions != null) {
doUpdate(partitions); // even if partitions number was changed, there can be no partition change event
}
@ -200,17 +194,21 @@ public class TbRuleEngineQueueConsumerManager {
consumerWrapper.updatePartitions(partitions);
}
private void doStop() {
public void stop() {
log.debug("[{}] Stopping consumers", queueKey);
consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::initiateStop);
consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitFinish);
log.debug("[{}] Unsubscribed and stopped consumers", queueKey);
stopped = true;
}
public void awaitStop() {
consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitCompletion);
log.debug("[{}] Unsubscribed and stopped consumers", queueKey);
}
private void doDelete() {
stopped = true;
log.info("[{}] Handling queue deletion", queueKey);
consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitFinish);
consumerWrapper.getConsumers().forEach(TbQueueConsumerTask::awaitCompletion);
List<TbQueueConsumer<TbProtoQueueMsg<ToRuleEngineMsg>>> queueConsumers = consumerWrapper.getConsumers().stream()
.map(TbQueueConsumerTask::getConsumer).collect(Collectors.toList());
@ -240,6 +238,7 @@ public class TbRuleEngineQueueConsumerManager {
Future<?> consumerLoop = ctx.getConsumersExecutor().submit(() -> {
ThingsBoardThreadFactory.updateCurrentThreadName(consumerTask.getKey().toString());
consumerLoop(consumerTask.getConsumer());
consumerTask.finished();
});
consumerTask.setTask(consumerLoop);
}
@ -253,7 +252,7 @@ public class TbRuleEngineQueueConsumerManager {
}
processMsgs(msgs, consumer, queue);
} catch (Exception e) {
if (!stopped) {
if (!consumer.isStopped()) {
log.warn("Failed to process messages from queue", e);
try {
Thread.sleep(ctx.getPollDuration());
@ -279,7 +278,7 @@ public class TbRuleEngineQueueConsumerManager {
TbMsgPackProcessingContext packCtx = new TbMsgPackProcessingContext(queue.getName(), submitStrategy, ackStrategy.isSkipTimeoutMsgs());
submitStrategy.submitAttempt((id, msg) -> submitMessage(packCtx, id, msg));
final boolean timeout = !packCtx.await(queue.getPackProcessingTimeout(), TimeUnit.MILLISECONDS);
final boolean timeout = !awaitPackProcessing(packCtx, queue.getPackProcessingTimeout(), true);
TbRuleEngineProcessingResult result = new TbRuleEngineProcessingResult(queue.getName(), timeout, packCtx);
if (timeout) {
@ -307,6 +306,19 @@ public class TbRuleEngineQueueConsumerManager {
}
}
private boolean awaitPackProcessing(TbMsgPackProcessingContext packCtx, long processingTimeout, boolean ignoreInterrupt) throws InterruptedException {
try {
return packCtx.await(processingTimeout, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
if (ignoreInterrupt) {
log.debug("Interrupt happened while waiting for pack processing, trying to await one more time");
return awaitPackProcessing(packCtx, processingTimeout, false);
} else {
throw new RuntimeException("Failed to await pack processing due to thread interrupt", e);
}
}
}
private TbRuleEngineSubmitStrategy getSubmitStrategy(Queue queue) {
return ctx.getSubmitStrategyFactory().newInstance(queue.getName(), queue.getSubmitStrategy());
}
@ -430,7 +442,7 @@ public class TbRuleEngineQueueConsumerManager {
consumers.get(tpi).initiateStop();
});
removedPartitions.forEach((tpi) -> {
consumers.remove(tpi).awaitFinish();
consumers.remove(tpi).awaitCompletion();
});
addedPartitions.forEach((tpi) -> {
@ -457,7 +469,7 @@ public class TbRuleEngineQueueConsumerManager {
if (partitions.isEmpty()) {
if (consumer != null && consumer.isRunning()) {
consumer.initiateStop();
consumer.awaitFinish();
consumer.awaitCompletion();
}
consumer = null;
return;

10
application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java
View File

@ -180,13 +180,16 @@ public class TbRuleEngineQueueConsumerManagerTest {
@After
public void afterEach() {
consumerManager.stop();
consumerManager.awaitStop();
ruleEngineConsumerContext.stop();
if (generateQueueMsgs) {
await().atMost(1, TimeUnit.SECONDS)
.until(() -> totalProcessedMsgs.get() == totalConsumedMsgs.get());
await().atMost(2, TimeUnit.SECONDS)
.untilAsserted(() -> {
log.debug("totalConsumedMsgs = {}, totalProcessedMsgs = {}", totalConsumedMsgs.get(), totalProcessedMsgs.get());
assertThat(totalProcessedMsgs.get()).isEqualTo(totalConsumedMsgs.get());
});
}
log.debug("totalConsumedMsgs = {}, totalProcessedMsgs = {}", totalConsumedMsgs.get(), totalProcessedMsgs.get());
}
@Test
@ -510,7 +513,6 @@ public class TbRuleEngineQueueConsumerManagerTest {
verify(consumer).unsubscribe();
int movedMsgs = consumer.msgCount - msgCount;
assertThat(movedMsgs).isGreaterThan(10);
verify(ruleEngineMsgProducer, atLeast(movedMsgs)).send(any(), any(), any());
verify(actorContext, never()).tell(any());
generateQueueMsgs = false;

Write Preview
Loading…
Cancel
Save