Open mishikaraj opened 6 months ago
Hi @mishikaraj --out of curiosity, how are you setting up the consumer? My concern with the proposed solution is that there is some other issue that is being masked; destination_name
should get set as the topic from ConsumerRecord (which is returned from the consumer iterator)
This is the sample app I am using, so I want to make sure I am not missing an obvious implementation:
import json
import time
from datetime import datetime
from threading import Thread
import kafka
TOPIC = "test-topic-%d" % datetime.now().timestamp()
BROKERS = ["localhost:9092"]
consumer = kafka.KafkaConsumer(
TOPIC,
bootstrap_servers=BROKERS,
client_id="whatsup",
value_deserializer=lambda v: json.loads(v.decode("utf-8")),
auto_offset_reset="earliest",
consumer_timeout_ms=5000,
fetch_max_wait_ms=304999,
)
producer = kafka.KafkaProducer(
bootstrap_servers=BROKERS, value_serializer=lambda v: json.dumps(v).encode("utf-8")
)
def consume():
print("Starting consumer...")
for message in consumer:
print(f"Recieved {message.value}")
print("Consumer finished.")
def produce():
print("Starting producer...")
for json_message in [
{"foo": "bar"},
{"baz": "bat"},
{"user1": "Hello!"},
{"user2": "Hola!"},
]:
time.sleep(1)
producer.send(TOPIC, value=json_message).get()
producer.flush()
print("Producer finished.")
def main():
t1 = Thread(target=produce)
t2 = Thread(target=consume)
t1.start()
t2.start()
t1.join()
t2.join()
print("Finished.")
consumer.close()
producer.close()
if __name__ == "__main__":
main()
Hii @lrafeei , For Temporary Fix I had Disabled kafka producer transaction on newrelic by keeping below in newrelic.ini file and was able to send the message to kafka
[import-hook:kafka.producer.kafka] enabled = false
And for answering to your above question, Yes I have initialized my consumer in the same way
self.consumer = KafkaConsumer(
self.topic,
group_id=self.group_id,
bootstrap_servers=bootstrap_servers,
value_deserializer=lambda x: json.loads(v.decode("utf-8"))
enable_auto_commit=False,
auto_offset_reset="latest",
max_poll_records=500,
**consumer_kwargs,
)
Please let me know if you need any other help with this and what should be the fix we can look for enabling back the kafka producer transaction on newrelic for kafka producer
Sorry about the delay, I thought I'd get an alert when you replied! Looking at this now.
Description While publishing message to apache kafka , getting error with Newrelic Message Transaction - Error 'MessageTransaction' object has no attribute 'destination_name' code
Error Stack Trace
newrelic.ini file
Expected Behavior Expected message should have sent to kafka topic
Steps to Reproduce
Your Environment