@ -17,6 +17,7 @@ package org.thingsboard.server.service.queue.processing;
import lombok.extern.slf4j.Slf4j ;
import org.springframework.stereotype.Component ;
import org.springframework.util.CollectionUtils ;
import org.thingsboard.server.common.msg.TbMsg ;
import org.thingsboard.server.common.msg.queue.TbMsgCallback ;
import org.thingsboard.server.gen.transport.TransportProtos ;
@ -77,6 +78,7 @@ public class TbRuleEngineProcessingStrategyFactory {
@Override
public TbRuleEngineProcessingDecision analyze ( TbRuleEngineProcessingResult result ) {
if ( result . isSuccess ( ) ) {
log . debug ( "[{}] The result of the msg pack processing is successful, going to proceed with processing of the following msgs" , queueName ) ;
return new TbRuleEngineProcessingDecision ( true , null ) ;
} else {
if ( retryCount = = 0 ) {
@ -91,6 +93,7 @@ public class TbRuleEngineProcessingStrategyFactory {
log . debug ( "[{}] Skip reprocess of the rule engine pack due to max allowed failure percentage" , queueName ) ;
return new TbRuleEngineProcessingDecision ( true , null ) ;
} else {
log . debug ( "[{}] The result of msg pack processing is unsuccessful, checking unprocessed msgs and going to reprocess them" , queueName ) ;
ConcurrentMap < UUID , TbProtoQueueMsg < TransportProtos . ToRuleEngineMsg > > toReprocess = new ConcurrentHashMap < > ( initialTotalCount ) ;
if ( retryFailed ) {
result . getFailedMap ( ) . forEach ( toReprocess : : put ) ;
@ -101,6 +104,15 @@ public class TbRuleEngineProcessingStrategyFactory {
if ( retrySuccessful ) {
result . getSuccessMap ( ) . forEach ( toReprocess : : put ) ;
}
if ( CollectionUtils . isEmpty ( toReprocess ) ) {
log . debug ( "[{}] Stopping the reprocessing logic due to reprocessing map is empty" , queueName ) ;
if ( retryFailed ) {
log . debug ( "[{}] Probably there are timedOut msgs, but the strategy is not set to reprocess them. Reprocess of the failed msgs is set" , queueName ) ;
} else if ( retryTimeout ) {
log . debug ( "[{}] Probably there are failed msgs, but the strategy is not set to reprocess them. Reprocess of the timedOut msgs is set" , queueName ) ;
}
return new TbRuleEngineProcessingDecision ( true , null ) ;
}
log . debug ( "[{}] Going to reprocess {} messages" , queueName , toReprocess . size ( ) ) ;
if ( log . isTraceEnabled ( ) ) {
toReprocess . forEach ( ( id , msg ) - > log . trace ( "Going to reprocess [{}]: {}" , id , TbMsg . fromBytes ( result . getQueueName ( ) , msg . getValue ( ) . getTbMsg ( ) . toByteArray ( ) , TbMsgCallback . EMPTY ) ) ) ;