sobychacko
commented
3 years ago @buyukim I tried to reproduce the issue with a sample project with the same dependencies as you have. I could not reproduce the issue, i.e, no exceptions on startup and I can invoke the metrics endpoint (see below). Here is that sample. Could you compare your setup with that sample and see where the mismatch is? If you still think that this is an issue, please try this sample and make it reproducible there.
Here is the metrics output when running the application:
curl http://localhost:8080/actuator/metrics/spring.cloud.stream.binder.kafka.offset | jq . <<<
% Total % Received % Xferd Average Speed Time Time Time Current
Dload Upload Total Spent Left Speed
100 342 0 342 0 0 4071 0 --:--:-- --:--:-- --:--:-- 4071
{
"name": "spring.cloud.stream.binder.kafka.offset",
"description": "Unconsumed messages for a particular group and topic",
"baseUnit": null,
"measurements": [
{
"statistic": "VALUE",
"value": 11
}
],
"availableTags": [
{
"tag": "topic",
"values": [
"hello4-in-0",
"hello5-in-0",
"hello1-in-0",
"hello3-in-0",
"hello2-in-0"
]
},
{
"tag": "group",
"values": [
"my-group-1"
]
}
]
}
buyukim
sabbyanandan
AlexandreCassagne
We recently upgraded to Spring Boot 2.4.1 (and tested 2.4.2 today) from Spring Boot 2.3.3 + Cloud Hoxton.SR6 and while our application starts fine and can process Kafka events, we are seeing the following exceptions in logs after startup that were not happening before:
2021-01-15 09:10:27.300 DEBUG 56330 --- [pool-5-thread-4] o.s.c.s.binder.kafka.KafkaBinderMetrics : Cannot generate metric for topic: digital.topic.a.dev java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:2452) at org.apache.kafka.clients.consumer.KafkaConsumer.acquireAndEnsureOpen(KafkaConsumer.java:2436) at org.apache.kafka.clients.consumer.KafkaConsumer.partitionsFor(KafkaConsumer.java:1954) at org.apache.kafka.clients.consumer.KafkaConsumer.partitionsFor(KafkaConsumer.java:1930) at org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics.findTotalTopicGroupLag(KafkaBinderMetrics.java:191) at org.springframework.cloud.stream.binder.kafka.KafkaBinderMetrics.lambda$computeUnconsumedMessagesRunnable$1(KafkaBinderMetrics.java:148) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) at java.base/java.lang.Thread.run(Thread.java:834)We see the above message for each topic, and the exceptions keep happening as the app runs.
We are using spring cloud stream v3.1.0, which is bundled with cloud 2020.0.0, and are also using:
implementation 'org.springframework.cloud:spring-cloud-starter-bootstrap'Here is the basic YAML config: