First message (produce and consume) in Kafka - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this Kafka producer code?

Consider this Kafka producer code snippet in Python using confluent_kafka. What will it print?

Kafka

from confluent_kafka import Producer

conf = {'bootstrap.servers': 'localhost:9092'}
producer = Producer(conf)

producer.produce('test-topic', key='key1', value='Hello Kafka!')
producer.flush()

print('Message sent')

A TimeoutError

B Message sent

C KeyError

D Hello Kafka!

💡 Hint

The produce method queues the message, and flush sends it. The print statement runs after sending.

The code prints 'Message sent' because the print statement executes after the message is sent. The message itself is sent to Kafka but not printed.

❓ Predict Output

intermediate

What does this Kafka consumer print after consuming one message?

Given this Kafka consumer code in Python, what will it print after receiving one message?

Kafka

from confluent_kafka import Consumer

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'mygroup',
    'auto.offset.reset': 'earliest'
}
consumer = Consumer(conf)
consumer.subscribe(['test-topic'])

msg = consumer.poll(1.0)
if msg is None:
    print('No message received')
elif msg.error():
    print(f'Error: {msg.error()}')
else:
    print(f'Received: {msg.value().decode("utf-8")}')
consumer.close()

A Error: KafkaError{code=_TIMED_OUT}

B No message received

C Received: key1

D Received: Hello Kafka!

💡 Hint

The consumer reads the message value, decoding bytes to string.

The consumer prints the message value 'Hello Kafka!' because it consumes the message sent to 'test-topic'.

🔧 Debug

advanced

Why does this Kafka consumer code raise an error?

This Kafka consumer code raises an exception. What is the cause?

Kafka

from confluent_kafka import Consumer

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'mygroup'
}
consumer = Consumer(conf)
consumer.subscribe(['test-topic'])

msg = consumer.poll(1.0)
print(msg.value().decode('utf-8'))
consumer.close()

A TimeoutError because poll timeout is too short

B KeyError because 'auto.offset.reset' is missing

C AttributeError because msg is None and msg.value() is called

D TypeError because decode is called on a string

💡 Hint

Check what happens if poll returns None.

If no message is received, poll returns None. Calling msg.value() on None causes an AttributeError.

📝 Syntax

advanced

Which option correctly produces a message with key and value in Kafka?

Which of these Python code snippets correctly produces a Kafka message with key 'user1' and value 'data'?

A producer.produce('topic', key='user1', value='data')

B producer.produce('topic', 'user1', 'data')

C producer.produce(topic='topic', key='user1', value='data')

D producer.produce('topic', key=user1, value=data)

💡 Hint

Remember that key and value must be strings or bytes, and keyword arguments are required.

Option A uses correct keyword arguments with string literals. Option A passes positional arguments incorrectly. Option A uses an invalid keyword argument 'topic'. Option A uses undefined variables.

🚀 Application

expert

How many messages will be consumed by this Kafka consumer loop?

Given this Kafka consumer code, how many messages will it consume before stopping?

Kafka

from confluent_kafka import Consumer

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'mygroup',
    'auto.offset.reset': 'earliest'
}
consumer = Consumer(conf)
consumer.subscribe(['test-topic'])

count = 0
while True:
    msg = consumer.poll(0.5)
    if msg is None:
        break
    if msg.error():
        break
    count += 1
consumer.close()
print(count)

A Number of messages currently in 'test-topic' partition(s)

B 0 because poll timeout is too short

C Infinite loop because break conditions are wrong

D 1 because it stops after first message

💡 Hint

The loop stops when poll returns None, which means no more messages are available.

The code counts all available messages in 'test-topic' partitions at the time of consumption. It stops when no more messages are returned.

Practice

1. What is the main purpose of producing a message in Kafka?

easy

A. To send data to a Kafka topic

B. To read data from a Kafka topic

C. To delete messages from a Kafka topic

D. To create a new Kafka topic

Solution

1. Step 1: Understand Kafka producing

   Producing means sending messages to a Kafka topic as bytes.

2. Step 2: Differentiate producing from consuming

   Consuming reads messages, but producing sends them.

3. Final Answer:

   To send data to a Kafka topic -> Option A

4. Quick Check:

   Producing = Sending messages [OK]

Hint: Producing means sending messages to Kafka topics [OK]

Common Mistakes:

• Confusing producing with consuming
• Thinking producing deletes messages
• Assuming producing creates topics

2. Which of the following is the correct way to create a Kafka producer in Python using the kafka-python library?

easy

A. producer = KafkaConsumer(topic='mytopic')

B. producer = KafkaConsumer(bootstrap_servers='localhost:9092')

C. producer = KafkaProducer(topic='mytopic')

D. producer = KafkaProducer(bootstrap_servers='localhost:9092')

Solution

1. Step 1: Identify Kafka producer creation syntax

   KafkaProducer is created with bootstrap_servers parameter specifying Kafka server address.

2. Step 2: Eliminate consumer and wrong parameters

   KafkaConsumer is for consuming, not producing. Topic is not a parameter for producer creation.

3. Final Answer:

   producer = KafkaProducer(bootstrap_servers='localhost:9092') -> Option D

4. Quick Check:

   Producer uses KafkaProducer with bootstrap_servers [OK]

Hint: Producer uses KafkaProducer with bootstrap_servers [OK]

Common Mistakes:

• Using KafkaConsumer instead of KafkaProducer
• Passing topic during producer creation
• Missing bootstrap_servers parameter

3. Consider this Python code snippet using kafka-python:

from kafka import KafkaProducer, KafkaConsumer

producer = KafkaProducer(bootstrap_servers='localhost:9092')
producer.send('test-topic', b'Hello Kafka')
producer.flush()

consumer = KafkaConsumer('test-topic', bootstrap_servers='localhost:9092', auto_offset_reset='earliest', consumer_timeout_ms=1000)
for msg in consumer:
    print(msg.value.decode())

What will be the output when this code runs?

medium

A. Error: topic not found

B. Hello Kafka

C. No output, consumer times out

D. b'Hello Kafka'

Solution

1. Step 1: Analyze producer sending message

   Producer sends bytes message b'Hello Kafka' to 'test-topic' and flushes to ensure delivery.

2. Step 2: Analyze consumer reading message

   Consumer reads from 'test-topic' from earliest offset and decodes bytes to string before printing.

3. Final Answer:

   Hello Kafka -> Option B

4. Quick Check:

   Consumer prints decoded message string [OK]

Hint: Consumer decodes bytes to string before printing [OK]

Common Mistakes:

• Expecting raw bytes output
• Assuming consumer misses message due to timeout
• Thinking topic does not exist

4. You wrote this code to consume messages but it raises an error:

consumer = KafkaConsumer('my-topic')
for message in consumer:
    print(message.value.decode())

What is the most likely cause of the error?

medium

A. Using decode() on message.value which is already a string

B. Not calling consumer.poll() before iterating

C. Missing bootstrap_servers parameter when creating KafkaConsumer

D. Topic name should be a list, not a string

Solution

1. Step 1: Check KafkaConsumer creation requirements

   KafkaConsumer requires bootstrap_servers parameter to connect to Kafka cluster.

2. Step 2: Validate other options

   decode() is needed because message.value is bytes; consumer.poll() is optional; topic can be string or list.

3. Final Answer:

   Missing bootstrap_servers parameter when creating KafkaConsumer -> Option C

4. Quick Check:

   KafkaConsumer needs bootstrap_servers [OK]

Hint: Always specify bootstrap_servers when creating KafkaConsumer [OK]

Common Mistakes:

• Omitting bootstrap_servers parameter
• Assuming message.value is string without decoding
• Thinking poll() is mandatory before iteration

5. You want to produce and consume the first message in a Kafka topic named 'orders'. Which sequence of steps correctly achieves this in Python using kafka-python?

hard

A. Create KafkaProducer with bootstrap_servers, send message as bytes, flush; create KafkaConsumer with bootstrap_servers and auto_offset_reset='earliest', then read and decode messages

B. Create KafkaConsumer first, then KafkaProducer; send message as string; consumer reads without decoding

C. Create KafkaProducer without bootstrap_servers, send string message; create KafkaConsumer with bootstrap_servers, read messages as bytes

D. Create KafkaProducer with topic parameter, send message; create KafkaConsumer without bootstrap_servers, read and decode messages

Solution

1. Step 1: Properly create KafkaProducer and send message

   KafkaProducer must be created with bootstrap_servers; messages sent as bytes and flushed to ensure delivery.

2. Step 2: Properly create KafkaConsumer and read messages

   KafkaConsumer needs bootstrap_servers and auto_offset_reset='earliest' to read from the first message; decode bytes to string when reading.

3. Final Answer:

   Create KafkaProducer with bootstrap_servers, send message as bytes, flush; create KafkaConsumer with bootstrap_servers and auto_offset_reset='earliest', then read and decode messages -> Option A

4. Quick Check:

   Correct producer and consumer setup with bytes and decoding [OK]

Hint: Producer sends bytes and flushes; consumer reads earliest and decodes [OK]

Common Mistakes:

• Creating producer or consumer without bootstrap_servers
• Sending string instead of bytes
• Not setting auto_offset_reset to 'earliest'
• Not decoding message bytes before printing