From 2ccce3b6d9c3d65382d2d3db7bae66861bbaad9a Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Tue, 24 Mar 2020 14:08:21 +0200 Subject: [PATCH] Main Logic of RuleChainActor to handle queue messages --- .../server/actors/ActorSystemContext.java | 10 + .../device/DeviceActorMessageProcessor.java | 28 +-- .../actors/ruleChain/DefaultTbContext.java | 8 +- .../actors/ruleChain/RuleChainActor.java | 6 +- .../RuleChainActorMessageProcessor.java | 214 +++++++++--------- .../ruleChain/RuleChainToRuleChainMsg.java | 1 - .../server/actors/tenant/TenantActor.java | 12 - .../DefaultTbRuleEngineConsumerService.java | 3 +- .../server/common/msg/MsgType.java | 1 - .../thingsboard/server/common/msg/TbMsg.java | 30 ++- .../MultipleTbQueueTbMsgCallbackWrapper.java | 43 ++++ .../queue/TbQueueTbMsgCallbackWrapper.java | 32 +-- .../ConsistentHashPartitionService.java | 31 ++- .../queue/discovery/PartitionService.java | 3 +- .../queue/discovery/TopicPartitionInfo.java | 12 +- .../discovery/TopicPartitionInfoKey.java | 43 ++++ .../queue/kafka/TBKafkaConsumerTemplate.java | 2 +- 17 files changed, 292 insertions(+), 187 deletions(-) create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/MultipleTbQueueTbMsgCallbackWrapper.java rename application/src/main/java/org/thingsboard/server/actors/device/DeviceActorToRuleEngineMsg.java => common/queue/src/main/java/org/thingsboard/server/queue/TbQueueTbMsgCallbackWrapper.java (52%) create mode 100644 common/queue/src/main/java/org/thingsboard/server/queue/discovery/TopicPartitionInfoKey.java diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index 46e6df60ab..e5158377cd 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -63,7 +63,9 @@ import org.thingsboard.server.dao.rule.RuleChainService; import org.thingsboard.server.dao.tenant.TenantService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.dao.user.UserService; +import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; +import org.thingsboard.server.queue.provider.TbRuleEngineQueueProvider; import org.thingsboard.server.service.component.ComponentDiscoveryService; import org.thingsboard.server.service.encoding.DataDecodingEncodingService; import org.thingsboard.server.service.executors.ClusterRpcCallbackExecutorService; @@ -151,6 +153,14 @@ public class ActorSystemContext { @Getter private RuleChainService ruleChainService; + @Autowired + @Getter + private PartitionService partitionService; + + @Autowired + @Getter + private TbRuleEngineQueueProvider ruleEngineQueueProvider; + @Autowired @Getter private TimeseriesService tsService; diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index 1844539b2c..4713662067 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java @@ -331,18 +331,18 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { } private void handleClientSideRPCRequest(ActorContext context, SessionInfoProto sessionInfo, TransportProtos.ToServerRpcRequestMsg request) { - UUID sessionId = getSessionId(sessionInfo); - JsonObject json = new JsonObject(); - json.addProperty("method", request.getMethodName()); - json.add("params", JsonUtils.parse(request.getParams())); - - TbMsgMetaData requestMetaData = defaultMetaData.copy(); - requestMetaData.putValue("requestId", Integer.toString(request.getRequestId())); - TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), SessionMsgType.TO_SERVER_RPC_REQUEST.name(), deviceId, requestMetaData, TbMsgDataType.JSON, gson.toJson(json), null, null, null); - context.parent().tell(new DeviceActorToRuleEngineMsg(context.self(), tbMsg), context.self()); - - scheduleMsgWithDelay(context, new DeviceActorClientSideRpcTimeoutMsg(request.getRequestId(), systemContext.getClientSideRpcTimeout()), systemContext.getClientSideRpcTimeout()); - toServerRpcPendingMap.put(request.getRequestId(), new ToServerRpcRequestMetadata(sessionId, getSessionType(sessionId), sessionInfo.getNodeId())); +// UUID sessionId = getSessionId(sessionInfo); +// JsonObject json = new JsonObject(); +// json.addProperty("method", request.getMethodName()); +// json.add("params", JsonUtils.parse(request.getParams())); +// +// TbMsgMetaData requestMetaData = defaultMetaData.copy(); +// requestMetaData.putValue("requestId", Integer.toString(request.getRequestId())); +// TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), SessionMsgType.TO_SERVER_RPC_REQUEST.name(), deviceId, requestMetaData, TbMsgDataType.JSON, gson.toJson(json), null, null, null); +// context.parent().tell(new DeviceActorToRuleEngineMsg(context.self(), tbMsg), context.self()); +// +// scheduleMsgWithDelay(context, new DeviceActorClientSideRpcTimeoutMsg(request.getRequestId(), systemContext.getClientSideRpcTimeout()), systemContext.getClientSideRpcTimeout()); +// toServerRpcPendingMap.put(request.getRequestId(), new ToServerRpcRequestMetadata(sessionId, getSessionType(sessionId), sessionInfo.getNodeId())); } private TransportProtos.SessionType getSessionType(UUID sessionId) { @@ -372,10 +372,6 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { } } - private void pushToRuleEngine(ActorContext context, TbMsg tbMsg) { - context.parent().tell(new DeviceActorToRuleEngineMsg(context.self(), tbMsg), context.self()); - } - void processAttributesUpdate(ActorContext context, DeviceAttributesEventNotificationMsg msg) { if (attributeSubscriptions.size() > 0) { boolean hasNotificationData = false; diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 337a234e2d..a5b5254174 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -65,6 +65,7 @@ import org.thingsboard.server.dao.rule.RuleChainService; import org.thingsboard.server.dao.tenant.TenantService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.dao.user.UserService; +import org.thingsboard.server.queue.discovery.ServiceType; import org.thingsboard.server.service.script.RuleNodeJsScriptEngine; import scala.concurrent.duration.Duration; @@ -118,10 +119,7 @@ class DefaultTbContext implements TbContext { @Override public boolean isLocalEntity(EntityId entityId) { - //TODO 2.5 -// Optional address = mainCtx.getRoutingService().resolveById(entityId); -// return !address.isPresent(); - return true; + return mainCtx.getPartitionService().resolve(ServiceType.TB_RULE_ENGINE, getTenantId(), entityId).isMyPartition(); } private void scheduleMsgWithDelay(Object msg, long delayInMs, ActorRef target) { @@ -143,7 +141,7 @@ class DefaultTbContext implements TbContext { @Override public TbMsg newMsg(String type, EntityId originator, TbMsgMetaData metaData, String data) { - return new TbMsg(UUIDs.timeBased(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId(),null); + return new TbMsg(UUIDs.timeBased(), type, originator, metaData.copy(), TbMsgDataType.JSON, data, nodeCtx.getSelf().getRuleChainId(), nodeCtx.getSelf().getId(), null); } @Override diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java index 76f831be98..83cde28a45 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java @@ -18,7 +18,6 @@ package org.thingsboard.server.actors.ruleChain; import akka.actor.OneForOneStrategy; import akka.actor.SupervisorStrategy; import org.thingsboard.server.actors.ActorSystemContext; -import org.thingsboard.server.actors.device.DeviceActorToRuleEngineMsg; import org.thingsboard.server.actors.service.ComponentActor; import org.thingsboard.server.actors.service.ContextBasedCreator; import org.thingsboard.server.common.data.id.RuleChainId; @@ -43,10 +42,7 @@ public class RuleChainActor extends ComponentActor { - private static final long DEFAULT_CLUSTER_PARTITION = 0L; private final ActorRef parent; private final ActorRef self; private final Map nodeActors; private final Map> nodeRoutes; private final RuleChainService service; + private final PartitionService partitionService; + private final TbQueueProducer> producer; private RuleNodeId firstId; private RuleNodeCtx firstNode; private boolean started; - private String ruleChainName; RuleChainActorMessageProcessor(TenantId tenantId, RuleChainId ruleChainId, ActorSystemContext systemContext , ActorRef parent, ActorRef self) { @@ -76,7 +83,8 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor(); this.nodeRoutes = new HashMap<>(); this.service = systemContext.getRuleChainService(); - this.ruleChainName = ruleChainId.toString(); + this.partitionService = systemContext.getPartitionService(); + this.producer = systemContext.getRuleEngineQueueProvider().getRuleEngineMsgProducer(); } @Override @@ -89,7 +97,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor ruleNodeList = service.getRuleChainNodes(tenantId, entityId); log.trace("[{}][{}] Starting rule chain with {} nodes", tenantId, entityId, ruleNodeList.size()); // Creating and starting the actors; @@ -110,7 +117,6 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor ruleNodeList = service.getRuleChainNodes(tenantId, entityId); log.trace("[{}][{}] Updating rule chain with {} nodes", tenantId, entityId, ruleNodeList.size()); for (RuleNode ruleNode : ruleNodeList) { @@ -189,101 +195,113 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor address = systemContext.getRoutingService().resolveById(originatorEntityId); -// if (address.isPresent()) { -// onRemoteTellNext(address.get(), envelope); -// } else { - onLocalTellNext(envelope); -// } + try { + checkActive(); + EntityId entityId = msg.getOriginator(); + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tenantId, entityId); + RuleNodeId originatorNodeId = envelope.getOriginator(); + List relations = nodeRoutes.get(originatorNodeId).stream() + .filter(r -> contains(envelope.getRelationTypes(), r.getType())) + .collect(Collectors.toList()); + int relationsCount = relations.size(); + if (relationsCount == 0) { + log.trace("[{}][{}][{}] No outbound relations to process", tenantId, entityId, msg.getId()); + //TODO 2.5: Maybe let's check that the output relation is not a Failure? + if (envelope.getRelationTypes().contains(TbRelationTypes.FAILURE)) { + log.debug("[{}] Failure during message processing by Rule Node [{}]. Enable and see debug events for more info", entityId, envelope.getOriginator().getId()); + msg.getCallback().onFailure(new RuntimeException("Failure during message processing by Rule Node [" + envelope.getOriginator().getId().toString() + "]")); + } else { + msg.getCallback().onSuccess(); + } + } else if (relationsCount == 1) { + for (RuleNodeRelation relation : relations) { + log.trace("[{}][{}][{}] Pushing message to single target: [{}]", tenantId, entityId, msg.getId(), relation.getOut()); + pushToTarget(tpi, msg, relation.getOut(), relation.getType()); + } + } else { + MultipleTbQueueTbMsgCallbackWrapper callbackWrapper = new MultipleTbQueueTbMsgCallbackWrapper(relationsCount, msg.getCallback()); + log.trace("[{}][{}][{}] Pushing message to multiple targets: [{}]", tenantId, entityId, msg.getId(), relations); + for (RuleNodeRelation relation : relations) { + EntityId target = relation.getOut(); + putToQueue(tpi, msg, callbackWrapper, target); + } + } + } catch (Exception e) { + msg.getCallback().onFailure(e); + } } - private void onRemoteTellNext(ServerAddress serverAddress, RuleNodeToRuleChainTellNextMsg envelope) { - TbMsg msg = envelope.getMsg(); - log.debug("Forwarding [{}] msg to remote server [{}] due to changed originator id: [{}]", msg.getId(), serverAddress, msg.getOriginator()); - envelope = new RemoteToRuleChainTellNextMsg(envelope, tenantId, entityId); - //TODO 2.5 -// systemContext.getRpcService().tell(systemContext.getEncodingService().convertToProtoDataMessage(serverAddress, envelope)); + private void putToQueue(TopicPartitionInfo tpi, TbMsg msg, TbQueueCallback callbackWrapper, EntityId target) { + switch (target.getEntityType()) { + case RULE_NODE: + putToQueue(tpi, msg.copyWithRuleNodeId(entityId, new RuleNodeId(target.getId())), callbackWrapper); + break; + case RULE_CHAIN: + putToQueue(tpi, msg.copyWithRuleChainId(new RuleChainId(target.getId())), callbackWrapper); + break; + } } - private void onLocalTellNext(RuleNodeToRuleChainTellNextMsg envelope) { - TbMsg msg = envelope.getMsg(); - RuleNodeId originatorNodeId = envelope.getOriginator(); - List relations = nodeRoutes.get(originatorNodeId).stream() - .filter(r -> contains(envelope.getRelationTypes(), r.getType())) - .collect(Collectors.toList()); - int relationsCount = relations.size(); - EntityId ackId = msg.getRuleNodeId() != null ? msg.getRuleNodeId() : msg.getRuleChainId(); - if (relationsCount == 0) { - log.trace("[{}][{}][{}] No outbound relations to process", tenantId, entityId, msg.getId()); - if (ackId != null) { -// TODO: Ack this message in Kafka -// queue.ack(tenantId, msg, ackId.getId(), msg.getClusterPartition()); - } - } else if (relationsCount == 1) { - for (RuleNodeRelation relation : relations) { - log.trace("[{}][{}][{}] Pushing message to single target: [{}]", tenantId, entityId, msg.getId(), relation.getOut()); - pushToTarget(msg, relation.getOut(), relation.getType()); + private void pushToTarget(TopicPartitionInfo tpi, TbMsg msg, EntityId target, String fromRelationType) { + if (tpi.isMyPartition()) { + switch (target.getEntityType()) { + case RULE_NODE: + pushMsgToNode(nodeActors.get(new RuleNodeId(target.getId())), msg, fromRelationType); + break; + case RULE_CHAIN: + parent.tell(new RuleChainToRuleChainMsg(new RuleChainId(target.getId()), entityId, msg, fromRelationType), self); + break; } } else { - for (RuleNodeRelation relation : relations) { - EntityId target = relation.getOut(); - log.trace("[{}][{}][{}] Pushing message to multiple targets: [{}]", tenantId, entityId, msg.getId(), relation.getOut()); - switch (target.getEntityType()) { - case RULE_NODE: - enqueueAndForwardMsgCopyToNode(msg, target, relation.getType()); - break; - case RULE_CHAIN: - enqueueAndForwardMsgCopyToChain(msg, target, relation.getType()); - break; - } - } - //TODO: Ideally this should happen in async way when all targets confirm that the copied messages are successfully written to corresponding target queues. - if (ackId != null) { -// TODO: Ack this message in Kafka -// queue.ack(tenantId, msg, ackId.getId(), msg.getClusterPartition()); - } + putToQueue(tpi, msg, new TbQueueTbMsgCallbackWrapper(msg.getCallback()), target); } } + private void putToQueue(TopicPartitionInfo tpi, TbMsg newMsg, TbQueueCallback callbackWrapper) { + ToRuleEngineMsg toQueueMsg = ToRuleEngineMsg.newBuilder() + .setTenantIdMSB(tenantId.getId().getMostSignificantBits()) + .setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) + .setTbMsg(ByteString.copyFrom(TbMsg.toByteArray(newMsg))) + .build(); + producer.send(tpi, new TbProtoQueueMsg<>(newMsg.getId(), toQueueMsg), callbackWrapper); + } + private boolean contains(Set relationTypes, String type) { if (relationTypes == null) { return true; @@ -296,38 +314,10 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor> myPartitions = new ConcurrentHashMap<>(); //TODO: Fetch this from the database, together with size of partitions for each service for each tenant. private ConcurrentMap> isolatedTenants = new ConcurrentHashMap<>(); + private ConcurrentMap tpiCache = new ConcurrentHashMap<>(); private Map tbCoreNotificationTopics = new HashMap<>(); private Map tbRuleEngineNotificationTopics = new HashMap<>(); @@ -87,12 +89,12 @@ public class ConsistentHashPartitionService implements PartitionService { } @Override - public List getCurrentPartitions(ServiceType serviceType) { + public Set getCurrentPartitions(ServiceType serviceType) { ServiceInfo currentService = serviceInfoProvider.getServiceInfo(); TenantId tenantId = getSystemOrIsolatedTenantId(currentService); ServiceKey serviceKey = new ServiceKey(serviceType, tenantId); List partitions = myPartitions.get(serviceKey); - List topicPartitions = new ArrayList<>(); + Set topicPartitions = new LinkedHashSet<>(); for (Integer partition : partitions) { TopicPartitionInfo.TopicPartitionInfoBuilder tpi = TopicPartitionInfo.builder(); tpi.topic(partitionTopics.get(serviceType)); @@ -112,7 +114,9 @@ public class ConsistentHashPartitionService implements PartitionService { .putLong(entityId.getId().getMostSignificantBits()) .putLong(entityId.getId().getLeastSignificantBits()).hash().asInt(); int partition = Math.abs(hash % partitionSizes.get(serviceType)); - return buildTopicPartitionInfo(serviceType, tenantId, partition); + boolean isolatedTenant = isIsolated(serviceType, tenantId); + TopicPartitionInfoKey cacheKey = new TopicPartitionInfoKey(serviceType, isolatedTenant ? tenantId : null, partition); + return tpiCache.computeIfAbsent(cacheKey, key -> buildTopicPartitionInfo(serviceType, tenantId, partition)); } @Override @@ -156,8 +160,8 @@ public class ConsistentHashPartitionService implements PartitionService { tpiList.add(getNotificationsTopic(serviceKey.getServiceType(), serviceInfoProvider.getServiceId())); applicationEventPublisher.publishEvent(new PartitionChangeEvent(this, serviceKey, tpiList)); } - }); + tpiCache.clear(); if (currentOtherServices == null) { currentOtherServices = new ArrayList<>(otherServices); @@ -207,7 +211,7 @@ public class ConsistentHashPartitionService implements PartitionService { } private TopicPartitionInfo buildTopicPartitionInfo(ServiceType serviceType, String serviceId) { - return new TopicPartitionInfo(serviceType.name().toLowerCase() + "." + serviceId, null, null); + return new TopicPartitionInfo(serviceType.name().toLowerCase() + "." + serviceId, null, null, false); } private TopicPartitionInfo buildTopicPartitionInfo(ServiceKey serviceKey, int partition) { @@ -215,16 +219,29 @@ public class ConsistentHashPartitionService implements PartitionService { } private TopicPartitionInfo buildTopicPartitionInfo(ServiceType serviceType, TenantId tenantId, int partition) { - boolean isolated = isolatedTenants.get(tenantId) != null && isolatedTenants.get(tenantId).contains(serviceType); TopicPartitionInfo.TopicPartitionInfoBuilder tpi = TopicPartitionInfo.builder(); tpi.topic(partitionTopics.get(serviceType)); tpi.partition(partition); - if (isolated) { + ServiceKey myPartitionsSearchKey; + if (isIsolated(serviceType, tenantId)) { tpi.tenantId(tenantId); + myPartitionsSearchKey = new ServiceKey(serviceType, tenantId); + } else { + myPartitionsSearchKey = new ServiceKey(serviceType, new TenantId(TenantId.NULL_UUID)); + } + List partitions = myPartitions.get(myPartitionsSearchKey); + if (partitions != null) { + tpi.myPartition(partitions.contains(partition)); + } else { + tpi.myPartition(false); } return tpi.build(); } + private boolean isIsolated(ServiceType serviceType, TenantId tenantId) { + return isolatedTenants.get(tenantId) != null && isolatedTenants.get(tenantId).contains(serviceType); + } + private void logServiceInfo(TransportProtos.ServiceInfo server) { TenantId tenantId = getSystemOrIsolatedTenantId(server); if (tenantId.isNullUid()) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java index 46864af79f..4eb4ca9a69 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java @@ -20,13 +20,14 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.gen.transport.TransportProtos; import java.util.List; +import java.util.Set; /** * Once application is ready or cluster topology changes, this Service will produce {@link PartitionChangeEvent} */ public interface PartitionService { - List getCurrentPartitions(ServiceType serviceType); + Set getCurrentPartitions(ServiceType serviceType); TopicPartitionInfo resolve(ServiceType serviceType, TenantId tenantId, EntityId entityId); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TopicPartitionInfo.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TopicPartitionInfo.java index 05c5c1586d..1164465ed7 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TopicPartitionInfo.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TopicPartitionInfo.java @@ -16,6 +16,7 @@ package org.thingsboard.server.queue.discovery; import lombok.Builder; +import lombok.Getter; import org.thingsboard.server.common.data.id.TenantId; import java.util.Objects; @@ -26,13 +27,17 @@ public class TopicPartitionInfo { private final String topic; private final TenantId tenantId; private final Integer partition; + @Getter private final String fullTopicName; + @Getter + private final boolean myPartition; @Builder - public TopicPartitionInfo(String topic, TenantId tenantId, Integer partition) { + public TopicPartitionInfo(String topic, TenantId tenantId, Integer partition, boolean myPartition) { this.topic = topic; this.tenantId = tenantId; this.partition = partition; + this.myPartition = myPartition; String tmp = topic; if (tenantId != null) { tmp += "." + tenantId.getId().toString(); @@ -40,7 +45,6 @@ public class TopicPartitionInfo { if (partition != null) { tmp += "." + partition; } - this.fullTopicName = tmp; } @@ -56,10 +60,6 @@ public class TopicPartitionInfo { return Optional.ofNullable(partition); } - public String getFullTopicName() { - return fullTopicName; - } - @Override public boolean equals(Object o) { if (this == o) return true; diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TopicPartitionInfoKey.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TopicPartitionInfoKey.java new file mode 100644 index 0000000000..4d02647fd0 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TopicPartitionInfoKey.java @@ -0,0 +1,43 @@ +/** + * Copyright © 2016-2020 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.discovery; + +import lombok.AllArgsConstructor; +import org.thingsboard.server.common.data.id.TenantId; + +import java.util.Objects; + +@AllArgsConstructor +public class TopicPartitionInfoKey { + private ServiceType serviceType; + private TenantId isolatedTenantId; + private int partition; + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + TopicPartitionInfoKey that = (TopicPartitionInfoKey) o; + return partition == that.partition && + serviceType == that.serviceType && + Objects.equals(isolatedTenantId, that.isolatedTenantId); + } + + @Override + public int hashCode() { + return Objects.hash(serviceType, isolatedTenantId, partition); + } +} 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 52086b3058..ec08d69be7 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 @@ -75,7 +75,7 @@ public class TBKafkaConsumerTemplate implements TbQueueCon @Override public void subscribe() { - partitions = Collections.singleton(new TopicPartitionInfo(topic, null, null)); + partitions = Collections.singleton(new TopicPartitionInfo(topic, null, null, true)); subscribed = false; }