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

Fix partition change event ignored

Browse Source
This commit is contained in:
ViacheslavKlimov 2024-02-09 14:15:24 +02:00
parent 25f1c82e8e
commit c1f16769f4
6 changed files with 44 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java
View File

@ -31,6 +31,7 @@ import org.thingsboard.server.actors.app.AppActor;
import org.thingsboard.server.actors.app.AppInitMsg;
import org.thingsboard.server.actors.stats.StatsActor;
import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.queue.discovery.TbApplicationEventListener;
import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
import org.thingsboard.server.queue.util.AfterStartUp;
@ -124,6 +125,11 @@ public class DefaultActorService extends TbApplicationEventListener<PartitionCha
this.appActor.tellWithHighPriority(new PartitionChangeMsg(event.getServiceType()));
}
@Override
protected boolean filterTbApplicationEvent(PartitionChangeEvent event) {
return event.getServiceType() == ServiceType.TB_RULE_ENGINE || event.getServiceType() == ServiceType.TB_CORE;
}
@PreDestroy
public void stopActorSystem() {
if (system != null) {

13
application/src/main/java/org/thingsboard/server/service/partition/AbstractPartitionBasedService.java
View File

@ -89,11 +89,14 @@ public abstract class AbstractPartitionBasedService<T extends EntityId> extends
*/
@Override
protected void onTbApplicationEvent(PartitionChangeEvent partitionChangeEvent) {
if (getServiceType().equals(partitionChangeEvent.getServiceType())) {
log.debug("onTbApplicationEvent, processing event: {}", partitionChangeEvent);
subscribeQueue.add(partitionChangeEvent.getPartitions());
scheduledExecutor.submit(this::pollInitStateFromDB);
}
log.debug("onTbApplicationEvent, processing event: {}", partitionChangeEvent);
subscribeQueue.add(partitionChangeEvent.getPartitions());
scheduledExecutor.submit(this::pollInitStateFromDB);
}
@Override
protected boolean filterTbApplicationEvent(PartitionChangeEvent event) {
return getServiceType().equals(event.getServiceType());
}
protected void pollInitStateFromDB() {

18
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java
View File

@ -220,16 +220,14 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService<ToCore
@Override
protected void onTbApplicationEvent(PartitionChangeEvent event) {
if (event.getServiceType().equals(getServiceType())) {
log.info("Subscribing to partitions: {}", event.getPartitions());
this.mainConsumer.subscribe(event.getPartitions());
this.usageStatsConsumer.subscribe(
event
.getPartitions()
.stream()
.map(tpi -> tpi.newByTopic(usageStatsConsumer.getTopic()))
.collect(Collectors.toSet()));
}
log.info("Subscribing to partitions: {}", event.getPartitions());
this.mainConsumer.subscribe(event.getPartitions());
this.usageStatsConsumer.subscribe(
event
.getPartitions()
.stream()
.map(tpi -> tpi.newByTopic(usageStatsConsumer.getTopic()))
.collect(Collectors.toSet()));
this.firmwareStatesConsumer.subscribe();
}

18
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java
View File

@ -110,16 +110,14 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService<
@Override
protected void onTbApplicationEvent(PartitionChangeEvent event) {
if (event.getServiceType().equals(getServiceType())) {
event.getPartitionsMap().forEach((queueKey, partitions) -> {
var consumer = consumers.get(queueKey);
if (consumer != null) {
consumer.update(partitions);
} else {
log.warn("Received invalid partition change event for {} that is not managed by this service", queueKey);
}
});
}
event.getPartitionsMap().forEach((queueKey, partitions) -> {
var consumer = consumers.get(queueKey);
if (consumer != null) {
consumer.update(partitions);
} else {
log.warn("Received invalid partition change event for {} that is not managed by this service", queueKey);
}
});
}
@AfterStartUp(order = AfterStartUp.REGULAR_SERVICE)

5
application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java
View File

@ -108,6 +108,11 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
launchMainConsumers();
}
@Override
protected boolean filterTbApplicationEvent(PartitionChangeEvent event) {
return event.getServiceType() == getServiceType();
}
protected abstract ServiceType getServiceType();
protected abstract void launchMainConsumers();

12
common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbApplicationEventListener.java
View File

@ -15,21 +15,27 @@
*/
package org.thingsboard.server.queue.discovery;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationListener;
import org.thingsboard.server.queue.discovery.event.TbApplicationEvent;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@Slf4j
public abstract class TbApplicationEventListener<T extends TbApplicationEvent> implements ApplicationListener<T> {
private int lastProcessedSequenceNumber = Integer.MIN_VALUE;
private final Lock seqNumberLock = new ReentrantLock();
private final Logger log = LoggerFactory.getLogger(getClass());
@Override
public void onApplicationEvent(T event) {
if (!filterTbApplicationEvent(event)) {
log.trace("Skipping event due to filter: {}", event);
return;
}
boolean validUpdate = false;
seqNumberLock.lock();
try {
@ -40,7 +46,7 @@ public abstract class TbApplicationEventListener<T extends TbApplicationEvent> i
} finally {
seqNumberLock.unlock();
}
if (validUpdate && filterTbApplicationEvent(event)) {
if (validUpdate) {
try {
onTbApplicationEvent(event);
} catch (Exception e) {