Volodymyr Babak
6 years ago
50 changed files with 1279 additions and 989 deletions
@ -0,0 +1,37 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2021 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 lombok.AllArgsConstructor; |
||||
|
import lombok.Getter; |
||||
|
import lombok.NoArgsConstructor; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
|
||||
|
@Component |
||||
|
@ConditionalOnProperty(prefix = "queue", value = "type", havingValue = "kafka") |
||||
|
@Getter |
||||
|
@AllArgsConstructor |
||||
|
@NoArgsConstructor |
||||
|
public class TbKafkaConsumerStatisticConfig { |
||||
|
@Value("${queue.kafka.consumer-stats.enabled:true}") |
||||
|
private Boolean enabled; |
||||
|
@Value("${queue.kafka.consumer-stats.print-interval-ms:60000}") |
||||
|
private Long printIntervalMs; |
||||
|
@Value("${queue.kafka.consumer-stats.kafka-response-timeout-ms:1000}") |
||||
|
private Long kafkaResponseTimeoutMs; |
||||
|
} |
||||
@ -0,0 +1,184 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2021 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 lombok.Builder; |
||||
|
import lombok.Data; |
||||
|
import lombok.RequiredArgsConstructor; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.apache.kafka.clients.admin.AdminClient; |
||||
|
import org.apache.kafka.clients.consumer.Consumer; |
||||
|
import org.apache.kafka.clients.consumer.ConsumerConfig; |
||||
|
import org.apache.kafka.clients.consumer.KafkaConsumer; |
||||
|
import org.apache.kafka.clients.consumer.OffsetAndMetadata; |
||||
|
import org.apache.kafka.common.TopicPartition; |
||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
import org.springframework.util.StringUtils; |
||||
|
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.msg.queue.ServiceType; |
||||
|
import org.thingsboard.server.queue.discovery.PartitionService; |
||||
|
|
||||
|
import javax.annotation.PostConstruct; |
||||
|
import javax.annotation.PreDestroy; |
||||
|
import java.time.Duration; |
||||
|
import java.util.ArrayList; |
||||
|
import java.util.List; |
||||
|
import java.util.Map; |
||||
|
import java.util.Properties; |
||||
|
import java.util.Set; |
||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||
|
import java.util.concurrent.Executors; |
||||
|
import java.util.concurrent.ScheduledExecutorService; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
@Slf4j |
||||
|
@Component |
||||
|
@RequiredArgsConstructor |
||||
|
@ConditionalOnProperty(prefix = "queue", value = "type", havingValue = "kafka") |
||||
|
public class TbKafkaConsumerStatsService { |
||||
|
private final Set<String> monitoredGroups = ConcurrentHashMap.newKeySet(); |
||||
|
|
||||
|
private final TbKafkaSettings kafkaSettings; |
||||
|
private final TbKafkaConsumerStatisticConfig statsConfig; |
||||
|
private final PartitionService partitionService; |
||||
|
|
||||
|
private AdminClient adminClient; |
||||
|
private Consumer<String, byte[]> consumer; |
||||
|
private ScheduledExecutorService statsPrintScheduler; |
||||
|
|
||||
|
@PostConstruct |
||||
|
public void init() { |
||||
|
if (!statsConfig.getEnabled()) { |
||||
|
return; |
||||
|
} |
||||
|
this.adminClient = AdminClient.create(kafkaSettings.toAdminProps()); |
||||
|
this.statsPrintScheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("kafka-consumer-stats")); |
||||
|
|
||||
|
Properties consumerProps = kafkaSettings.toConsumerProps(); |
||||
|
consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer-stats-loader-client"); |
||||
|
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-stats-loader-client-group"); |
||||
|
this.consumer = new KafkaConsumer<>(consumerProps); |
||||
|
|
||||
|
startLogScheduling(); |
||||
|
} |
||||
|
|
||||
|
private void startLogScheduling() { |
||||
|
Duration timeoutDuration = Duration.ofMillis(statsConfig.getKafkaResponseTimeoutMs()); |
||||
|
statsPrintScheduler.scheduleWithFixedDelay(() -> { |
||||
|
if (!isStatsPrintRequired()) { |
||||
|
return; |
||||
|
} |
||||
|
for (String groupId : monitoredGroups) { |
||||
|
try { |
||||
|
Map<TopicPartition, OffsetAndMetadata> groupOffsets = adminClient.listConsumerGroupOffsets(groupId).partitionsToOffsetAndMetadata() |
||||
|
.get(statsConfig.getKafkaResponseTimeoutMs(), TimeUnit.MILLISECONDS); |
||||
|
Map<TopicPartition, Long> endOffsets = consumer.endOffsets(groupOffsets.keySet(), timeoutDuration); |
||||
|
|
||||
|
List<GroupTopicStats> lagTopicsStats = getTopicsStatsWithLag(groupOffsets, endOffsets); |
||||
|
if (!lagTopicsStats.isEmpty()) { |
||||
|
StringBuilder builder = new StringBuilder(); |
||||
|
for (int i = 0; i < lagTopicsStats.size(); i++) { |
||||
|
builder.append(lagTopicsStats.get(i).toString()); |
||||
|
if (i != lagTopicsStats.size() - 1) { |
||||
|
builder.append(", "); |
||||
|
} |
||||
|
} |
||||
|
log.info("[{}] Topic partitions with lag: [{}].", groupId, builder.toString()); |
||||
|
} |
||||
|
} catch (Exception e) { |
||||
|
log.warn("[{}] Failed to get consumer group stats. Reason - {}.", groupId, e.getMessage()); |
||||
|
log.trace("Detailed error: ", e); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
}, statsConfig.getPrintIntervalMs(), statsConfig.getPrintIntervalMs(), TimeUnit.MILLISECONDS); |
||||
|
} |
||||
|
|
||||
|
private boolean isStatsPrintRequired() { |
||||
|
boolean isMyRuleEnginePartition = partitionService.resolve(ServiceType.TB_RULE_ENGINE, TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID).isMyPartition(); |
||||
|
boolean isMyCorePartition = partitionService.resolve(ServiceType.TB_CORE, TenantId.SYS_TENANT_ID, TenantId.SYS_TENANT_ID).isMyPartition(); |
||||
|
return log.isInfoEnabled() && (isMyRuleEnginePartition || isMyCorePartition); |
||||
|
} |
||||
|
|
||||
|
private List<GroupTopicStats> getTopicsStatsWithLag(Map<TopicPartition, OffsetAndMetadata> groupOffsets, Map<TopicPartition, Long> endOffsets) { |
||||
|
List<GroupTopicStats> consumerGroupStats = new ArrayList<>(); |
||||
|
for (TopicPartition topicPartition : groupOffsets.keySet()) { |
||||
|
long endOffset = endOffsets.get(topicPartition); |
||||
|
long committedOffset = groupOffsets.get(topicPartition).offset(); |
||||
|
long lag = endOffset - committedOffset; |
||||
|
if (lag != 0) { |
||||
|
GroupTopicStats groupTopicStats = GroupTopicStats.builder() |
||||
|
.topic(topicPartition.topic()) |
||||
|
.partition(topicPartition.partition()) |
||||
|
.committedOffset(committedOffset) |
||||
|
.endOffset(endOffset) |
||||
|
.lag(lag) |
||||
|
.build(); |
||||
|
consumerGroupStats.add(groupTopicStats); |
||||
|
} |
||||
|
} |
||||
|
return consumerGroupStats; |
||||
|
} |
||||
|
|
||||
|
public void registerClientGroup(String groupId) { |
||||
|
if (statsConfig.getEnabled() && !StringUtils.isEmpty(groupId)) { |
||||
|
monitoredGroups.add(groupId); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public void unregisterClientGroup(String groupId) { |
||||
|
if (statsConfig.getEnabled() && !StringUtils.isEmpty(groupId)) { |
||||
|
monitoredGroups.remove(groupId); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
@PreDestroy |
||||
|
public void destroy() { |
||||
|
if (statsPrintScheduler != null) { |
||||
|
statsPrintScheduler.shutdownNow(); |
||||
|
} |
||||
|
if (adminClient != null) { |
||||
|
adminClient.close(); |
||||
|
} |
||||
|
if (consumer != null) { |
||||
|
consumer.close(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
|
||||
|
@Builder |
||||
|
@Data |
||||
|
private static class GroupTopicStats { |
||||
|
private String topic; |
||||
|
private int partition; |
||||
|
private long committedOffset; |
||||
|
private long endOffset; |
||||
|
private long lag; |
||||
|
|
||||
|
@Override |
||||
|
public String toString() { |
||||
|
return "[" + |
||||
|
"topic=[" + topic + ']' + |
||||
|
", partition=[" + partition + "]" + |
||||
|
", committedOffset=[" + committedOffset + "]" + |
||||
|
", endOffset=[" + endOffset + "]" + |
||||
|
", lag=[" + lag + "]" + |
||||
|
"]"; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
Binary file not shown.
@ -0,0 +1,53 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2021 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. |
||||
|
*/ |
||||
|
|
||||
|
import com.google.gson.JsonParser; |
||||
|
import org.junit.Assert; |
||||
|
import org.junit.Test; |
||||
|
import org.junit.runner.RunWith; |
||||
|
import org.mockito.junit.MockitoJUnitRunner; |
||||
|
import org.thingsboard.server.common.transport.adaptor.JsonConverter; |
||||
|
|
||||
|
@RunWith(MockitoJUnitRunner.class) |
||||
|
public class JsonConverterTest { |
||||
|
|
||||
|
private static final JsonParser JSON_PARSER = new JsonParser(); |
||||
|
|
||||
|
@Test |
||||
|
public void testParseBigDecimalAsLong() { |
||||
|
var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 1E+1}"), 0L); |
||||
|
Assert.assertEquals(10L, result.get(0L).get(0).getLongValue().get().longValue()); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testParseBigDecimalAsDouble() { |
||||
|
var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 101E-1}"), 0L); |
||||
|
Assert.assertEquals(10.1, result.get(0L).get(0).getDoubleValue().get(), 0.0); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testParseAsDouble() { |
||||
|
var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 1.1}"), 0L); |
||||
|
Assert.assertEquals(1.1, result.get(0L).get(0).getDoubleValue().get(), 0.0); |
||||
|
} |
||||
|
|
||||
|
@Test |
||||
|
public void testParseAsLong() { |
||||
|
var result = JsonConverter.convertToTelemetry(JSON_PARSER.parse("{\"meterReadingDelta\": 11}"), 0L); |
||||
|
Assert.assertEquals(11L, result.get(0L).get(0).getLongValue().get().longValue()); |
||||
|
} |
||||
|
|
||||
|
} |
||||
Binary file not shown.
Loading…
Reference in new issue