Questions tagged [kafka-python]
Kafka-Python provides low-level protocol support for Apache Kafka as well as high-level consumer and producer classes. Request batching is supported by the protocol as well as broker-aware request routing. Gzip and Snappy compression is also supported for message sets.
431
questions
1
vote
0
answers
52
views
Unable to connect to kafka running in docker container [duplicate]
I am unable to connect to kafka running in container.
I have .env file
KAFKA_BROKER_ID=1
KAFKA_ENABLE_KRAFT=true
KAFKA_CFG_PROCESS_ROLES=broker,controller
KAFKA_CFG_CONTROLLER_LISTENER_NAMES=...
0
votes
0
answers
6
views
kafka.consumer.fetcher.ConsumerRecord access to topic name
I'm trying to process a message received from akafka consumer
for message in self.consumer:
try:
self.process_message(message)
except Exception as e:
print(f&...
1
vote
1
answer
21
views
python kafka consumer.poll msg.error dict object has no attribute error
I am trying to set up kafka message polling using consumer poll.
I can see loads of example code that uses message.error
e.g.
while True:
try:
message = consumer.poll(10.0)
if ...
0
votes
1
answer
44
views
ModuleNotFoundError: No module named 'kafka'
I have downloaded Kafka and installed kafka-python library using "pip install kafka-python" and "conda install -c conda-forge kafka-python".
I am able to run "from kafka ...
0
votes
0
answers
23
views
How to setup Python connection to Kafka instance running in Kubernetes? [duplicate]
I am trying to connect to Kafka using Python and I am able to connect and see the topic names however when I try to read messages for any topic nothing happens. It feels like the connection times-out ...
0
votes
0
answers
17
views
kafka docker python API does not recognizes the broker [duplicate]
I have a kafka consumer in VSCode IDE (MS Windows) that works properly. I take the same python script in docker container, but the kafka broker is not recognized. I put them inside the same network; ...
0
votes
2
answers
49
views
dockerized Kafka service stuck at producer.send [duplicate]
I am very new to docker and Kafka, and have a simple kafka python publisher shown below
The following are in my dockerfile:
FROM python:3.10
WORKDIR /app
COPY . /app
RUN pip install --user pip==23....
1
vote
1
answer
48
views
How can I mock the instantiation of a Class
I can't seem to figure out how to mock the instantiation of a class, any pointers would be greatly appreciated. Here is what I am trying to do:
I would like to test the method ClassA.some_method() and ...
0
votes
2
answers
39
views
Python How to pass continuous generated data stream between methods
I have a method that continuously generates data and prints them on the console. Let's say something simple like generating random number:
def number_generator():
while True:
random....
3
votes
1
answer
382
views
Kafka + FastAPI + Docker template
Introduction
I am currently experimenting with Kafka and FastAPI and trying to build a template to enable me to quickly write programs in a microservice pattern.
Goal - Vision
Building a repository of ...
0
votes
1
answer
80
views
Can produce message with kafka-python, but can't with confluent-kafka
I try to use confluent-kafka, but can't manage to send message to topic.
I did the same with kafka-python, and message sends.
What i am missing?
Thank you!
i use this code
# using confluent-kafka
...
0
votes
1
answer
49
views
Python - Optimal way to check for a condition with a timeout?
Wrote two functions, 1. uploads a gzip file to Artifactory and 2. listens to the Kafka topic.
As further steps, wanted to validate whether the upload is successful within 5 minutes by listening to the ...
0
votes
0
answers
31
views
Getting 403 Error while trying to stream data to kafka from tweepy
from tweepy.streaming import StreamListener
import tweepy
from tweepy import OAuthHandler
from tweepy import Stream
from kafka import SimpleProducer, KafkaClient
import json
access_token = "&...
0
votes
0
answers
92
views
Python KafkaTimeoutError: Timeout waiting for future
I'm using Kafka to send logs to a topic. While sending the messages, I always get this error
Message: 'test log'
Arguments: ()
--- Logging error ---
Traceback (most recent call last):
File "/...
0
votes
0
answers
16
views
Cannot connect to Kafka broker, BrokerConnection error 111 [duplicate]
I have a container in the same docker-compose that is trying to connect to my broker. It suddenly stopped connecting and is resulting with error:
2023-02-22 14:24:00 ERROR:kafka.conn:Connect attempt ...
0
votes
2
answers
186
views
When kafka consumer connected to a broker by domain name, if the ip changed after broker crash, will the kafka consumer reconnect?
My kafka consumer is connected to a cluster of brokers. There is a domain name server in between. Each broker has a domain name associated with an IP address.
Problem is, the IPs are not static and ...
0
votes
1
answer
75
views
Kafka Consumer Committing With Auto-Commit Disabled
I'm missing events when reading from a Kafaka queue because the consumer is updating the offset without an explicit commit even when enable_auto_commit is disabled.
from kafka import KafkaClient, ...
0
votes
0
answers
43
views
Alter client quotas using kafka-python
I'm using kafka-python library to manage my kafka cluster. I want to add the quotas features (alter/describe), these features are not yet implented in kafka-python library but they are available in ...
0
votes
1
answer
162
views
Is there a suitable python library for doing stream processing with Kafka topics? [duplicate]
I am trying to find a suitable Python library to do stream processing with streams Kafka topics, Kafka streams. Specifically, I am looking for libraries that support the following operations.
KStream-...
0
votes
1
answer
44
views
kafka produce messages in consumer
I have an application with the need to pass messages into multiple layers of processing.
I need to do this because all the new messages should be put into the first generic topic so they can be ...
0
votes
1
answer
330
views
Kafka producer and consumer not working properly on python with docker
Im working on a project that uses kafka producer and consumer in order to acquire articles (with specific topics) from a news_api every two hours and then with a consumer save them in a mongodb.
So i ...
0
votes
1
answer
80
views
enable_auto_commit=False, but still offsets are committed automatically
I'm using kafka-python==2.0.2, and have disabled auto_commit but still if I don't commit through code, offsets are automatically getting committed
In the below code even if I comment out self.consumer....
0
votes
1
answer
45
views
kafka broker not collect data - CentOS7
I made a simple configuration with Kafka. I ran it as a service on 2 Centos servers. While I can send and see logs to the logs topic with my python scripts that I created on one of my servers, I only ...
0
votes
1
answer
267
views
How to perform async commit when using kafka-python
I'm using kafka-python library for my fastapi consumer app and I'm consuming messages in batch with maximum of 100 records. Since the topic has huge traffic and have only one partition, consuming, ...
2
votes
1
answer
433
views
Kafka consumer not consuming the data produced into topic
I have to create a kafka producer that generates a sequence of numbers from 1 to 300. Each of the message I wrote have to contain information about the topic, a key and the value which is a the binary ...
0
votes
0
answers
60
views
Calling async function within Tornado overriden methods
I am trying to call async functions in tornado overriden methods.
Initially, the function generate() is syncronous, but that seems to block the entire program, including on_connection_close method
...
1
vote
0
answers
29
views
Python Kafka Producer connection error despite consumer being able to connect [duplicate]
I am trying to create a producer:
producer = KafkaProducer(bootstrap_servers=SERVER, security_protocol="PLAINTEXT") # this line does not have issue. From my experiments, if it cannot ...
0
votes
0
answers
92
views
python-kafka asyncio is blocking
I am trying to create a websocket server that receives input from user, and at the same time read from kafka and continuously send data to user. I am having trouble getting both to run concurrently.
...
0
votes
1
answer
81
views
Python Kafka read message once only
I am trying to read the earliest/oldest message which has not been read before in Kafka.
import json
from kafka import KafkaConsumer
from kafka import KafkaProducer
ORDER_KAFKA_TOPIC = "...
0
votes
0
answers
89
views
Unable to consume Eventhub messages using Kafka API
I am trying to read messages from Azure Event Hub using kafka API. But I am unable to connect to the event hub. The bootstrap_connected() function always return false and the code stops there. It is ...
0
votes
0
answers
31
views
DNS lookup failed for kafka - ERROR kafka.conn:conn.py [duplicate]
I am getting the following error:
WARNING kafka.conn:conn.py:1527 DNS lookup failed for kafka:9092, exception was [Errno -3] Temporary failure in name resolution. Is your advertised.listeners (called ...
0
votes
0
answers
30
views
Kafka in GKE kubernetes internal communication between pods is not working [duplicate]
I'm trying to set up Kafka in Kubernetes (GKE) with pods for zookeeper, 1 broker, and 2 python apps: producer and consumer. However, I'm not able read and write to the topic correctly.
I have set up ...
0
votes
0
answers
20
views
How to consume messages from Kafka consumer to Django app in new docker container? [duplicate]
The Django server is going stuck if Kafka-comsumer starts. What should I do? I want to run this consumer as a sidecar container. Can anyone help me?
consumer = KafkaConsumer("my-topic", ...
0
votes
0
answers
49
views
Connect kafka on cloudera cdh using kerborose authentication with keytab file
We have a keborosed cloudera cluster setup for apache kafka, I am trying to connect it using python and I have installed the MIT kerborose for windos system
Main code:
import os
from kafka import ...
0
votes
1
answer
91
views
When creating a local Kafka instance for kafka-python, is it a good idea using Kafka 2.8+ that don't need Zookeeper?
In April 2021 Kafka released a version with early access to eliminating its dependency on Zookeeper. I've read many posts (mostly from 2021) saying that it was still not a good idea to use those ...
0
votes
0
answers
32
views
Unable to connect to On Prem Kafka topic from GCP
I am trying to consume a kafka topic using SASL but getting an Error which has no documentation :
ERROR:kafka.conn:Connect attempt to <BrokerConnection node_id=<nodeid> host=<...
0
votes
1
answer
407
views
read kafka topics from python
This code doesn't display any messages successfully
What's wrong?
from kafka import KafkaConsumer
consumer = KafkaConsumer(
'topictest',
bootstrap_servers=['server:9092'],
...
0
votes
0
answers
355
views
Consumer do not join a consumer group
I am using kafka-python and I want to consume messages from a topic. For monitoring reasons, I want to create a consumer and assign it to a consumer'group.
I am using the following functions:
...
0
votes
1
answer
301
views
Kafka consumer failing due to nulls for value serializer
I have a Kafka consumer reading from a producer, but sometimes upon a record deletion from a table, null values get picked up by the consumer resulting in NoneType error message in the value ...
0
votes
1
answer
825
views
how to poll/listen to a Kafka topic continuously?
I have to listen to a topic on Kafka and consume the messages being published on it.
My piece of code is working fine but sometimes gets stuck in the try block as poll returns empty dictionary.
Please ...
0
votes
1
answer
406
views
Reason on uneven lag on kafka partitions
I have a kafka topic partitioned into 60 partitions. Producer distributes messages in a default round-robin manner. So far so good. But when it comes to consume, a lag is growing. Nothing surprising. ...
0
votes
1
answer
452
views
Kafka topic appears to be only consumed by one KafkaConsumer instance in Airflow
I have a system where I start two separate processes, one that produces messages to a kafka topic, and one process that runs a number of KafkaConsumer instances, and consumes the messages. However, it ...
0
votes
0
answers
31
views
Unable to Send Messages through Kafka-Python Producer [duplicate]
I've dockerize kafka using wurstmeister/kafka image. I'm new to Kafka and integrating with my django project. I've created a management commands for consumer.py and producer.py.
docker-compose.yml
...
0
votes
2
answers
875
views
Reading the last message from Kafka
I am trying to read the last message from a Kafka topic but I cannot make it work. I tried different methods that you can find below with their errors or problems
Topic description:
$ kafka-topics.sh -...
1
vote
1
answer
89
views
Does Kafka to support bidirectional?
I have two servers where bidirectional communication needs to be happened.
For kafka I have 2 topics 2 producers and 2 consumers. Trying to communicate server 1 to 2 with 1topic and 1 consumer and 1 ...
0
votes
1
answer
227
views
How to reduce Kafka lag? and How to create multiple consumers for same topic?
def getKafkaConsumer(topicName, kafkaIP, kafkaPort, kafkaConsumerGroup): try: bootstrap = [str(kafkaIP + ':' + kafkaPort)] consumer = KafkaConsumer(topicName, bootstrap_servers = bootstrap, group_id = ...
1
vote
2
answers
1k
views
Add health check to Kafka consumer k8s pods
How to add health check(liveness/readiness) in Kafka consumer pods in my Django application. New to this infra-structure related things. Please help in implementing a periodic health check for Kafka ...
0
votes
1
answer
497
views
How to use KafkaConsumer with Django4
I have a Django 4 project and using KafkaConsumer from kafka-python. I want to update django models after receiving a Kafka message. The goal here is to have some Kafka worker running and consuming ...
0
votes
1
answer
291
views
in Apache Kafka, can I limit the number of partitions assigned to a certain consumer?
I'm using Kafka to distribute load between my AI agents but my servers have different configurations and different rates for processing the input data, a few of them remain mostly idle while others ...
1
vote
1
answer
929
views
How to mock Kafka Producer and producer.send method for Python unit test
I am new to Python and trying to write a unit test which involves Kafka. I have a class with a function which calls another function to initialize Kafka Producer and then call producer.send(). I want ...