Questions tagged [kafka-consumer-api]

Use for questions related to the Apache Kafka consumer API

Filter by
Sorted by
Tagged with
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 ...
RedRum69's user avatar
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 ...
Dami00's user avatar
  • 1
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:...
Thor's user avatar
  • 133
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 ...
Siddharth's user avatar
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 ...
Ritik's user avatar
  • 51
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. ...
Zixel's user avatar
  • 321
-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 (...
Bindu Yadav's user avatar
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 ...
Сергей Степин's user avatar
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 ...
techie11's user avatar
  • 1,257
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 ...
Agustin's user avatar
  • 41
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 ...
Ulrich Schuster's user avatar
-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 ...
AP1709's user avatar
  • 29
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 ...
Firman Alam's user avatar
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 ...
BenS1's user avatar
  • 103
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 ...
Charu Jain's user avatar
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 ...
james's user avatar
  • 95
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 ...
Surajit Chongder's user avatar
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 ...
Steven Kristian's user avatar
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 ...
Mdarende's user avatar
  • 529
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 ...
Deepak_Spark_Beginner's user avatar
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 ...
thao nguyen's user avatar
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 ...
SHIVAM SINGH's user avatar
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-...
tuk's user avatar
  • 6,079
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 ...
tuk's user avatar
  • 6,079
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 ...
Al3x4ndru1's user avatar
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 ...
Prabhat Mishra's user avatar
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, ...
smert97's user avatar
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. ...
CoderClown's user avatar
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 ...
Archit Mishra's user avatar
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 ...
Michael C's user avatar
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 ...
treuss's user avatar
  • 2,177
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 --...
Tiz's user avatar
  • 413
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 ...
F1zz4's user avatar
  • 7
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 ...
Bertug's user avatar
  • 934
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) { ...
Vasily's user avatar
  • 5
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 ...
vamsi krishna's user avatar
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 ...
the_he_man's user avatar
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: ...
vitu's user avatar
  • 1
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() ...
Manzz's user avatar
  • 1
-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 ...
user471011's user avatar
  • 7,171
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", "...
Xiaoyue Cheng's user avatar
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 ...
Jdv's user avatar
  • 976
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 ...
Petro Prydorozhnyi's user avatar
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 ...
Ashwin's user avatar
  • 43
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> = ...
Omi's user avatar
  • 998
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 ...
Aakanksha Sharma's user avatar
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) { ...
Kohei Nozaki's user avatar
  • 1,174
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(...
javaLearnMushigi's user avatar
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=...
Huy Nguyễn's user avatar
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 ...
Anupam's user avatar
  • 41

1
2 3 4 5
86