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.

Filter by
Sorted by
Tagged with
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 ...
Tarique's user avatar
  • 430
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 ...
NVS's user avatar
  • 400
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 ...
Stene's user avatar
  • 156
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 ...
Fonexn's user avatar
  • 199
-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....
Anatoly's user avatar
  • 3,472
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 ...
Marinos An's user avatar
  • 9,810
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 ...
SCALIST's user avatar
  • 11
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 ...
Srinish's user avatar
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 ...
tusharRawat's user avatar
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 ...
Al3x4ndru1's user avatar
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, ...
smert97's user avatar
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 ...
Al3x4ndru1's user avatar
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 ...
Liu's user avatar
  • 67
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
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....
Ben Fourie's user avatar
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 ...
paul's user avatar
  • 11
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 ...
bonjugi's user avatar
  • 11
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 = ...
Parag Panda's user avatar
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-...
Aleksandar Hardi's user avatar
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 ...
marionmaiden's user avatar
  • 3,270
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 ...
Chathurika's user avatar
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. ...
selbstereg's user avatar
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> &...
Half Blood Prince's user avatar
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 ...
harish bollina's user avatar
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 ...
web laptrinh's user avatar
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 ...
exquisitecharm's user avatar
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. ...
vashi's user avatar
  • 9
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 ...
Ankur Das Shanto's user avatar
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 ...
hololensen's user avatar
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) { ...
LIANJI LI's user avatar
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 ...
Kiran Kumar's user avatar
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" ...
Marinos An's user avatar
  • 9,810
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 ...
Omri. B's user avatar
  • 385
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 ...
Vusal Rehimov's user avatar
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 ...
dorony's user avatar
  • 1,026
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 ...
ashique's user avatar
  • 993
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: ...
Pedro Silva's user avatar
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 ...
user983447's user avatar
  • 1,609
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] ...
Rakesh Yadav's user avatar
  • 1,976
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: ...
Marco Aviles's user avatar
  • 5,536
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.
kapilcs032's user avatar
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 ...
Deepak Kaushik's user avatar
-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
Gayatri Mahesh's user avatar
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 ...
Shasu's user avatar
  • 462
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:...
CHRISTOPHE LANNOY's user avatar
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 ...
shiva789's user avatar
-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 ...
user avatar
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 ...
Suhaib Al Hindi's user avatar
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 ...
user13774712's user avatar
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 ...
Oak's user avatar
  • 36

1
2 3 4 5
39