CODEWITHSUNDEEP
pythonjsonapache-kafka
hasherBaba
hasherBaba

Reputation: 339

Unable to get key in Kafka Python Consumer

I have written a code that would produce messages in key-value fashion in a topic:

from kafka import KafkaProducer
from kafka.errors import KafkaError
import json

producer = KafkaProducer(bootstrap_servers=['localhost:9092'])

# produce json messages
producer = KafkaProducer(value_serializer=lambda m: json.dumps(m).encode('ascii'))

deviceId = "4bc03533ccc94065"
responseId = "c03c4851-701f-4265-aafd-eb133c09c08e"

print deviceId
print responseId

producer.send('collect-response-devices', {'deviceId': deviceId})
producer.send('collect-response-responses', {'responseId' : responseId})

def on_send_success(record_metadata):
    print(record_metadata.topic)
    print(record_metadata.partition)
    print(record_metadata.offset)

def on_send_error(excp):
    log.error('I am an errback', exc_info=excp)
    print excp
    # handle exception

# block until all async messages are sent
producer.flush()

# configure multiple retries
producer = KafkaProducer(retries=5)

However, my consumer would consume messages, assign nothing(None in the key) and everything in the value part. 

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(bootstrap_servers='localhost:9092', auto_offset_reset='earliest', value_deserializer=lambda m: json.loads(m.decode('utf-8')))
consumer.subscribe(['collect-response-devices'])
for message in consumer:
  print (message.key, message.value)

This is the output of the consumer:

(None, {u'deviceId': u'4bc03533ccc94065'})

Upvotes: 0

Views: 2598

Answers (1)

Robin Moffatt
Robin Moffatt

Reputation: 32140

Per the docs, you need to specify the key when you produce the messages: 

# produce keyed messages to enable hashed partitioning
producer.send('my-topic', key=b'foo', value=b'bar')

At the moment you're not specifying a key when you produce the message, hence getting None

Upvotes: 1

Related Questions