dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #5181 from smatvienko-tb/RuleChainActorMessageProcessor-onTellNext-NullPointerException-fix

Browse Source 
Rule chain actor message processor on tell next NullPointerException fix
This commit is contained in:
Igor Kulikov 2021-09-01 20:03:21 +03:00 committed by GitHub
commit 8f53d4a255
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 13 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActorMessageProcessor.java
View File

@ -250,10 +250,17 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
checkActive(msg); checkActive(msg);
EntityId entityId = msg.getOriginator(); EntityId entityId = msg.getOriginator();
TopicPartitionInfo tpi = systemContext.resolve(ServiceType.TB_RULE_ENGINE, msg.getQueueName(), tenantId, entityId); TopicPartitionInfo tpi = systemContext.resolve(ServiceType.TB_RULE_ENGINE, msg.getQueueName(), tenantId, entityId);
List<RuleNodeRelation> relations = nodeRoutes.get(originatorNodeId).stream()
List<RuleNodeRelation> ruleNodeRelations = nodeRoutes.get(originatorNodeId);
if (ruleNodeRelations == null) { // When unchecked, this will cause NullPointerException when rule node doesn't exist anymore
log.warn("[{}][{}][{}] No outbound relations (null). Probably rule node does not exist. Probably old message.", tenantId, entityId, msg.getId());
ruleNodeRelations = Collections.emptyList();
}
List<RuleNodeRelation> relationsByTypes = ruleNodeRelations.stream()
.filter(r -> contains(relationTypes, r.getType())) .filter(r -> contains(relationTypes, r.getType()))
.collect(Collectors.toList()); .collect(Collectors.toList());
int relationsCount = relations.size(); int relationsCount = relationsByTypes.size();
if (relationsCount == 0) { if (relationsCount == 0) {
log.trace("[{}][{}][{}] No outbound relations to process", tenantId, entityId, msg.getId()); log.trace("[{}][{}][{}] No outbound relations to process", tenantId, entityId, msg.getId());
if (relationTypes.contains(TbRelationTypes.FAILURE)) { if (relationTypes.contains(TbRelationTypes.FAILURE)) {
@ -268,14 +275,14 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
msg.getCallback().onSuccess(); msg.getCallback().onSuccess();
} }
} else if (relationsCount == 1) { } else if (relationsCount == 1) {
for (RuleNodeRelation relation : relations) { for (RuleNodeRelation relation : relationsByTypes) {
log.trace("[{}][{}][{}] Pushing message to single target: [{}]", tenantId, entityId, msg.getId(), relation.getOut()); log.trace("[{}][{}][{}] Pushing message to single target: [{}]", tenantId, entityId, msg.getId(), relation.getOut());
pushToTarget(tpi, msg, relation.getOut(), relation.getType()); pushToTarget(tpi, msg, relation.getOut(), relation.getType());
} }
} else { } else {
MultipleTbQueueTbMsgCallbackWrapper callbackWrapper = new MultipleTbQueueTbMsgCallbackWrapper(relationsCount, msg.getCallback()); MultipleTbQueueTbMsgCallbackWrapper callbackWrapper = new MultipleTbQueueTbMsgCallbackWrapper(relationsCount, msg.getCallback());
log.trace("[{}][{}][{}] Pushing message to multiple targets: [{}]", tenantId, entityId, msg.getId(), relations); log.trace("[{}][{}][{}] Pushing message to multiple targets: [{}]", tenantId, entityId, msg.getId(), relationsByTypes);
for (RuleNodeRelation relation : relations) { for (RuleNodeRelation relation : relationsByTypes) {
EntityId target = relation.getOut(); EntityId target = relation.getOut();
putToQueue(tpi, msg, callbackWrapper, target); putToQueue(tpi, msg, callbackWrapper, target);
} }
@ -283,6 +290,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
} catch (RuleNodeException rne) { } catch (RuleNodeException rne) {
msg.getCallback().onFailure(rne); msg.getCallback().onFailure(rne);
} catch (Exception e) { } catch (Exception e) {
log.warn("[" + tenantId + "]" + "[" + entityId + "]" + "[" + msg.getId() + "]" + " onTellNext failure", e);
msg.getCallback().onFailure(new RuleEngineException("onTellNext - " + e.getMessage())); msg.getCallback().onFailure(new RuleEngineException("onTellNext - " + e.getMessage()));
} }
} }