Paolo Cristiani
1 year ago
GitHub
|
Before Width: | Height: | Size: 25 KiB After Width: | Height: | Size: 25 KiB |
|
Before Width: | Height: | Size: 18 KiB After Width: | Height: | Size: 18 KiB |
|
Before Width: | Height: | Size: 83 KiB After Width: | Height: | Size: 83 KiB |
|
Before Width: | Height: | Size: 11 KiB After Width: | Height: | Size: 11 KiB |
|
Before Width: | Height: | Size: 19 KiB After Width: | Height: | Size: 19 KiB |
|
Before Width: | Height: | Size: 16 KiB After Width: | Height: | Size: 16 KiB |
|
Before Width: | Height: | Size: 111 KiB After Width: | Height: | Size: 111 KiB |
|
Before Width: | Height: | Size: 26 KiB After Width: | Height: | Size: 26 KiB |
|
Before Width: | Height: | Size: 25 KiB After Width: | Height: | Size: 25 KiB |
|
Before Width: | Height: | Size: 137 KiB After Width: | Height: | Size: 137 KiB |
|
Before Width: | Height: | Size: 17 KiB After Width: | Height: | Size: 18 KiB |
|
Before Width: | Height: | Size: 31 KiB After Width: | Height: | Size: 32 KiB |
|
Before Width: | Height: | Size: 24 KiB After Width: | Height: | Size: 24 KiB |
|
Before Width: | Height: | Size: 16 KiB After Width: | Height: | Size: 16 KiB |
|
Before Width: | Height: | Size: 15 KiB After Width: | Height: | Size: 15 KiB |
|
Before Width: | Height: | Size: 113 KiB After Width: | Height: | Size: 113 KiB |
|
Before Width: | Height: | Size: 84 KiB After Width: | Height: | Size: 84 KiB |
|
Before Width: | Height: | Size: 111 KiB After Width: | Height: | Size: 111 KiB |
|
Before Width: | Height: | Size: 118 KiB After Width: | Height: | Size: 118 KiB |
|
Before Width: | Height: | Size: 119 KiB After Width: | Height: | Size: 119 KiB |
|
Before Width: | Height: | Size: 112 KiB After Width: | Height: | Size: 112 KiB |
|
Before Width: | Height: | Size: 57 KiB After Width: | Height: | Size: 57 KiB |
|
Before Width: | Height: | Size: 54 KiB After Width: | Height: | Size: 54 KiB |
|
Before Width: | Height: | Size: 58 KiB After Width: | Height: | Size: 58 KiB |
|
Before Width: | Height: | Size: 72 KiB After Width: | Height: | Size: 72 KiB |
|
Before Width: | Height: | Size: 57 KiB After Width: | Height: | Size: 57 KiB |
|
Before Width: | Height: | Size: 54 KiB After Width: | Height: | Size: 54 KiB |
|
Before Width: | Height: | Size: 30 KiB After Width: | Height: | Size: 30 KiB |
|
Before Width: | Height: | Size: 42 KiB After Width: | Height: | Size: 42 KiB |
|
Before Width: | Height: | Size: 22 KiB After Width: | Height: | Size: 22 KiB |
|
Before Width: | Height: | Size: 22 KiB After Width: | Height: | Size: 22 KiB |
|
Before Width: | Height: | Size: 99 KiB After Width: | Height: | Size: 99 KiB |
|
Before Width: | Height: | Size: 46 KiB After Width: | Height: | Size: 46 KiB |
|
Before Width: | Height: | Size: 44 KiB After Width: | Height: | Size: 44 KiB |
|
Before Width: | Height: | Size: 46 KiB After Width: | Height: | Size: 46 KiB |
|
Before Width: | Height: | Size: 104 KiB After Width: | Height: | Size: 104 KiB |
|
Before Width: | Height: | Size: 117 KiB After Width: | Height: | Size: 117 KiB |
|
Before Width: | Height: | Size: 118 KiB After Width: | Height: | Size: 118 KiB |
|
Before Width: | Height: | Size: 120 KiB After Width: | Height: | Size: 120 KiB |
|
Before Width: | Height: | Size: 107 KiB After Width: | Height: | Size: 107 KiB |
|
Before Width: | Height: | Size: 119 KiB After Width: | Height: | Size: 119 KiB |
|
Before Width: | Height: | Size: 27 KiB After Width: | Height: | Size: 27 KiB |
|
Before Width: | Height: | Size: 22 KiB After Width: | Height: | Size: 22 KiB |
|
Before Width: | Height: | Size: 15 KiB After Width: | Height: | Size: 15 KiB |
|
Before Width: | Height: | Size: 100 KiB After Width: | Height: | Size: 100 KiB |
|
Before Width: | Height: | Size: 112 KiB After Width: | Height: | Size: 112 KiB |
|
Before Width: | Height: | Size: 17 KiB After Width: | Height: | Size: 17 KiB |
|
Before Width: | Height: | Size: 21 KiB After Width: | Height: | Size: 22 KiB |
@ -0,0 +1,150 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.service.edge.stats; |
||||
|
|
||||
|
import com.google.common.util.concurrent.FutureCallback; |
||||
|
import com.google.common.util.concurrent.Futures; |
||||
|
import com.google.common.util.concurrent.ListenableFuture; |
||||
|
import com.google.common.util.concurrent.MoreExecutors; |
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
||||
|
import org.springframework.scheduling.annotation.Scheduled; |
||||
|
import org.springframework.stereotype.Service; |
||||
|
import org.thingsboard.server.common.data.id.EdgeId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.kv.BasicTsKvEntry; |
||||
|
import org.thingsboard.server.common.data.kv.LongDataEntry; |
||||
|
import org.thingsboard.server.common.data.kv.TimeseriesSaveResult; |
||||
|
import org.thingsboard.server.common.data.kv.TsKvEntry; |
||||
|
import org.thingsboard.server.dao.edge.stats.EdgeStatsCounterService; |
||||
|
import org.thingsboard.server.dao.edge.stats.MsgCounters; |
||||
|
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
||||
|
import org.thingsboard.server.queue.discovery.TopicService; |
||||
|
import org.thingsboard.server.queue.kafka.KafkaAdmin; |
||||
|
import org.thingsboard.server.queue.util.TbCoreComponent; |
||||
|
|
||||
|
import java.util.Collections; |
||||
|
import java.util.HashMap; |
||||
|
import java.util.HashSet; |
||||
|
import java.util.List; |
||||
|
import java.util.Map; |
||||
|
import java.util.Optional; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
import java.util.stream.Collectors; |
||||
|
|
||||
|
import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_ADDED; |
||||
|
import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_LAG; |
||||
|
import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_PERMANENTLY_FAILED; |
||||
|
import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_PUSHED; |
||||
|
import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_TMP_FAILED; |
||||
|
|
||||
|
@TbCoreComponent |
||||
|
@ConditionalOnProperty(prefix = "edges.stats", name = "enabled", havingValue = "true", matchIfMissing = false) |
||||
|
@RequiredArgsConstructor |
||||
|
@Service |
||||
|
@Slf4j |
||||
|
public class EdgeStatsService { |
||||
|
|
||||
|
private final TimeseriesService tsService; |
||||
|
private final EdgeStatsCounterService statsCounterService; |
||||
|
private final TopicService topicService; |
||||
|
private final Optional<KafkaAdmin> kafkaAdmin; |
||||
|
|
||||
|
@Value("${edges.stats.ttl:30}") |
||||
|
private int edgesStatsTtlDays; |
||||
|
@Value("${edges.stats.report-interval-millis:600000}") |
||||
|
private long reportIntervalMillis; |
||||
|
|
||||
|
|
||||
|
@Scheduled( |
||||
|
fixedDelayString = "${edges.stats.report-interval-millis:600000}", |
||||
|
initialDelayString = "${edges.stats.report-interval-millis:600000}" |
||||
|
) |
||||
|
public void reportStats() { |
||||
|
log.debug("Reporting Edge communication stats..."); |
||||
|
long now = System.currentTimeMillis(); |
||||
|
long ts = now - (now % reportIntervalMillis); |
||||
|
|
||||
|
Map<EdgeId, MsgCounters> countersByEdge = statsCounterService.getCounterByEdge(); |
||||
|
Map<EdgeId, Long> lagByEdgeId = kafkaAdmin.isPresent() ? getEdgeLagByEdgeId(countersByEdge) : Collections.emptyMap(); |
||||
|
Map<EdgeId, MsgCounters> countersByEdgeSnapshot = new HashMap<>(statsCounterService.getCounterByEdge()); |
||||
|
countersByEdgeSnapshot.forEach((edgeId, counters) -> { |
||||
|
TenantId tenantId = counters.getTenantId(); |
||||
|
|
||||
|
if (kafkaAdmin.isPresent()) { |
||||
|
counters.getMsgsLag().set(lagByEdgeId.getOrDefault(edgeId, 0L)); |
||||
|
} |
||||
|
List<TsKvEntry> statsEntries = List.of( |
||||
|
entry(ts, DOWNLINK_MSGS_ADDED.getKey(), counters.getMsgsAdded().get()), |
||||
|
entry(ts, DOWNLINK_MSGS_PUSHED.getKey(), counters.getMsgsPushed().get()), |
||||
|
entry(ts, DOWNLINK_MSGS_PERMANENTLY_FAILED.getKey(), counters.getMsgsPermanentlyFailed().get()), |
||||
|
entry(ts, DOWNLINK_MSGS_TMP_FAILED.getKey(), counters.getMsgsTmpFailed().get()), |
||||
|
entry(ts, DOWNLINK_MSGS_LAG.getKey(), counters.getMsgsLag().get()) |
||||
|
); |
||||
|
|
||||
|
log.trace("Reported Edge communication stats: {} tenantId - {}, edgeId - {}", statsEntries, tenantId, edgeId); |
||||
|
saveTs(tenantId, edgeId, statsEntries); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
private Map<EdgeId, Long> getEdgeLagByEdgeId(Map<EdgeId, MsgCounters> countersByEdge) { |
||||
|
Map<EdgeId, String> edgeToTopicMap = countersByEdge.entrySet().stream() |
||||
|
.collect(Collectors.toMap( |
||||
|
Map.Entry::getKey, |
||||
|
e -> topicService.buildEdgeEventNotificationsTopicPartitionInfo(e.getValue().getTenantId(), e.getKey()).getTopic() |
||||
|
)); |
||||
|
|
||||
|
Map<String, Long> lagByTopic = kafkaAdmin.get().getTotalLagForGroupsBulk(new HashSet<>(edgeToTopicMap.values())); |
||||
|
|
||||
|
return edgeToTopicMap.entrySet().stream() |
||||
|
.collect(Collectors.toMap( |
||||
|
Map.Entry::getKey, |
||||
|
e -> lagByTopic.getOrDefault(e.getValue(), 0L) |
||||
|
)); |
||||
|
} |
||||
|
|
||||
|
private void saveTs(TenantId tenantId, EdgeId edgeId, List<TsKvEntry> statsEntries) { |
||||
|
try { |
||||
|
ListenableFuture<TimeseriesSaveResult> future = tsService.save( |
||||
|
tenantId, |
||||
|
edgeId, |
||||
|
statsEntries, |
||||
|
TimeUnit.DAYS.toSeconds(edgesStatsTtlDays) |
||||
|
); |
||||
|
|
||||
|
Futures.addCallback(future, new FutureCallback<>() { |
||||
|
@Override |
||||
|
public void onSuccess(TimeseriesSaveResult result) { |
||||
|
log.debug("Successfully saved edge time-series stats: {} for edge: {}", statsEntries, edgeId); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
public void onFailure(Throwable t) { |
||||
|
log.warn("Failed to save edge time-series stats for edge: {}", edgeId, t); |
||||
|
} |
||||
|
}, MoreExecutors.directExecutor()); |
||||
|
} finally { |
||||
|
statsCounterService.clear(edgeId); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private BasicTsKvEntry entry(long ts, String key, long value) { |
||||
|
return new BasicTsKvEntry(ts, new LongDataEntry(key, value)); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,173 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.service.edge; |
||||
|
|
||||
|
import com.google.common.util.concurrent.Futures; |
||||
|
import org.junit.jupiter.api.Assertions; |
||||
|
import org.junit.jupiter.api.BeforeEach; |
||||
|
import org.junit.jupiter.api.Test; |
||||
|
import org.junit.jupiter.api.extension.ExtendWith; |
||||
|
import org.mockito.ArgumentCaptor; |
||||
|
import org.mockito.Mock; |
||||
|
import org.mockito.junit.jupiter.MockitoExtension; |
||||
|
import org.springframework.test.util.ReflectionTestUtils; |
||||
|
import org.thingsboard.server.common.data.id.EdgeId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.kv.TimeseriesSaveResult; |
||||
|
import org.thingsboard.server.common.data.kv.TsKvEntry; |
||||
|
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
||||
|
import org.thingsboard.server.dao.edge.stats.EdgeStatsCounterService; |
||||
|
import org.thingsboard.server.dao.edge.stats.MsgCounters; |
||||
|
import org.thingsboard.server.dao.timeseries.TimeseriesService; |
||||
|
import org.thingsboard.server.queue.discovery.TopicService; |
||||
|
import org.thingsboard.server.queue.kafka.KafkaAdmin; |
||||
|
import org.thingsboard.server.service.edge.stats.EdgeStatsService; |
||||
|
|
||||
|
import java.util.List; |
||||
|
import java.util.Map; |
||||
|
import java.util.Optional; |
||||
|
import java.util.Set; |
||||
|
import java.util.UUID; |
||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||
|
import java.util.stream.Collectors; |
||||
|
|
||||
|
import static org.mockito.ArgumentMatchers.anyLong; |
||||
|
import static org.mockito.ArgumentMatchers.eq; |
||||
|
import static org.mockito.Mockito.mock; |
||||
|
import static org.mockito.Mockito.verify; |
||||
|
import static org.mockito.Mockito.when; |
||||
|
import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_ADDED; |
||||
|
import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_LAG; |
||||
|
import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_PERMANENTLY_FAILED; |
||||
|
import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_PUSHED; |
||||
|
import static org.thingsboard.server.dao.edge.stats.EdgeStatsKey.DOWNLINK_MSGS_TMP_FAILED; |
||||
|
|
||||
|
@ExtendWith(MockitoExtension.class) |
||||
|
public class EdgeStatsTest { |
||||
|
|
||||
|
@Mock |
||||
|
private TimeseriesService tsService; |
||||
|
@Mock |
||||
|
private TopicService topicService; |
||||
|
@Mock |
||||
|
private EdgeStatsCounterService statsCounterService; |
||||
|
private EdgeStatsService edgeStatsService; |
||||
|
|
||||
|
private final TenantId tenantId = TenantId.fromUUID(UUID.randomUUID()); |
||||
|
private final EdgeId edgeId = new EdgeId(UUID.randomUUID()); |
||||
|
|
||||
|
@BeforeEach |
||||
|
void setUp() { |
||||
|
edgeStatsService = new EdgeStatsService( |
||||
|
tsService, |
||||
|
statsCounterService, |
||||
|
topicService, |
||||
|
Optional.empty() |
||||
|
); |
||||
|
|
||||
|
ReflectionTestUtils.setField(edgeStatsService, "edgesStatsTtlDays", 30); |
||||
|
ReflectionTestUtils.setField(edgeStatsService, "reportIntervalMillis", 600_000L); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testReportStatsSavesTelemetry() { |
||||
|
// given
|
||||
|
MsgCounters counters = new MsgCounters(tenantId); |
||||
|
counters.getMsgsAdded().set(5); |
||||
|
counters.getMsgsPushed().set(3); |
||||
|
counters.getMsgsPermanentlyFailed().set(1); |
||||
|
counters.getMsgsTmpFailed().set(0); |
||||
|
counters.getMsgsLag().set(10); |
||||
|
|
||||
|
ConcurrentHashMap<EdgeId, MsgCounters> countersByEdge = new ConcurrentHashMap<>(); |
||||
|
countersByEdge.put(edgeId, counters); |
||||
|
|
||||
|
when(statsCounterService.getCounterByEdge()).thenReturn(countersByEdge); |
||||
|
|
||||
|
ArgumentCaptor<List<TsKvEntry>> captor = ArgumentCaptor.forClass(List.class); |
||||
|
when(tsService.save(eq(tenantId), eq(edgeId), captor.capture(), anyLong())) |
||||
|
.thenReturn(Futures.immediateFuture(mock(TimeseriesSaveResult.class))); |
||||
|
|
||||
|
// when
|
||||
|
edgeStatsService.reportStats(); |
||||
|
|
||||
|
// then
|
||||
|
List<TsKvEntry> entries = captor.getValue(); |
||||
|
Assertions.assertEquals(5, entries.size()); |
||||
|
|
||||
|
Map<String, Long> valuesByKey = entries.stream() |
||||
|
.collect(Collectors.toMap(TsKvEntry::getKey, e -> e.getLongValue().orElse(-1L))); |
||||
|
|
||||
|
Assertions.assertEquals(5L, valuesByKey.get(DOWNLINK_MSGS_ADDED.getKey()).longValue()); |
||||
|
Assertions.assertEquals(3L, valuesByKey.get(DOWNLINK_MSGS_PUSHED.getKey()).longValue()); |
||||
|
Assertions.assertEquals(1L, valuesByKey.get(DOWNLINK_MSGS_PERMANENTLY_FAILED.getKey()).longValue()); |
||||
|
Assertions.assertEquals(0L, valuesByKey.get(DOWNLINK_MSGS_TMP_FAILED.getKey()).longValue()); |
||||
|
Assertions.assertEquals(10L, valuesByKey.get(DOWNLINK_MSGS_LAG.getKey()).longValue()); |
||||
|
|
||||
|
|
||||
|
verify(statsCounterService).clear(edgeId); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testReportStatsWithKafkaLag() { |
||||
|
// given
|
||||
|
MsgCounters counters = new MsgCounters(tenantId); |
||||
|
counters.getMsgsAdded().set(2); |
||||
|
counters.getMsgsPushed().set(2); |
||||
|
counters.getMsgsPermanentlyFailed().set(0); |
||||
|
counters.getMsgsTmpFailed().set(1); |
||||
|
counters.getMsgsLag().set(0); |
||||
|
|
||||
|
ConcurrentHashMap<EdgeId, MsgCounters> countersByEdge = new ConcurrentHashMap<>(); |
||||
|
countersByEdge.put(edgeId, counters); |
||||
|
|
||||
|
// mocks
|
||||
|
when(statsCounterService.getCounterByEdge()).thenReturn(countersByEdge); |
||||
|
|
||||
|
String topic = "edge-topic"; |
||||
|
TopicPartitionInfo partitionInfo = new TopicPartitionInfo(topic, tenantId, 0, false); |
||||
|
when(topicService.buildEdgeEventNotificationsTopicPartitionInfo(tenantId, edgeId)).thenReturn(partitionInfo); |
||||
|
|
||||
|
KafkaAdmin kafkaAdmin = mock(KafkaAdmin.class); |
||||
|
when(kafkaAdmin.getTotalLagForGroupsBulk(Set.of(topic))) |
||||
|
.thenReturn(Map.of(topic, 15L)); |
||||
|
|
||||
|
ArgumentCaptor<List<TsKvEntry>> captor = ArgumentCaptor.forClass(List.class); |
||||
|
when(tsService.save(eq(tenantId), eq(edgeId), captor.capture(), anyLong())) |
||||
|
.thenReturn(Futures.immediateFuture(mock(TimeseriesSaveResult.class))); |
||||
|
|
||||
|
edgeStatsService = new EdgeStatsService( |
||||
|
tsService, |
||||
|
statsCounterService, |
||||
|
topicService, |
||||
|
Optional.of(kafkaAdmin) |
||||
|
); |
||||
|
ReflectionTestUtils.setField(edgeStatsService, "edgesStatsTtlDays", 30); |
||||
|
ReflectionTestUtils.setField(edgeStatsService, "reportIntervalMillis", 600_000L); |
||||
|
|
||||
|
// when
|
||||
|
edgeStatsService.reportStats(); |
||||
|
|
||||
|
// then
|
||||
|
List<TsKvEntry> entries = captor.getValue(); |
||||
|
Map<String, Long> valuesByKey = entries.stream() |
||||
|
.collect(Collectors.toMap(TsKvEntry::getKey, e -> e.getLongValue().orElse(-1L))); |
||||
|
|
||||
|
Assertions.assertEquals(15L, valuesByKey.get(DOWNLINK_MSGS_LAG.getKey())); |
||||
|
verify(statsCounterService).clear(edgeId); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,285 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.kafka; |
||||
|
|
||||
|
import jakarta.annotation.PreDestroy; |
||||
|
import lombok.SneakyThrows; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.apache.commons.lang3.concurrent.ConcurrentException; |
||||
|
import org.apache.commons.lang3.concurrent.LazyInitializer; |
||||
|
import org.apache.kafka.clients.admin.AdminClient; |
||||
|
import org.apache.kafka.clients.admin.ListOffsetsResult; |
||||
|
import org.apache.kafka.clients.admin.NewTopic; |
||||
|
import org.apache.kafka.clients.admin.OffsetSpec; |
||||
|
import org.apache.kafka.clients.admin.TopicDescription; |
||||
|
import org.apache.kafka.clients.consumer.OffsetAndMetadata; |
||||
|
import org.apache.kafka.common.TopicPartition; |
||||
|
import org.apache.kafka.common.errors.TopicExistsException; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
import org.springframework.context.annotation.Lazy; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.thingsboard.common.util.CachedValue; |
||||
|
import org.thingsboard.server.queue.util.TbKafkaComponent; |
||||
|
|
||||
|
import java.util.Collections; |
||||
|
import java.util.HashMap; |
||||
|
import java.util.List; |
||||
|
import java.util.Map; |
||||
|
import java.util.Optional; |
||||
|
import java.util.Set; |
||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||
|
import java.util.concurrent.ExecutionException; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
import java.util.concurrent.TimeoutException; |
||||
|
import java.util.stream.Collectors; |
||||
|
|
||||
|
@TbKafkaComponent |
||||
|
@Component |
||||
|
@Slf4j |
||||
|
public class KafkaAdmin { |
||||
|
|
||||
|
/* |
||||
|
* TODO: Get rid of per consumer/producer TbKafkaAdmin, |
||||
|
* use single KafkaAdmin instance that accepts topicConfigs. |
||||
|
* */ |
||||
|
|
||||
|
private final TbKafkaSettings settings; |
||||
|
|
||||
|
@Value("${queue.kafka.request.timeout.ms:30000}") |
||||
|
private int requestTimeoutMs; |
||||
|
@Value("${queue.kafka.topics_cache_ttl_ms:300000}") // 5 minutes by default
|
||||
|
private int topicsCacheTtlMs; |
||||
|
|
||||
|
private final LazyInitializer<AdminClient> adminClient; |
||||
|
private final CachedValue<Set<String>> topics; |
||||
|
|
||||
|
public KafkaAdmin(@Lazy TbKafkaSettings settings) { |
||||
|
this.settings = settings; |
||||
|
this.adminClient = LazyInitializer.<AdminClient>builder() |
||||
|
.setInitializer(() -> AdminClient.create(settings.toAdminProps())) |
||||
|
.get(); |
||||
|
this.topics = new CachedValue<>(() -> { |
||||
|
Set<String> topics = ConcurrentHashMap.newKeySet(); |
||||
|
topics.addAll(listTopics()); |
||||
|
return topics; |
||||
|
}, topicsCacheTtlMs); |
||||
|
} |
||||
|
|
||||
|
public void createTopicIfNotExists(String topic, Map<String, String> properties, boolean force) { |
||||
|
Set<String> topics = getTopics(); |
||||
|
if (!force && topics.contains(topic)) { |
||||
|
log.trace("Topic {} already present in cache", topic); |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
log.debug("Creating topic {} with properties {}", topic, properties); |
||||
|
String numPartitionsStr = properties.remove(TbKafkaTopicConfigs.NUM_PARTITIONS_SETTING); |
||||
|
int partitions = numPartitionsStr != null ? Integer.parseInt(numPartitionsStr) : 1; |
||||
|
NewTopic newTopic = new NewTopic(topic, partitions, settings.getReplicationFactor()).configs(properties); |
||||
|
|
||||
|
try { |
||||
|
getClient().createTopics(List.of(newTopic)).all().get(requestTimeoutMs, TimeUnit.MILLISECONDS); |
||||
|
topics.add(topic); |
||||
|
} catch (ExecutionException ee) { |
||||
|
log.trace("Failed to create topic {} with properties {}", topic, properties, ee); |
||||
|
if (ee.getCause() instanceof TopicExistsException) { |
||||
|
//do nothing
|
||||
|
} else { |
||||
|
log.warn("[{}] Failed to create topic", topic, ee); |
||||
|
throw new RuntimeException(ee); |
||||
|
} |
||||
|
} catch (Exception e) { |
||||
|
log.warn("[{}] Failed to create topic", topic, e); |
||||
|
throw new RuntimeException(e); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public void deleteTopic(String topic) { |
||||
|
log.debug("Deleting topic {}", topic); |
||||
|
try { |
||||
|
getClient().deleteTopics(List.of(topic)).all().get(requestTimeoutMs, TimeUnit.MILLISECONDS); |
||||
|
} catch (Exception e) { |
||||
|
log.error("Failed to delete kafka topic [{}].", topic, e); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
private Set<String> getTopics() { |
||||
|
return topics.get(); |
||||
|
} |
||||
|
|
||||
|
public Set<String> listTopics() { |
||||
|
try { |
||||
|
Set<String> topics = getClient().listTopics().names().get(requestTimeoutMs, TimeUnit.MILLISECONDS); |
||||
|
log.trace("Listed topics: {}", topics); |
||||
|
return topics; |
||||
|
} catch (Exception e) { |
||||
|
log.error("Failed to get all topics.", e); |
||||
|
return Collections.emptySet(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public Map<String, Long> getTotalLagForGroupsBulk(Set<String> groupIds) { |
||||
|
Map<String, Long> result = new HashMap<>(); |
||||
|
for (String groupId : groupIds) { |
||||
|
result.put(groupId, getTotalConsumerGroupLag(groupId)); |
||||
|
} |
||||
|
return result; |
||||
|
} |
||||
|
|
||||
|
public long getTotalConsumerGroupLag(String groupId) { |
||||
|
try { |
||||
|
Map<TopicPartition, OffsetAndMetadata> committedOffsets = getConsumerGroupOffsets(groupId); |
||||
|
if (committedOffsets.isEmpty()) { |
||||
|
return 0L; |
||||
|
} |
||||
|
|
||||
|
Map<TopicPartition, OffsetSpec> latestOffsetsSpec = committedOffsets.keySet().stream() |
||||
|
.collect(Collectors.toMap(tp -> tp, tp -> OffsetSpec.latest())); |
||||
|
|
||||
|
Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> endOffsets = |
||||
|
getClient().listOffsets(latestOffsetsSpec).all().get(requestTimeoutMs, TimeUnit.MILLISECONDS); |
||||
|
|
||||
|
return committedOffsets.entrySet().stream() |
||||
|
.mapToLong(entry -> { |
||||
|
TopicPartition tp = entry.getKey(); |
||||
|
long committed = entry.getValue().offset(); |
||||
|
long end = endOffsets.getOrDefault(tp, |
||||
|
new ListOffsetsResult.ListOffsetsResultInfo(0L, 0L, Optional.empty())).offset(); |
||||
|
return end - committed; |
||||
|
}).sum(); |
||||
|
|
||||
|
} catch (Exception e) { |
||||
|
log.error("Failed to get total lag for consumer group: {}", groupId, e); |
||||
|
return 0L; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@SneakyThrows |
||||
|
public Map<TopicPartition, OffsetAndMetadata> getConsumerGroupOffsets(String groupId) { |
||||
|
return getClient().listConsumerGroupOffsets(groupId).partitionsToOffsetAndMetadata().get(requestTimeoutMs, TimeUnit.MILLISECONDS); |
||||
|
} |
||||
|
|
||||
|
/** |
||||
|
* Sync offsets from a fat group to a single-partition group |
||||
|
* Migration back from single-partition consumer to a fat group is not supported |
||||
|
* TODO: The best possible approach to synchronize the offsets is to do the synchronization as a part of the save Queue parameters with stop all consumers |
||||
|
* */ |
||||
|
public void syncOffsets(String fatGroupId, String newGroupId, Integer partitionId) { |
||||
|
try { |
||||
|
log.info("syncOffsets [{}][{}][{}]", fatGroupId, newGroupId, partitionId); |
||||
|
if (partitionId == null) { |
||||
|
return; |
||||
|
} |
||||
|
syncOffsetsUnsafe(fatGroupId, newGroupId, "." + partitionId); |
||||
|
} catch (Exception e) { |
||||
|
log.warn("Failed to syncOffsets from {} to {} partitionId {}", fatGroupId, newGroupId, partitionId, e); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public void syncOffsetsUnsafe(String fatGroupId, String newGroupId, String topicSuffix) throws ExecutionException, InterruptedException, TimeoutException { |
||||
|
Map<TopicPartition, OffsetAndMetadata> oldOffsets = getConsumerGroupOffsets(fatGroupId); |
||||
|
if (oldOffsets.isEmpty()) { |
||||
|
return; |
||||
|
} |
||||
|
|
||||
|
for (var consumerOffset : oldOffsets.entrySet()) { |
||||
|
var tp = consumerOffset.getKey(); |
||||
|
if (!tp.topic().endsWith(topicSuffix)) { |
||||
|
continue; |
||||
|
} |
||||
|
var om = consumerOffset.getValue(); |
||||
|
Map<TopicPartition, OffsetAndMetadata> newOffsets = getConsumerGroupOffsets(newGroupId); |
||||
|
|
||||
|
var existingOffset = newOffsets.get(tp); |
||||
|
if (existingOffset == null) { |
||||
|
log.info("[{}] topic offset does not exists in the new node group {}, all found offsets {}", tp, newGroupId, newOffsets); |
||||
|
} else if (existingOffset.offset() >= om.offset()) { |
||||
|
log.info("[{}] topic offset {} >= than old node group offset {}", tp, existingOffset.offset(), om.offset()); |
||||
|
break; |
||||
|
} else { |
||||
|
log.info("[{}] SHOULD alter topic offset [{}] less than old node group offset [{}]", tp, existingOffset.offset(), om.offset()); |
||||
|
} |
||||
|
getClient().alterConsumerGroupOffsets(newGroupId, Map.of(tp, om)).all().get(requestTimeoutMs, TimeUnit.MILLISECONDS); |
||||
|
log.info("[{}] altered new consumer groupId {}", tp, newGroupId); |
||||
|
break; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public boolean isTopicEmpty(String topic) { |
||||
|
return areAllTopicsEmpty(Set.of(topic)); |
||||
|
} |
||||
|
|
||||
|
public boolean areAllTopicsEmpty(Set<String> topics) { |
||||
|
try { |
||||
|
List<String> existingTopics = getTopics().stream().filter(topics::contains).toList(); |
||||
|
if (existingTopics.isEmpty()) { |
||||
|
return true; |
||||
|
} |
||||
|
|
||||
|
List<TopicPartition> allPartitions = getClient().describeTopics(existingTopics).allTopicNames().get(requestTimeoutMs, TimeUnit.MILLISECONDS) |
||||
|
.entrySet().stream() |
||||
|
.flatMap(entry -> { |
||||
|
String topic = entry.getKey(); |
||||
|
TopicDescription topicDescription = entry.getValue(); |
||||
|
return topicDescription.partitions().stream().map(partitionInfo -> new TopicPartition(topic, partitionInfo.partition())); |
||||
|
}) |
||||
|
.toList(); |
||||
|
|
||||
|
Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> beginningOffsets = getClient().listOffsets(allPartitions.stream() |
||||
|
.collect(Collectors.toMap(partition -> partition, partition -> OffsetSpec.earliest()))).all().get(requestTimeoutMs, TimeUnit.MILLISECONDS); |
||||
|
Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> endOffsets = getClient().listOffsets(allPartitions.stream() |
||||
|
.collect(Collectors.toMap(partition -> partition, partition -> OffsetSpec.latest()))).all().get(requestTimeoutMs, TimeUnit.MILLISECONDS); |
||||
|
|
||||
|
for (TopicPartition partition : allPartitions) { |
||||
|
long beginningOffset = beginningOffsets.get(partition).offset(); |
||||
|
long endOffset = endOffsets.get(partition).offset(); |
||||
|
|
||||
|
if (beginningOffset != endOffset) { |
||||
|
log.debug("Partition [{}] of topic [{}] is not empty. Returning false.", partition.partition(), partition.topic()); |
||||
|
return false; |
||||
|
} |
||||
|
} |
||||
|
return true; |
||||
|
} catch (Exception e) { |
||||
|
log.error("Failed to check if topics [{}] empty.", topics, e); |
||||
|
return false; |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public void deleteConsumerGroup(String consumerGroupId) { |
||||
|
try { |
||||
|
getClient().deleteConsumerGroups(List.of(consumerGroupId)).all().get(requestTimeoutMs, TimeUnit.MILLISECONDS); |
||||
|
} catch (Exception e) { |
||||
|
log.warn("Failed to delete consumer group {}", consumerGroupId, e); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public AdminClient getClient() { |
||||
|
try { |
||||
|
return adminClient.get(); |
||||
|
} catch (ConcurrentException e) { |
||||
|
throw new RuntimeException("Failed to initialize Kafka admin client", e); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@PreDestroy |
||||
|
private void destroy() throws Exception { |
||||
|
if (adminClient.isInitialized()) { |
||||
|
adminClient.get().close(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,29 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.util; |
||||
|
|
||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
||||
|
|
||||
|
import java.lang.annotation.Inherited; |
||||
|
import java.lang.annotation.Retention; |
||||
|
import java.lang.annotation.RetentionPolicy; |
||||
|
import java.lang.annotation.Target; |
||||
|
|
||||
|
@Inherited |
||||
|
@Retention(RetentionPolicy.RUNTIME) |
||||
|
@Target({java.lang.annotation.ElementType.TYPE, java.lang.annotation.ElementType.METHOD}) |
||||
|
@ConditionalOnProperty(prefix = "queue", value = "type", havingValue = "kafka") |
||||
|
public @interface TbKafkaComponent {} |
||||
@ -0,0 +1,38 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.common.util; |
||||
|
|
||||
|
import com.github.benmanes.caffeine.cache.Caffeine; |
||||
|
import com.github.benmanes.caffeine.cache.LoadingCache; |
||||
|
|
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
import java.util.function.Supplier; |
||||
|
|
||||
|
public class CachedValue<V> { |
||||
|
|
||||
|
private final LoadingCache<Object, V> cache; |
||||
|
|
||||
|
public CachedValue(Supplier<V> supplier, long valueTtlMs) { |
||||
|
this.cache = Caffeine.newBuilder() |
||||
|
.expireAfterWrite(valueTtlMs, TimeUnit.MILLISECONDS) |
||||
|
.build(__ -> supplier.get()); |
||||
|
} |
||||
|
|
||||
|
public V get() { |
||||
|
return cache.get(this); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,57 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.dao.edge.stats; |
||||
|
|
||||
|
import lombok.Getter; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
||||
|
import org.springframework.stereotype.Service; |
||||
|
import org.thingsboard.server.common.data.id.EdgeId; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
|
||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||
|
|
||||
|
@ConditionalOnProperty(prefix = "edges.stats", name = "enabled", havingValue = "true", matchIfMissing = false) |
||||
|
@Service |
||||
|
@Slf4j |
||||
|
@Getter |
||||
|
public class EdgeStatsCounterService { |
||||
|
|
||||
|
private final ConcurrentHashMap<EdgeId, MsgCounters> counterByEdge = new ConcurrentHashMap<>(); |
||||
|
|
||||
|
public void recordEvent(EdgeStatsKey type, TenantId tenantId, EdgeId edgeId, long value) { |
||||
|
MsgCounters counters = getOrCreateCounters(tenantId, edgeId); |
||||
|
switch (type) { |
||||
|
case DOWNLINK_MSGS_ADDED -> counters.getMsgsAdded().addAndGet(value); |
||||
|
case DOWNLINK_MSGS_PUSHED -> counters.getMsgsPushed().addAndGet(value); |
||||
|
case DOWNLINK_MSGS_PERMANENTLY_FAILED -> counters.getMsgsPermanentlyFailed().addAndGet(value); |
||||
|
case DOWNLINK_MSGS_TMP_FAILED -> counters.getMsgsTmpFailed().addAndGet(value); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public void setDownlinkMsgsLag(TenantId tenantId, EdgeId edgeId, long value) { |
||||
|
getOrCreateCounters(tenantId, edgeId).getMsgsLag().set(value); |
||||
|
} |
||||
|
|
||||
|
public void clear(EdgeId edgeId) { |
||||
|
counterByEdge.remove(edgeId); |
||||
|
} |
||||
|
|
||||
|
public MsgCounters getOrCreateCounters(TenantId tenantId, EdgeId edgeId) { |
||||
|
return counterByEdge.computeIfAbsent(edgeId, id -> new MsgCounters(tenantId)); |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,34 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.dao.edge.stats; |
||||
|
|
||||
|
import lombok.Getter; |
||||
|
|
||||
|
@Getter |
||||
|
public enum EdgeStatsKey { |
||||
|
DOWNLINK_MSGS_ADDED("downlinkMsgsAdded"), |
||||
|
DOWNLINK_MSGS_PUSHED("downlinkMsgsPushed"), |
||||
|
DOWNLINK_MSGS_PERMANENTLY_FAILED("downlinkMsgsPermanentlyFailed"), |
||||
|
DOWNLINK_MSGS_TMP_FAILED("downlinkMsgsTmpFailed"), |
||||
|
DOWNLINK_MSGS_LAG("downlinkMsgsLag"); |
||||
|
|
||||
|
private final String key; |
||||
|
|
||||
|
EdgeStatsKey(String key) { |
||||
|
this.key = key; |
||||
|
} |
||||
|
|
||||
|
} |
||||
@ -0,0 +1,41 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2025 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.dao.edge.stats; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
|
||||
|
import java.util.concurrent.atomic.AtomicLong; |
||||
|
|
||||
|
@Data |
||||
|
public class MsgCounters { |
||||
|
|
||||
|
private final TenantId tenantId; |
||||
|
private final AtomicLong msgsAdded = new AtomicLong(); |
||||
|
private final AtomicLong msgsPushed = new AtomicLong(); |
||||
|
private final AtomicLong msgsPermanentlyFailed = new AtomicLong(); |
||||
|
private final AtomicLong msgsTmpFailed = new AtomicLong(); |
||||
|
private final AtomicLong msgsLag = new AtomicLong(); |
||||
|
|
||||
|
public void clear() { |
||||
|
msgsAdded.set(0); |
||||
|
msgsPushed.set(0); |
||||
|
msgsPermanentlyFailed.set(0); |
||||
|
msgsTmpFailed.set(0); |
||||
|
msgsLag.set(0); |
||||
|
} |
||||
|
|
||||
|
} |
||||