Python producer (confluent-kafka) - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this Kafka producer code?

Consider the following Python code using confluent-kafka to produce messages. What will be printed when this code runs?

Kafka

from confluent_kafka import Producer

def delivery_report(err, msg):
    if err is not None:
        print(f"Message delivery failed: {err}")
    else:
        print(f"Message delivered to {msg.topic()} [{msg.partition()}] at offset {msg.offset()}")

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

p.produce('test_topic', key='key1', value='value1', callback=delivery_report)
p.flush()

A No output printed

B Message delivery failed: Broker transport failure

C Message delivered to test_topic [0] at offset 0

D SyntaxError: invalid syntax

💡 Hint

Think about what the delivery_report callback prints when the message is successfully sent.

The delivery_report callback prints a success message with topic, partition, and offset if there is no error. Since the broker is reachable, the message is delivered successfully.

❓ Predict Output

intermediate

What error does this Kafka producer code raise?

What error will this code raise when executed?

Kafka

from confluent_kafka import Producer

p = Producer({'bootstrap.servers': 'localhost:9092'})
p.produce('test_topic', value=12345)
p.flush()

A TypeError: value must be bytes or string

B ValueError: topic name invalid

C No error, message sent successfully

D KeyError: 'value'

💡 Hint

Check the type of the value argument expected by produce.

The produce method expects the value to be a string or bytes. Passing an integer causes a TypeError.

🔧 Debug

advanced

Why does this Kafka producer code hang indefinitely?

This code hangs and never finishes. What is the cause?

Kafka

from confluent_kafka import Producer

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

for i in range(5):
    p.produce('test_topic', value=f'message {i}')

# Missing flush call here

A The produce method is blocking and never returns

B The topic name is invalid causing a hang

C The loop causes an infinite loop

D The producer buffers messages and needs <code>flush()</code> to send them

💡 Hint

Think about how the producer sends messages to Kafka brokers.

The produce method queues messages asynchronously. Without calling flush(), the program may exit before messages are sent, or appear to hang waiting for delivery.

📝 Syntax

advanced

Which option causes a syntax error in Kafka producer code?

Which of these code snippets will cause a syntax error when defining the Kafka producer?

A p = Producer({'bootstrap.servers' = 'localhost:9092'})

B p = Producer({'bootstrap.servers': 'localhost:9092', 'enable.idempotence': True})

C p = Producer({'bootstrap.servers': 'localhost:9092'})

D p = Producer({'bootstrap.servers': 'localhost:9092', 'client.id': 'myclient'})

💡 Hint

Check the syntax for dictionary key-value pairs in Python.

In Python dictionaries, key-value pairs use a colon :, not an equals sign =. Option A uses = which causes a syntax error.

🚀 Application

expert

How many messages are successfully produced by this code?

Given this code snippet, how many messages will be successfully produced and delivered?

Kafka

from confluent_kafka import Producer

results = []

def delivery_report(err, msg):
    if err is None:
        results.append(msg.value().decode('utf-8'))

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

for i in range(3):
    p.produce('test_topic', value=f'msg{i}'.encode('utf-8'), callback=delivery_report)

p.flush()

print(len(results))

A 1

B 3

C 0

D Raises a TypeError

💡 Hint

Consider how the callback updates the results list and when flush() is called.

The flush() call waits for all messages to be delivered, so the callback runs for each message, appending to results. Thus, len(results) is 3.