SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Polishing, renamed method

Browse Source 
Resolves #452
This commit is contained in:
Oleg Zhurakousky
2018-10-02 14:43:07 -04:00
parent 3b4bff959f
commit d44f2348e6
2 changed files with 6 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java
View File

@@ -112,15 +112,15 @@ public class KafkaBinderMetrics implements MeterBinder, ApplicationListener<Bind
String group = topicInfo.getValue().getConsumerGroup(); String group = topicInfo.getValue().getConsumerGroup();
Gauge.builder(METRIC_NAME, this, Gauge.builder(METRIC_NAME, this,
o -> calculateConsumerLagOnTopic(topic, group)) o -> computeUnconsumedMessages(topic, group))
.tag("group", group) .tag("group", group)
.tag("topic", topic) .tag("topic", topic)
.description("Consumer lag for a particular group and topic") .description("Unconsumed messages for a particular group and topic")
.register(registry); .register(registry);
} }
} }
private double calculateConsumerLagOnTopic(String topic, String group) { private long computeUnconsumedMessages(String topic, String group) {
ExecutorService exec = Executors.newSingleThreadExecutor(); ExecutorService exec = Executors.newSingleThreadExecutor();
Future<Long> future = exec.submit(() -> { Future<Long> future = exec.submit(() -> {
@@ -144,11 +144,9 @@ public class KafkaBinderMetrics implements MeterBinder, ApplicationListener<Bind
for (Map.Entry<TopicPartition, Long> endOffset : endOffsets.entrySet()) { for (Map.Entry<TopicPartition, Long> endOffset : endOffsets.entrySet()) {
OffsetAndMetadata current = metadataConsumer.committed(endOffset.getKey()); OffsetAndMetadata current = metadataConsumer.committed(endOffset.getKey());
lag += endOffset.getValue();
if (current != null) { if (current != null) {
lag += endOffset.getValue() - current.offset(); lag -= current.offset();
}
else {
lag += endOffset.getValue();
} }
} }
} }

2
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderActuatorTests.java
View File

@@ -91,7 +91,7 @@ public class KafkaBinderActuatorTests {
assertThat(this.meterRegistry.get("spring.cloud.stream.binder.kafka.offset") assertThat(this.meterRegistry.get("spring.cloud.stream.binder.kafka.offset")
.tag("group", TEST_CONSUMER_GROUP) .tag("group", TEST_CONSUMER_GROUP)
.tag("topic", Sink.INPUT) .tag("topic", Sink.INPUT)
.timeGauge().value()).isGreaterThan(0); .gauge().value()).isGreaterThan(0);
} }
@Test @Test