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.

Filter by
Sorted by
Tagged with
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=...
Panda's user avatar
  • 417
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&...
Paolo Colombo's user avatar
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 ...
mez63's user avatar
  • 166
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 ...
SHIVANI SINGH's user avatar
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 ...
Prabhat_Seth's user avatar
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; ...
FSH's user avatar
  • 1
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....
azerila's user avatar
  • 879
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 ...
jimfawkes's user avatar
  • 345
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....
Varun Pius Rodrigues's user avatar
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 ...
mm117's user avatar
  • 71
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 ...
Vitalik's user avatar
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 ...
Goku's user avatar
  • 560
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 = "&...
Len Kagamine's user avatar
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 "/...
node_man's user avatar
  • 1,333
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 ...
LauraAlice's user avatar
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 ...
sovon's user avatar
  • 837
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, ...
Gree Tree Python's user avatar
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 ...
a.anghour's user avatar
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-...
Eagle1992's user avatar
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 ...
Firouziam's user avatar
  • 777
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 ...
chris's user avatar
  • 1
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....
Swastik's user avatar
  • 122
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 ...
Buket's user avatar
  • 1
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, ...
Swastik's user avatar
  • 122
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 ...
Isabel Lopez's user avatar
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 ...
jigiy43106's user avatar
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 ...
jigiy43106's user avatar
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. ...
jigiy43106's user avatar
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 = "...
jigiy43106's user avatar
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 ...
potterson11's user avatar
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 ...
Angel's user avatar
  • 15
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 ...
Kalleni's user avatar
  • 29
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", ...
Abhay Braja's user avatar
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 ...
Vishal Kamlapure's user avatar
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 ...
Vicky C's user avatar
  • 11
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=<...
Sam's user avatar
  • 151
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'], ...
Alex's user avatar
  • 5
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: ...
Andrea Fresa's user avatar
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 ...
Shaggy's user avatar
  • 159
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 ...
WW_'s user avatar
  • 99
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. ...
Piotr Pradzynski's user avatar
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 ...
Frank's user avatar
  • 609
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 ...
M Usman Wahab's user avatar
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 -...
AVarf's user avatar
  • 4,321
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 ...
kenu mor's user avatar
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 = ...
Rishikesh mahajan's user avatar
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 ...
Dev's user avatar
  • 756
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 ...
Tim's user avatar
  • 1
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 ...
aSaffary's user avatar
  • 793
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 ...
purvakashyap's user avatar

1
2 3 4 5
9