4,624 questions
1
vote
0
answers
26
views
IdempotencyBreachExceptionHandling in service used in kafka listener
I’m encountering an issue with possible idempotencyBreachException in a service used by Kafka listener.
I’m not entirely sure how to handle this situation correctly. I have a lockExecutor that locks a ...
- 11
Best practices
0
votes
0
replies
27
views
Throw message on DLT/DLQ from GlobalKTable using Kafka Streams
I am reading data from a Kafka topic using Kafka Streams and more specifically GlobalKTable to populate my store. In the case of corrupt data that can not successfully be parsed I wish to throw a ...
- 23
1
vote
0
answers
52
views
Ack messages with StreamBridge::send and Kafka
Context. I have to use StreamBridge::send() with Kafka. This is a hard, external requirement. The call itself looks like streambridge.send(binding, message). All good.
To capture the non-immediate ...
- 183
1
vote
0
answers
57
views
Could not instantiate KafkaProperties
I'm migrating from Spring Boot 2 to Spring Boot3, i'm NOT migrating to Sprinc Cloud 4 but i will stay to SpringCloud 3 (i know Spring Boot 3 is mandatory if you want to use Spring Cloud 4 but not the ...
Best practices
0
votes
0
replies
52
views
Spring Kafka consumer reprocessing messages after switching from AckMode.MANUAL_IMMEDIATE to MANUAL
I have two Spring Kafka consumers that process batches of messages with manual acknowledgements. This setup has been running correctly for a long time with AckMode.MANUAL_IMMEDIATE.
Recently, I ...
- 3
1
vote
1
answer
131
views
How to Pass org.apache.kafka.sasl.oauthbearer.allowed.urls JVM Argument to Spring Boot Application
I have a Spring Boot Application that is using Kafka. When I run the code, I was getting the error:
Caused by: org.apache.kafka.common.KafkaException: org.apache.kafka.common.config.ConfigException: ...
- 1,669
2
votes
1
answer
72
views
How to log only the root cause exception from a Spring Kafka @KafkaListener?
I am using Spring Kafka, and when an exception is thrown inside my @KafkaListener method, it gets wrapped by Spring Kafka exceptions. This makes my logs very difficult to read and analyze.
The Problem
...
- 21
0
votes
1
answer
42
views
Spring kafka exhuast recoverer for non-blocking retries with @RetryableTopic
I want to point my existing dlq topic as a dlt to the retry topic configuration so that another application can continue consuming from that dlq topic but on my producer application for the dlq which ...
0
votes
0
answers
63
views
springboot integration Kafka @RetryableTopic is not effective
describe
springboot integration Kafka @RetryableTopic is not effective,I cannot control retry count; it is retrying continuously.There are two types of error logs continuously printed in the console: ...
- 11
0
votes
0
answers
34
views
Calculate the correct record lag between Azure EventHub and Spring Kafka
We have spring kafka consuming events from azure event hub. Is there any way to measure the lag in the consumption of messages. Actuator metrics seems to be in-memory and if there is any issue with ...
- 239
0
votes
1
answer
96
views
Spring boot Kafka Message Listener stopped automatically [Consumer stopped]
In a spring boot app, I have a message listener (@KafkaListener) with a concurrency of 28 listening from a topic which has 28 partitions. I have default configurations for all consumer properties. ...
- 245
0
votes
1
answer
33
views
HW rollback after leader election?
Assume the Leader’s LEO = 1000 and HW = 900.
FollowerA has just restarted, with LEO = 100. Since it keeps fetching in time, it enters the ISR (even though it is far behind).
ISR = {Leader, FollowerA} →...
- 195
1
vote
1
answer
237
views
How to enforce delay between Kafka retry topic consumers (e.g., 5m, 15m, 30m) in Spring Boot?
I'm building a Kafka-based retry mechanism in a Spring Boot application. The setup involves:
A primary topic for normal processing.
Three retry topics (topic-retry-5m, topic-retry-15m, topic-retry-...
0
votes
0
answers
69
views
Spring Kafka @RetryableTopic does not identify Manually Created Retry Topics and DLT
I am using RetryableTopic to enable kafka retry mechanism in a consumer.
This is the sample code snippet I have tried.
@RetryableTopic(
attempts = "3",
backoff = @Backoff(...
- 57
0
votes
2
answers
123
views
Kafka Manual commit scenario Springboot
We are using Spring Boot Kafka listener to consume messages from Kafka. Currently, we have the auto commit offset enabled and so the offset commits happen every 5 seconds (default auto commit interval)...
- 325
1
vote
0
answers
67
views
creating dynamic groupId for kafkaListener in java
i am trying to create group.id dynamically as shown below. the purpose of this is to consume message for all running instance of my application. e.g there are 15 instances running with different ...
- 51
0
votes
1
answer
40
views
How to customise RetryTopicNamesProvider with groupId or similar properties
I've finally broken and thought I'd see if anyone could help, considering some similar conversations elsewhere; which don't have example answers.
Due to a custom implementation of retry/error handling ...
- 128
-1
votes
1
answer
80
views
How could I catch all exceptions while starting or running a Spring Cloud Stream app?
This application is for DBAs and they couldn't modify the source codes. So I must consider all situations in the application from the start of the program to the end of the program and report these ...
- 1
1
vote
0
answers
28
views
When migrating Kafka Streams to Spring Cloud Stream how to resolve InconsistentGroupProtocolException
The prior application is on kafka-streams (KStream) and has been rewritten to use Spring Cloud Stream. Both apps are on kafka-clients-3.1.2.
When starting the upgraded application it gives:
java.lang....
- 1,899
0
votes
1
answer
138
views
How does spring kafka batch listener work?
I have a little question about spring kafka batch listener work. As I know, if we set this property
factory.setBatchListener(true);
Then we will be able to consume message batches. But will we be ...
- 366
0
votes
0
answers
73
views
Kafka Config missing in spring.factories file
Tried searching for kafkaAutoConfiguration in the spring.factories file inside spring-boot-auto-configure:3.5.3 but its not there. So, how is spring-boot autoconfiguring my kafka properties in the ...
- 1
1
vote
1
answer
85
views
org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before successfully committing offsets with kafkaTransactionManager
I am using kafka @KafkaListener with @Transactional("kafkaTransactionManager").My flow will be something read/process/commit using transaction consumer.
public void listen(List<...
- 47
0
votes
0
answers
27
views
How to check whether Kafka consumer retried at least once with use of @RetryableTopic?
I have below Kafka consumer. I want to do some work before and after myServiceEventHandler.handleRequestEvent(record.value()) but only once. cannot execute on retries. I do not mind of changing the ...
- 1,070
0
votes
0
answers
45
views
How to update Kafka Admin Client SSL properties using SslBundle
I am trying to publish messages to a Kafka cluster that is SSL protected.
I have correct certificates on the application side. I checked I can consume from this kafka cluster using those. Also, the ...
- 21
0
votes
0
answers
15
views
Spring ConsumerRecordRecoverer with infinite execution
I have a Spring application which listens messages from Kafka topic.
DefaultErrorHandler is used as an error handler with an implementation of ConsumerRecordRecoverer and fixed back off.
I need for ...
- 87