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 95745b48fa..4f73a59961 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 @@ -77,6 +77,7 @@ public class RemoteJsInvokeService extends AbstractJsInvokeService { public void init() { TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder requestBuilder = TBKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("producer-js-invoke-" + nodeIdProvider.getNodeId()); requestBuilder.defaultTopic(requestTopic); requestBuilder.encoder(new RemoteJsRequestEncoder()); requestBuilder.enricher((request, responseTopic, requestId) -> { diff --git a/application/src/main/java/org/thingsboard/server/service/transport/RemoteRuleEngineTransportService.java b/application/src/main/java/org/thingsboard/server/service/transport/RemoteRuleEngineTransportService.java index 36b38e262f..4c0f9bdfee 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/RemoteRuleEngineTransportService.java +++ b/application/src/main/java/org/thingsboard/server/service/transport/RemoteRuleEngineTransportService.java @@ -112,6 +112,7 @@ public class RemoteRuleEngineTransportService implements RuleEngineTransportServ public void init() { TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder notificationsProducerBuilder = TBKafkaProducerTemplate.builder(); notificationsProducerBuilder.settings(kafkaSettings); + notificationsProducerBuilder.clientId("producer-transport-notification-" + nodeIdProvider.getNodeId()); notificationsProducerBuilder.encoder(new ToTransportMsgEncoder()); notificationsProducer = notificationsProducerBuilder.build(); diff --git a/application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java b/application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java index 770648be16..724fb3f8b4 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java +++ b/application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java @@ -68,6 +68,7 @@ public class RemoteTransportApiService { TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder responseBuilder = TBKafkaProducerTemplate.builder(); responseBuilder.settings(kafkaSettings); + responseBuilder.clientId("producer-transport-api-response-" + nodeIdProvider.getNodeId()); responseBuilder.encoder(new TransportApiResponseEncoder()); TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder requestBuilder = TBKafkaConsumerTemplate.builder(); diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaProducerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaProducerTemplate.java index ad489c14a3..bcb6797781 100644 --- a/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaProducerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaProducerTemplate.java @@ -62,10 +62,13 @@ public class TBKafkaProducerTemplate { @Builder private TBKafkaProducerTemplate(TbKafkaSettings settings, TbKafkaEncoder encoder, TbKafkaEnricher enricher, - TbKafkaPartitioner partitioner, String defaultTopic) { + TbKafkaPartitioner partitioner, String defaultTopic, String clientId) { Properties props = settings.toProps(); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer"); + if (!StringUtils.isEmpty(clientId)) { + props.put(ProducerConfig.CLIENT_ID_CONFIG, clientId); + } this.settings = settings; this.producer = new KafkaProducer<>(props); this.encoder = encoder; diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/RemoteTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/RemoteTransportService.java index c13569c1f1..58f3ce22a9 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/RemoteTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/RemoteTransportService.java @@ -104,6 +104,7 @@ public class RemoteTransportService extends AbstractTransportService { TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder requestBuilder = TBKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("producer-transport-api-request-" + nodeIdProvider.getNodeId()); requestBuilder.defaultTopic(transportApiRequestsTopic); requestBuilder.encoder(new TransportApiRequestEncoder()); @@ -128,6 +129,7 @@ public class RemoteTransportService extends AbstractTransportService { TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder ruleEngineProducerBuilder = TBKafkaProducerTemplate.builder(); ruleEngineProducerBuilder.settings(kafkaSettings); + ruleEngineProducerBuilder.clientId("producer-rule-engine-request-" + nodeIdProvider.getNodeId()); ruleEngineProducerBuilder.defaultTopic(ruleEngineTopic); ruleEngineProducerBuilder.encoder(new ToRuleEngineMsgEncoder()); ruleEngineProducer = ruleEngineProducerBuilder.build(); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java index 3a2970d750..5486b50352 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/kafka/TbKafkaNode.java @@ -25,6 +25,7 @@ import org.thingsboard.server.common.msg.TbMsgMetaData; import java.util.Properties; import java.util.concurrent.ExecutionException; +import java.util.concurrent.atomic.AtomicInteger; @Slf4j @RuleNode( @@ -54,6 +55,7 @@ public class TbKafkaNode implements TbNode { public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { this.config = TbNodeUtils.convert(configuration, TbKafkaNodeConfiguration.class); Properties properties = new Properties(); + properties.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-tb-kafka-node-" + ctx.getSelfId().getId().toString() + "-" + ctx.getNodeId()); properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServers()); properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, config.getValueSerializer()); properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, config.getKeySerializer()); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetTelemetryNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetTelemetryNode.java index f0af22dd40..82df912f9c 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetTelemetryNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetTelemetryNode.java @@ -24,6 +24,7 @@ import com.google.common.util.concurrent.ListenableFuture; import lombok.Data; import lombok.NoArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.math.NumberUtils; import org.thingsboard.rule.engine.api.RuleNode; import org.thingsboard.rule.engine.api.TbContext; @@ -66,6 +67,8 @@ import static org.thingsboard.server.common.data.kv.Aggregation.NONE; configDirective = "tbEnrichmentNodeGetTelemetryFromDatabase") public class TbGetTelemetryNode implements TbNode { + private static final String DESC_ORDER = "DESC"; + private TbGetTelemetryNodeConfiguration config; private List tsKeyNames; private int limit; @@ -80,6 +83,9 @@ public class TbGetTelemetryNode implements TbNode { limit = config.getFetchMode().equals(FETCH_MODE_ALL) ? MAX_FETCH_SIZE : 1; fetchMode = config.getFetchMode(); orderBy = config.getOrderBy(); + if (StringUtils.isEmpty(orderBy)) { + orderBy = DESC_ORDER; + } mapper = new ObjectMapper(); mapper.configure(JsonGenerator.Feature.QUOTE_FIELD_NAMES, false); mapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES, true); diff --git a/ui/src/app/components/widget/widget.controller.js b/ui/src/app/components/widget/widget.controller.js index 3262adc1bb..a58064320c 100644 --- a/ui/src/app/components/widget/widget.controller.js +++ b/ui/src/app/components/widget/widget.controller.js @@ -138,7 +138,7 @@ export default function WidgetController($scope, $state, $timeout, $window, $ele headerAction.icon = descriptor.icon; headerAction.descriptor = descriptor; headerAction.onAction = function($event) { - var entityInfo = getFirstEntityInfo(); + var entityInfo = getActiveEntityInfo(); var entityId = entityInfo ? entityInfo.entityId : null; var entityName = entityInfo ? entityInfo.entityName : null; handleWidgetAction($event, this.descriptor, entityId, entityName); @@ -502,13 +502,15 @@ export default function WidgetController($scope, $state, $timeout, $window, $ele } } - function getFirstEntityInfo() { - var entityInfo; - for (var id in widgetContext.subscriptions) { - var subscription = widgetContext.subscriptions[id]; - entityInfo = subscription.getFirstEntityInfo(); - if (entityInfo) { - break; + function getActiveEntityInfo() { + var entityInfo = widgetContext.activeEntityInfo; + if (!entityInfo) { + for (var id in widgetContext.subscriptions) { + var subscription = widgetContext.subscriptions[id]; + entityInfo = subscription.getFirstEntityInfo(); + if (entityInfo) { + break; + } } } return entityInfo; diff --git a/ui/src/app/widget/lib/timeseries-table-widget.js b/ui/src/app/widget/lib/timeseries-table-widget.js index 6a34bd574a..780065bb25 100644 --- a/ui/src/app/widget/lib/timeseries-table-widget.js +++ b/ui/src/app/widget/lib/timeseries-table-widget.js @@ -44,7 +44,7 @@ function TimeseriesTableWidget() { } /*@ngInject*/ -function TimeseriesTableWidgetController($element, $scope, $filter, $timeout) { +function TimeseriesTableWidgetController($element, $scope, $filter, $timeout, types) { var vm = this; let dateFormatFilter = 'yyyy-MM-dd HH:mm:ss'; @@ -228,9 +228,29 @@ function TimeseriesTableWidgetController($element, $scope, $filter, $timeout) { $scope.$watch('vm.sourceIndex', function(newIndex, oldIndex) { if (newIndex != oldIndex) { updateSourceData(vm.sources[vm.sourceIndex]); + updateActiveEntityInfo(); } }); + function updateActiveEntityInfo() { + var source = vm.sources[vm.sourceIndex]; + var activeEntityInfo = null; + if (source) { + var datasource = source.datasource; + if (datasource.type === types.datasourceType.entity && + datasource.entityType && datasource.entityId) { + activeEntityInfo = { + entityId: { + entityType: datasource.entityType, + id: datasource.entityId + }, + entityName: datasource.entityName + }; + } + } + vm.ctx.activeEntityInfo = activeEntityInfo; + } + function updateDatasources() { vm.sources = []; vm.sourceIndex = 0; @@ -314,6 +334,7 @@ function TimeseriesTableWidgetController($element, $scope, $filter, $timeout) { vm.sources.push(source); } } + updateActiveEntityInfo(); } function updatePage(source) {