From 71324536a30b11a1cdf46af48300686345107ca1 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Tue, 17 Nov 2020 15:01:40 +0200 Subject: [PATCH] Additional Kafka settings and logging of time consuming messages --- .../queue/DefaultTbCoreConsumerService.java | 74 ++++++++++++------- .../src/main/resources/thingsboard.yml | 1 + .../queue/kafka/TbKafkaConsumerTemplate.java | 7 +- .../server/queue/kafka/TbKafkaSettings.java | 5 ++ 4 files changed, 55 insertions(+), 32 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index f56e08b10f..4406ff9ade 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -15,6 +15,8 @@ */ package org.thingsboard.server.service.queue; +import lombok.Getter; +import lombok.Setter; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.context.event.ApplicationReadyEvent; @@ -76,6 +78,7 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.function.Function; import java.util.stream.Collectors; @@ -175,39 +178,48 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService> ctx = new TbPackProcessingContext<>( processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>()); - pendingMap.forEach((id, msg) -> { - log.trace("[{}] Creating main callback for message: {}", id, msg.getValue()); - TbCallback callback = new TbPackCallback<>(id, ctx); - try { - ToCoreMsg toCoreMsg = msg.getValue(); - if (toCoreMsg.hasToSubscriptionMgrMsg()) { - log.trace("[{}] Forwarding message to subscription manager service {}", id, toCoreMsg.getToSubscriptionMgrMsg()); - forwardToSubMgrService(toCoreMsg.getToSubscriptionMgrMsg(), callback); - } else if (toCoreMsg.hasToDeviceActorMsg()) { - log.trace("[{}] Forwarding message to device actor {}", id, toCoreMsg.getToDeviceActorMsg()); - forwardToDeviceActor(toCoreMsg.getToDeviceActorMsg(), callback); - } else if (toCoreMsg.hasDeviceStateServiceMsg()) { - log.trace("[{}] Forwarding message to state service {}", id, toCoreMsg.getDeviceStateServiceMsg()); - forwardToStateService(toCoreMsg.getDeviceStateServiceMsg(), callback); - } else if (toCoreMsg.getToDeviceActorNotificationMsg() != null && !toCoreMsg.getToDeviceActorNotificationMsg().isEmpty()) { - Optional actorMsg = encodingService.decode(toCoreMsg.getToDeviceActorNotificationMsg().toByteArray()); - if (actorMsg.isPresent()) { - TbActorMsg tbActorMsg = actorMsg.get(); - if (tbActorMsg.getMsgType().equals(MsgType.DEVICE_RPC_REQUEST_TO_DEVICE_ACTOR_MSG)) { - tbCoreDeviceRpcService.forwardRpcRequestToDeviceActor((ToDeviceRpcRequestActorMsg) tbActorMsg); - } else { - log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get()); - actorContext.tell(actorMsg.get()); + PendingMsgHolder pendingMsgHolder = new PendingMsgHolder(); + Future packSubmitFuture = consumersExecutor.submit(() -> { + pendingMap.forEach((id, msg) -> { + log.trace("[{}] Creating main callback for message: {}", id, msg.getValue()); + TbCallback callback = new TbPackCallback<>(id, ctx); + try { + ToCoreMsg toCoreMsg = msg.getValue(); + pendingMsgHolder.setToCoreMsg(toCoreMsg); + if (toCoreMsg.hasToSubscriptionMgrMsg()) { + log.trace("[{}] Forwarding message to subscription manager service {}", id, toCoreMsg.getToSubscriptionMgrMsg()); + forwardToSubMgrService(toCoreMsg.getToSubscriptionMgrMsg(), callback); + } else if (toCoreMsg.hasToDeviceActorMsg()) { + log.trace("[{}] Forwarding message to device actor {}", id, toCoreMsg.getToDeviceActorMsg()); + forwardToDeviceActor(toCoreMsg.getToDeviceActorMsg(), callback); + } else if (toCoreMsg.hasDeviceStateServiceMsg()) { + log.trace("[{}] Forwarding message to state service {}", id, toCoreMsg.getDeviceStateServiceMsg()); + forwardToStateService(toCoreMsg.getDeviceStateServiceMsg(), callback); + } else if (toCoreMsg.getToDeviceActorNotificationMsg() != null && !toCoreMsg.getToDeviceActorNotificationMsg().isEmpty()) { + Optional actorMsg = encodingService.decode(toCoreMsg.getToDeviceActorNotificationMsg().toByteArray()); + if (actorMsg.isPresent()) { + TbActorMsg tbActorMsg = actorMsg.get(); + if (tbActorMsg.getMsgType().equals(MsgType.DEVICE_RPC_REQUEST_TO_DEVICE_ACTOR_MSG)) { + tbCoreDeviceRpcService.forwardRpcRequestToDeviceActor((ToDeviceRpcRequestActorMsg) tbActorMsg); + } else { + log.trace("[{}] Forwarding message to App Actor {}", id, actorMsg.get()); + actorContext.tell(actorMsg.get()); + } } + callback.onSuccess(); } - callback.onSuccess(); + } catch (Throwable e) { + log.warn("[{}] Failed to process message: {}", id, msg, e); + callback.onFailure(e); } - } catch (Throwable e) { - log.warn("[{}] Failed to process message: {}", id, msg, e); - callback.onFailure(e); - } + }); }); if (!processingTimeoutLatch.await(packProcessingTimeout, TimeUnit.MILLISECONDS)) { + if (!packSubmitFuture.isDone()) { + packSubmitFuture.cancel(true); + ToCoreMsg lastSubmitMsg = pendingMsgHolder.getToCoreMsg(); + log.info("Timeout to process message: {}", lastSubmitMsg); + } ctx.getAckMap().forEach((id, msg) -> log.debug("[{}] Timeout to process message: {}", id, msg.getValue())); ctx.getFailedMap().forEach((id, msg) -> log.warn("[{}] Failed to process message: {}", id, msg.getValue())); } @@ -227,6 +239,12 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService extends AbstractTbQue private TbKafkaConsumerTemplate(TbKafkaSettings settings, TbKafkaDecoder decoder, String clientId, String groupId, String topic, boolean autoCommit, int autoCommitIntervalMs, - int maxPollRecords, TbQueueAdmin admin) { super(topic); Properties props = settings.toProps(); @@ -54,6 +53,9 @@ public class TbKafkaConsumerTemplate extends AbstractTbQue if (groupId != null) { props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); } + if (settings.getMaxPollIntervalMs() > 0) { + props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, settings.getMaxPollIntervalMs()); + } props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, settings.getMaxPollRecords()); props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, settings.getMaxPartitionFetchBytes()); props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, settings.getFetchMaxBytes()); @@ -61,9 +63,6 @@ public class TbKafkaConsumerTemplate extends AbstractTbQue props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitIntervalMs); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer"); - if (maxPollRecords > 0) { - props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords); - } this.admin = admin; this.consumer = new KafkaConsumer<>(props); this.decoder = decoder; diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java index 37e978c07d..15057d64d3 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaSettings.java @@ -63,6 +63,10 @@ public class TbKafkaSettings { @Getter private int maxPollRecords; + @Value("${queue.kafka.max_poll_interval_ms:0}") + @Getter + private int maxPollIntervalMs; + @Value("${queue.kafka.max_partition_fetch_bytes:16777216}") @Getter private int maxPartitionFetchBytes; @@ -111,4 +115,5 @@ public class TbKafkaSettings { } return props; } + }