Questions tagged [kafka-producer-api]
Use for questions related to the Apache Kafka producer API. Any question related to producing into Kafka topics. Producer failures and recover, idempotency, and transactional API.
1,928
questions
0
votes
0
answers
27
views
Kafka producer client is not able to connect to schema registry
I am trying to run a spark job on google Dataproc which pushes a record into a Kafka topic. It uses a schema registry for schema validation and serialization. Representative code below:
case class ...
0
votes
0
answers
12
views
How to mock Kafka Producer and producer.send method in PyTests
I am using Python Django Project and I have Kafka implemented in my project. Now I am writing my pytest cases for APIs and functions. However, I am unable to mock the Kafka Producers and Consumers.
I ...
1
vote
0
answers
85
views
Giving completion instructions ahead of calling send with MockProducer
I have the following code (Obviously made as an example):
class RandomClass(
val producer: Producer<String, String>
) {
fun randomFunction(): Boolean {
// Using .get() because I want to ...
1
vote
1
answer
83
views
How work maxAge in Spring Kafka Factory Producer?
We are using transactional producers and sometimes we find ourselves in the situation where there is no traffic for more than 7 days, which leads to the loss of transactional id metadata. The next ...
-1
votes
1
answer
27
views
where should I get missing properties for creating the Kafka producer?
Could you explain where should I get missing properties in the following code?
(smth should be in the lsat line)
KafkaProperties prop = new KafkaProperties();
KafkaProperties.Producer producer = prop....
1
vote
1
answer
26
views
How to get access to the confluent registry schema id used by kafka producer?
I have a kafka producer that uses confluent schema registry. I know that there is an algorithm based on which the KafkaAvroSerializer finds out the matching schemaId from the confluent schema registry ...
1
vote
0
answers
21
views
Does kafka producer reconnect fs2.Kafka
the microservice does not always switch to another kafka node, in case of loss of connection with the node to which it was connected and simply crashes.
If we create a producer and transfer several ...
0
votes
0
answers
34
views
Application running out of memory due to apache kafka metric objects while using kafkaProducer
We try to send 1000s of messages to a kafka-topic using kafkaProducer(org.apache.kafka.clients.producer.KafkaProducer) objects on our tomcat application. Lately we have been observing an issue with ...
0
votes
0
answers
68
views
Intermittently getting TimeoutException while pushing to kafka topic
Error:
org.springframework.kafka.KafkaException: Send failed; nested exception is org.apache.kafka.common.errors.TimeoutException: Topic topic_name not present in metadata after 500 ms
We are getting ...
2
votes
2
answers
134
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
...
0
votes
1
answer
47
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, ...
2
votes
1
answer
106
views
Failed to construct kafka producer when sending list of strings
I am new with Kafka and I am trying to read a text file and create a list of strings that I want to send for the consumers. I am using Java 21 and Spring Boot 3.2.0 (SNAPSHOT).
This is the Kafka ...
0
votes
0
answers
30
views
How to restart a kafka producer when SASL/SCRAM credential updates
I have a bean running in spring with a method called sendToKafka, in this method, a native kafka producer will be called to send messages to kafka, the client authentication is via SASL/SCRAM, i.e ...
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:
...
1
vote
0
answers
31
views
reporoducing org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) error
We had an incident in prod where some messages were lost. On post incident investigation they wanted to see the error occur, and see if our fix would work.
Mission: get org.apache.kafka.common.errors....
0
votes
0
answers
23
views
How to handle SSL/TLS with kafka broker and schema registry all together?
i sadly havent found any infos on my topic when doing research. I am currently working on a kafka producer and the messages which are being produced to the broker should be encrypted via SSL. In the ...
0
votes
0
answers
37
views
Is there a race condition when the kafka producer sends the message?
Kafka guarantees ordering "if the partition is well loaded".
The messages in the partition are processed in order when consumed.
But isn't it possible for a race condition to occur even when ...
0
votes
0
answers
32
views
Flink application running in kubernetes failing with ProducerFencedException
I have a flink application where I am setting execution.checkpointing.mode='EXACTLY_ONCE' and then I have a KafkaSink with the following setup
KafkaSink<KafkaSinkEvent> sink =
...
0
votes
1
answer
103
views
MSK - kafka-console-producer error with Kafka client version 3.4.0
I have been working with MSK Serverless/Provisioned and kafka-clients 3.4.0.
Everything worked as it should when I use * for a topic name.
arn:aws:kafka:region:account-id:topic/cluster-name/cluster-...
0
votes
1
answer
34
views
Transactional KafkaProducer recover from timeout
I am testing a Kotlin KafkaProducer application using Transactions and the possible scenarios:
Happy path: (working fine)
begin transaction
producer.send
commit transaction
Assert consume.poll has a ...
0
votes
0
answers
63
views
Kafka Producer time out during initTransactions while one broker is down
I'm using Kafka 2.7.0 and I have created kafka producer with Kafka Transaction. But it hangs and timeout while initialializing the Kafka transaction. In my Kafka cluster, I have 3 brokers.
public ...
1
vote
1
answer
64
views
Zombie fencing of a single producer with Spring Kafka
I have a Spring Kafka application, which has a single producer, which is NOT part of a pure kafka consume->process->produce chain, i.e. the producer is not triggered by a kafka consumer.
...
0
votes
1
answer
184
views
Spring Cloud Stream 4.0.4 | Kafka Producer | Functional Style
I have come across below example of writing Kafka Producer with Spring Cloud Stream in functional style.
Dependency
<dependency>
<groupId>org.springframework.cloud</groupId>
&...
0
votes
1
answer
44
views
Kafka Producer to distribute messages to all partitions equally
I have a producer that send messages to kafka topic. My topic has 10 partitions in same consumer group and I am looking for round robin distribution of messages. I have specified RoundRobinPartitioner ...
0
votes
0
answers
37
views
Kafka producer & consumer sync process
I want delete 10 items so I send 10 items to topic of kafka, my idea is that when producer send 10 items to kafka and it reponse ack, I return status 200 to client. But it has a problem if consumer ...
0
votes
1
answer
96
views
How to fix KafkaTimeoutError when connecting to Red Panda using kafka-python
I get the following error when trying to connect a kafka-python producer to my Red Panda database.
KafkaTimeoutError: KafkaTimeoutError: Failed to update metadata after 60.0 secs.
I have a ...
0
votes
2
answers
46
views
kafka topic with 1 partition . 1 producer is already producing to that partition. What issues can I face if i add another producer
I have a topic with 1 partition.
One of the systems is acting as a producer and sending data to that partition.
I want to add another different system as producer and send data to that partition only.
...
0
votes
0
answers
64
views
Kafka consumer with React js
I am developing a real-time data streaming app using Spring Boot(backend) and react.js(frontend). I have a Kafka producer through which I am producing messages to a specific topic and a consumer that ...
0
votes
0
answers
21
views
Should I recreate producer after ProducerFencedException?
If I got ProducerFencedException in my transactional producer, should I try to recreate it?
If I got this exception, it means that there is another producer with the same transaction.id, but if a new ...
0
votes
1
answer
67
views
Kafka producer - Cannot append records to read-only mirror topic
I running a producer and raising an exception with the following code.
@GetMapping("/send")
public ResultVo send(@RequestParam(value = "content") String content) {
...
1
vote
2
answers
138
views
All produced kafka messages are sending always to partition 0 even the key hash value is 1
I have a Kafka topic with 2 partitions. and I have created a message handler with 2 different keys.
@Bean
@ServiceActivator(inputChannel = "pushDataRequestChannel")
public MessageHandler ...
1
vote
1
answer
77
views
Does kafka include any schema registry version as part of avro messages?
I am working on a kafka project, implementing a producer, and one of the requirements is: "Each message generated shall include the respective schema id and version of the schema registry"
...
1
vote
1
answer
111
views
Kafka golang producer changing partition count after error
I have a kafka producer that lost communication to the broker temporarily. Each topic is defined with 10 partitions. When a connection is lost, I see the following log in the process running the ...
0
votes
0
answers
283
views
Kafka - Received invalid metadata error in produce request on partition. NetworkException when send large data to kafka topic
I try to upload file using kafka. When I send large file to kafka I got exception like this.
2023-08-01 14:46:00.754 INFO 1 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient : [Producer ...
0
votes
0
answers
77
views
Unclear behavior when using async Kafka writer and RequiredAcks set to RequireNone in segmentio/kafka-go
When using segmentio/kafka-go
It is unclear what is the difference in the behavior when writing with async mode and setting the RequiredAcks to different values: RequireNone, RequireOne, and ...
0
votes
1
answer
384
views
KafkaJSProtocolError: Request is not valid given the current SASL state
I have an AWS MKS cluster in which I could create a topic, produce a message and consume a message using installed Kafka in my ec2 server. But when I am trying to use the Kafka producer in my nodejs ...
1
vote
0
answers
113
views
Kafka Producer not producing through gunicorn and supervisor
I'm facing an odd situation because i can't seem to make my Kafka Producer work through my normal application flow. So in my `settings.py` file I create a global producer by calling this function:
...
0
votes
1
answer
67
views
WAL in a Kafka producer
I have an application (in Java 8) that sends messages to Kafka. In order to improve the performance I want to configure the producer to send messages in batches (using ProducerConfig.BATCH_SIZE_CONFIG ...
0
votes
1
answer
339
views
Kafka Failed authentication Unexpected handshake request with client mechanism PLAIN enabled mechanisms are []
I am getting below exception while trying to connect kafka in Ubuntu from Java Program in Windows
[main] INFO org.apache.kafka.clients.producer.KafkaProducer - [Producer clientId=producer-1] ...
0
votes
1
answer
174
views
Java docker kafka kraft simple connection [duplicate]
Hi I'm using docker to host a local kafka kraft server. I'm trying to make a simple connection using java
My docker docker-compose.yml looks like this
version: "3.5"
services:
kafka-gen:
...
0
votes
0
answers
28
views
Kakfa SASL password rotate in each and every 24 hours
Kafka password rotate each and every 24 hours. We have schedule that fetch latest password from file but how can we refresh Kafka consumers and producers with latest password.
2
votes
0
answers
212
views
Kafka producer getting disconnected after some time with error node -1 diconnected?
We have to publish messages to our Kafka broker, I am receiving messages every 2 sec from a service.
So I created a producer in Java, and I am using that producer throughout my application, Like ...
-1
votes
1
answer
16
views
Producer properties in a kafka streams app
In a kafka streams topology which produces data into multiple topics, is there a way to configure a single producer that produces into one of the output topics
0
votes
0
answers
23
views
Deduplacting the messages at Kafka Producer side
I have an uses to process CDC through Kafka SQL connector, I have an order with muliple items inside. If there is any change in the cart items , I need to send the respective order_id to consumer ...
0
votes
0
answers
41
views
Kafka Rest Proxy v2 with docker image 7.3.1: error when producing message with schema validation
I am using docker image 7.31 to use Rest Proxy.
I will use it to produce message to a confluent server Kafka.
I am using V2 use REST PROXY to produce records, such as this below
curl --location 'https:...
0
votes
0
answers
164
views
Getting IAMClientCallbackHandler class not found exception randomly for some messages
In our usecase currently we send messages one by one [ not in batch ] to maintain incoming order sequence while sending around more than 5000 messages one by one or sending a single message using ...
-1
votes
1
answer
454
views
Is it possible to send messages from Kafka to Open Search?
i'm writing a service using Java, i'm learning so it is for study purpose.
I wrote my rest app and it's working fine, then i added kafka that's sending a message to a consumer every time i do a ...
0
votes
1
answer
40
views
GStreamer cannot pickle 'Sample' object
I'm trying to send an encoded frame sample using Kafka, but Kafka requires that the sample to be Serializable, so the following error emerge:
TypeError: cannot pickle 'Sample' object
is there a way ...
1
vote
1
answer
346
views
Write to Dead Letter Topic in case on Deserialization Exception
I have a Spring Boot application that has a simple Consumer with @KafkaListener. I have blocking retry logic in place that works as expected but in case of Deserialization exception I want to store ...
0
votes
1
answer
51
views
Can a Spring Kafka Produce attempt return a "bad" offset without throwing an exception?
I am currently trying to override the KafkaTemplate to force a synchronous send method similar to the suggestion here.
Here is the code:
public class SyncKafkaTemplate<K, V> extends ...