Questions tagged [kafka-consumer-api]
Use for questions related to the Apache Kafka consumer API
4,278
questions
0
votes
0
answers
8
views
Do I make hundreds of Kafka brokers, or do I make a single super broker?
I am working on setting up kafka and debezium to connect to a bunch of MySQL databases, for sake of conversation around 300
This is my first time working with Kafka so I was wondering what is the ...
0
votes
0
answers
6
views
Kafka Producer - multiple topics
I have to make sure that based on a request from a consumer, a new topic is created and a new producer is assigned to it who has to take the name of the topic and use them to add information to the ...
0
votes
0
answers
23
views
Pausing offset and commit in Kafka
I am new to Kafka and trying to handle a case where I want the offset and commit to pause and want it to get increase at some specific point(when the message is read successfully).
This is my consumer:...
0
votes
0
answers
12
views
Issue with Kafka Connectors
We have been utilizing Kafka Connectors to facilitate data transfer between databases. However, we have encountered an issue with the new connectors where only partial data is being transferred. The ...
0
votes
1
answer
24
views
Resuming Kafka Stream from Custom Offset
I am trying to find some way to resume the Kafka Streams from manual offset.
from finding on the Internet I didn't find any answer that says it can.
is there any way? or have to go back to low-level ...
0
votes
0
answers
17
views
Kafka Consumers consumes processed messages
We are experiencing issue with Kafka consuming old messages after commiting latest offset. Here is logs:
2023-11-17_19:26:11.142 Resetting offset for partition <TOPIC_NAME>-2 to offset 3812355.
...
-1
votes
0
answers
27
views
confluent-kafka-python consumer unable to read messages from Kafka Topic
I am stuck with an issue related to Kafka consumer using confluent-kafka's python library.
CONTEXT
I have a Kafka topic on Azure Ubuntu VM that I need to consume.
SCENARIO
Consumer Script (...
0
votes
0
answers
18
views
How to send headers from Pentaho DI to Kafka?
I have a need to pass Kafka headers (key-value pairs that can be attached to Kafka messages to provide metadata and additional information about the message) to send a JSON message to a topic.
Please ...
0
votes
1
answer
44
views
kafkaconsumerfactory vs kafkalistenercontainerfactory
What is the relationship / difference between consumer and container in Spring Kafka?
I'm asking this question because Spring provides kafkaconsumerfactory and kafkalistenercontainerfactory and I'm ...
0
votes
0
answers
42
views
Java Kafka-client doesn’t recover automatically after timeout
I’m working in a project that uses kafka for communication. There is a business communication associated with a payment, so I configured kafka-client with exactly-once semantic in that topic.
While ...
0
votes
0
answers
19
views
Duplicate message consumption on Apache Camel Kafka
I am using the Kafka Consumer Component with Apache Camel 4.1.0 and Spring-Boot 3.1.5. To test message processing, I use a Kafka Testcontainer (one broker, one partition) and publish one test message ...
-1
votes
1
answer
34
views
How does java kafka consumer class handle partitioning?
I basically have a system with multiple tasks of a spring boot application using the Kafka consumer class running with the same consumer group id. They are all connected to the same topic. I didn’t ...
0
votes
0
answers
54
views
.NET Core Consumer Kafka error deadlock SQL Server
I am new to .NET. I have micro services using .NET Core. This service is used to become a Kafka consumer. In the consumer function there is a function to insert into the SQL Server database. This ...
0
votes
0
answers
20
views
Handling Kafka Consumer Processing Timeouts
I have a C++ service that is processing Kafka messages, like this:
while (!shutdown)
{
auto msg = KafkaConsumer.poll();
ProcessMessage(msg);
}
Calling poll does two things. It commits the ...
0
votes
1
answer
32
views
Spring kafka infinite retries without retryabletopic annotation
We are using spring Kafka 3.x and using DefaultErrorHandler.
In this handler, we check if the exception is retriable then do retries till Integer.maxValue else put a message in DLT.
Whole idea is to ...
0
votes
0
answers
32
views
Kafka Consumer partition assignment delayed
When using the Kafka consumer subscribe(Collection<String> topics, ConsumerRebalanceListener listener) API, I'm using the onPartitionsAssigned callback in ConsumerRebalanceListener to seek newly ...
0
votes
0
answers
19
views
Reactor Kafka is causing the interruption of consumer messages from random partitions
Seems like a persistent issue with my reactive Kafka consumer—it intermittently halts message consumption from different partitions.
A restart temporarily fixes the problem, but it resurfaces after ...
0
votes
2
answers
42
views
How can I get when the last time that consumer commit an offset?
I'm trying to search when that a consumer group last commit an offset. I'm not trying to know the position of the offset, but rather the time that the consumer group last commit the offset.
I've tried ...
0
votes
1
answer
40
views
How can I read always the newest (last) message from a Kafka Topic that has more then one partition (In my case three partitions)?
In my .net C# project (with Confluent Kafka library) currently I am using following code to read the newest message from a Kafka topic. But with this code I can read the newest message from the ...
0
votes
0
answers
25
views
Kafka dead letter queue pattern
I am working on a Kafka project in which one of the requirement is to implement a dead letter queue pattern.
The producer sends a record in a topic based on an Avro schema and is working fine and ...
0
votes
1
answer
63
views
Flink kafka consumer group.id not works
I use apache fink with FlinkKafkaConsumer, I have 2 consumer read 1 topic(1partition). At one time, both consumers read the same event. I expect consumer1 receive event1, consumer2 can not receive ...
0
votes
0
answers
15
views
kafka-python not receiving message on consumer side [duplicate]
I have my kafka running on my docker.
When I create a topic from CLI and start producer and consumer on CLI then kafka is able to transfer messages. but when I do same thing with python, then I don't ...
0
votes
0
answers
17
views
How do consumer behaves when poll timeout is expired with Kafka Static membership?
In case of static membership I am observing that consumers are not sending LeaveGroup request when max.poll.interval.ms is breached.
I am only seeing the below log from kafka-coordinator-heartbeat-...
0
votes
0
answers
28
views
How does Kafka Consumer behave when poll timeout has expired?
Can someone let me know how a consumer is expected to behave after the below log? Will the consumer be considered dead and a new instance will be spawned due to consumer group rebalancing? How is this ...
2
votes
2
answers
132
views
Send Map from KafkaProducer to KafkaConsumer
I am using java 21 and Spring Boot 3.2.0 (SNAPSHOT). I want to use send a Map<Short,List<Customer>> from KafkaProducer to KafkaConsumer.
The KafkaConsumerConfig class:
@Configuration
...
1
vote
0
answers
56
views
Receiving duplicate msg in the consumer
Issue: Our consumer service is receiving duplicate msg even after placing a commit on the offset.
Tried: Initially we had the interval level commit and then moved to the offset level commit but still ...
0
votes
1
answer
45
views
How to gracefully stop Kafka consumer after processing a specific number of messages in Python?
I have an Airflow DAG with a BashOperator that runs a Kafka producer, generating a random count of messages. These messages are consumed by a Kafka consumer, which writes them to a JSON file. However, ...
0
votes
1
answer
30
views
How does migrated kafka consumer behave after repartitioning while cluster migration?
I have couple of conceptual doubts on the behaviour of consumers.
Cluster migration gives great opportunity while repartition topics which are over partitioned or under partitioned in current cluster.
...
0
votes
0
answers
23
views
Kafka producer producing message but consumer not consuming using Python
I am working with AWS EC2 instance. I am able to produce and consume data normally with kraft in terminal from my ubuntu 22.04 using ssh. When I try to produce data using a python script to the same ...
0
votes
1
answer
60
views
Spring KafkaListener disconnects and discontinues consuming - CommitFailedException
I have a Kafka topic that never purges messages (intentionally). I want to continually consume messages from it, even if my consumer goes offline for days/weeks/months. By my understanding, this may ...
0
votes
1
answer
46
views
KafkaConsumer with sasl_mechanism "OAUTHBEARER" not calling token() function
I am trying to get a minimal kafka consumer working, using kafka-python. Authentication is via OAuth.
I created a TokenProvider class, which is working fine. Any time, token() is called, it will check ...
0
votes
2
answers
35
views
Set retention period for Kafka
I'm trying to set the retention period on a kafka topic to 1000 ms to delete the unwanted messages on it.
I've used
bin/kafka-configs.sh --zookeeper my-zookeeper:2181 --alter --entity-type topics --...
0
votes
1
answer
33
views
Spring Kafka : How to prevent from partition revococation when processing long time jobs
I would like to ask, if there is any way to prevent consumer from being revoked when processing long time operation using @KafkaListener?
I have an application where there are jobs which takes a lot ...
0
votes
1
answer
23
views
Dynamic Topic Subscription for Kafka Consumer in Spring Boot
In my library class, I'm creating a Kafka consumer, and my configuration class is as follows:
@Service
public class ConsumerFactory {
private TopicNameProvider topicNameProvider;
private ...
0
votes
0
answers
40
views
Read message lag in Spring Kafka: consumer starts read message from retry topic instead main topic
KafkaProducerConfiguration.java:
@Configuration
public class KafkaProducerConfiguration {
@Bean
public ProducerFactory<String, String> producerFactory(KafkaProperties kafkaProperties) {
...
1
vote
1
answer
18
views
Retry job when consumer dies using Kafka
Can Kafka monitor the heartbeat of consumer and retry the job by persisting in the queue(Eventually sending it to a different consumer)?
Also I know that SQL table can be used as a queue for this ...
0
votes
2
answers
80
views
ConfluentKafka in Python: Using consumer to consumer set of records
I have started to learn confluent kafka(python). There is 1 producer, 1 topic, 1 partition and 1 consumer(a simple setup). My requirement is that I wish to fetch data collectively. I read that using ...
0
votes
0
answers
21
views
Error when trying to initialize a consumer with the librdkafka-dev 1.8.0-1build1 library in Kubuntu 22.04
I am trying to initialize a kafka consumer (C language) with the librdkafka-dev 1.8.0-1build1 library in Kubuntu 22.04. The problem is that I am getting an error when running the following line:
...
0
votes
0
answers
23
views
How to stop an infinite consumer in a java project?
I want to stop my consumer from running infinitely and also change topic name and group id in a loop and run my consumer. How can I proceed to do this?
I was experimenting if I could use currentLag() ...
-1
votes
2
answers
61
views
Spring-kafka: what is the right way to print RecordHeaders?
Just checked the logs and recognized that ConsumerRecord headers printed in absolutely unreadable manner and the reason is that RecordHeader holds value as byte[] as result in the logs I see the ...
0
votes
1
answer
50
views
How to consume the last message on a kafka message or consume a message based on timestamp? [duplicate]
def kafkaa(self, auto_offset_reset, timeout=500):
group_name = "group name"
config = {"bootstrap.servers": "server",
"...
0
votes
0
answers
26
views
Add persistent state store to kafka consumer in spring boot
Currently, I have a simple setup with a kafka producer and a kafka consumer, connected with a topic. On restart, the consumer reads all events from the topic from offset 0 to build an in memory cache ...
0
votes
1
answer
38
views
Spring Kafka Not Retryable Exceptions Filter
I'm trying to implement dynamic Kafka exception handling, specifically reties on some conditions but cannot find any way to filter exceptions that should not be retried not only by their class. My ...
0
votes
0
answers
13
views
Unable to use kafka broker from a Kubernetes pod [duplicate]
I have currently deployed a kafka broker in a kubernetes cluster which is listening to another service in the same namespace . I have to now test the service from my local . I need to send Json files ...
0
votes
0
answers
51
views
Kafka ssl.keystore.password field does not accept value as char array
I have a kafka client written in Java.How its done:
Properties props;
char[] keystorepwd = {t,e,s,t};
props.put("ssl.keystore.password",keystorepwd);
KafkaProducer<String,Object> = ...
0
votes
0
answers
39
views
Metrics misreporting for certain spring kafka config settings
I have a spring kafka consumer that receives batches and runs a loop to process each record in the batch. Also have a counter in the loop that increments each time we process a record.
I recently ...
1
vote
0
answers
37
views
KafkaConsumer polling timeout and performance impact on the broker
I have a Kafka consumer app with a polling consumption loop like this:
Consumer<byte[], byte[]> consumer = createConsumer();
Duration pollingTimeout = Duration.ofMillis(100);
while (true) {
...
0
votes
1
answer
78
views
Consumer not listen message from retry topic
I use Spring-Kafka 2.7.7 with @RetryableTopic to re-process messages from my two topics, MyTopic1 and MyTopic2, after some time
@RetryableTopic(
attempts ="3",
backoff = @Backoff(...
0
votes
0
answers
47
views
What causes: Attempt to heartbeat failed for since member id ... is not valid
I have an Kafka Streams application which running with 2 stream threads, everything is work fine until my log shows me:
2023-10-03 13:29:21 INFO FetchSessionHandler:381 - [Consumer clientId=...
0
votes
1
answer
44
views
Kafka Consumer not working on local using KafkaJS
I am trying to run a Kafka consumer in local using KafkaJS. It shows that the consumer is running but it's not consuming from a topic in which the producer is actively pushing events to. The consumer ...