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

Improvement

Browse Source
This commit is contained in:
Andrii Landiak 2024-11-29 15:45:25 +02:00
parent 60790d2b38
commit 1099ca2aa2
1 changed files with 6 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
application/src/main/java/org/thingsboard/server/service/ttl/KafkaEdgeTopicsCleanUpService.java
View File

@ -43,7 +43,6 @@ import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Set; import java.util.Set;
import java.util.UUID; import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAST_CONNECT_TIME; import static org.thingsboard.server.service.state.DefaultDeviceStateService.LAST_CONNECT_TIME;
@ -60,9 +59,7 @@ public class KafkaEdgeTopicsCleanUpService extends AbstractCleanUpService {
private final TenantService tenantService; private final TenantService tenantService;
private final EdgeService edgeService; private final EdgeService edgeService;
private final AttributesService attributesService; private final AttributesService attributesService;
private final TbKafkaAdmin kafkaAdmin;
private final TbKafkaSettings kafkaSettings;
private final TbKafkaTopicConfigs kafkaTopicConfigs;
@Value("${sql.ttl.edge_events.edge_events_ttl:2628000}") @Value("${sql.ttl.edge_events.edge_events_ttl:2628000}")
private long ttlSeconds; private long ttlSeconds;
@ -75,8 +72,7 @@ public class KafkaEdgeTopicsCleanUpService extends AbstractCleanUpService {
this.tenantService = tenantService; this.tenantService = tenantService;
this.edgeService = edgeService; this.edgeService = edgeService;
this.attributesService = attributesService; this.attributesService = attributesService;
this.kafkaSettings = kafkaSettings; this.kafkaAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeEventConfigs());
this.kafkaTopicConfigs = kafkaTopicConfigs;
} }
@Scheduled(initialDelayString = "#{T(org.apache.commons.lang3.RandomUtils).nextLong(0, ${sql.ttl.edge_events.execution_interval_ms})}", fixedDelayString = "${sql.ttl.edge_events.execution_interval_ms}") @Scheduled(initialDelayString = "#{T(org.apache.commons.lang3.RandomUtils).nextLong(0, ${sql.ttl.edge_events.execution_interval_ms})}", fixedDelayString = "${sql.ttl.edge_events.execution_interval_ms}")
@ -85,7 +81,6 @@ public class KafkaEdgeTopicsCleanUpService extends AbstractCleanUpService {
return; return;
} }
TbKafkaAdmin kafkaAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getEdgeEventConfigs());
Set<String> topics = kafkaAdmin.getAllTopics(); Set<String> topics = kafkaAdmin.getAllTopics();
if (topics == null || topics.isEmpty()) { if (topics == null || topics.isEmpty()) {
return; return;
@ -103,10 +98,10 @@ public class KafkaEdgeTopicsCleanUpService extends AbstractCleanUpService {
long currentTimeMillis = System.currentTimeMillis(); long currentTimeMillis = System.currentTimeMillis();
long ttlMillis = TimeUnit.SECONDS.toMillis(ttlSeconds); long ttlMillis = TimeUnit.SECONDS.toMillis(ttlSeconds);
tenantEdgeMap.forEach((tenantId, edgeIds) -> processTenantCleanUp(kafkaAdmin, tenantId, edgeIds, ttlMillis, currentTimeMillis)); tenantEdgeMap.forEach((tenantId, edgeIds) -> processTenantCleanUp(tenantId, edgeIds, ttlMillis, currentTimeMillis));
} }
private void processTenantCleanUp(TbKafkaAdmin kafkaAdmin, TenantId tenantId, List<EdgeId> edgeIds, long ttlMillis, long currentTimeMillis) { private void processTenantCleanUp(TenantId tenantId, List<EdgeId> edgeIds, long ttlMillis, long currentTimeMillis) {
boolean tenantExists = tenantService.tenantExists(tenantId); boolean tenantExists = tenantService.tenantExists(tenantId);
if (tenantExists) { if (tenantExists) {
for (EdgeId edgeId : edgeIds) { for (EdgeId edgeId : edgeIds) {
@ -129,8 +124,8 @@ public class KafkaEdgeTopicsCleanUpService extends AbstractCleanUpService {
log.info("[{}] Removed topic {} for deleted edge {}", tenantId, topic, edgeId); log.info("[{}] Removed topic {} for deleted edge {}", tenantId, topic, edgeId);
} }
}); });
} catch (InterruptedException | ExecutionException e) { } catch (Exception e) {
log.error("[{}] Failed to delete topic", tenantId); log.error("[{}] Failed to delete topic for edge {}", tenantId, edgeId, e);
} }
} }
} else { } else {