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.

Practice

1. What is the main purpose of the produce() method in the Python confluent-kafka producer?

easy

A. To consume messages from a Kafka topic

B. To connect to the Kafka server

C. To send a message to a Kafka topic

D. To wait for all messages to be delivered

Solution

1. Step 1: Understand the role of produce()

   The produce() method is used to send messages to a Kafka topic from the producer.

2. Step 2: Differentiate from other methods

   flush() waits for messages to be sent, and connection is set by configuration, not produce().

3. Final Answer:

   To send a message to a Kafka topic -> Option C

4. Quick Check:

   produce() sends messages [OK]

Hint: Remember: produce() means send message [OK]

Common Mistakes:

• Confusing produce() with flush()
• Thinking produce() connects to Kafka
• Mixing producer and consumer roles

2. Which of the following is the correct way to set the Kafka server address in the Python confluent-kafka producer configuration?

easy

A. "bootstrap.servers" : "localhost:9092"

B. "server.bootstrap" : "localhost:9092"

C. "kafka.server" : "localhost:9092"

D. "broker.address" : "localhost:9092"

Solution

1. Step 1: Recall the correct configuration key

   The correct key to specify Kafka server address is bootstrap.servers.

2. Step 2: Match the option

   Only "bootstrap.servers" : "localhost:9092" uses the exact key "bootstrap.servers" with a valid address format.

3. Final Answer:

   "bootstrap.servers" : "localhost:9092" -> Option A

4. Quick Check:

   Kafka server config = bootstrap.servers [OK]

Hint: Look for exact key: bootstrap.servers [OK]

Common Mistakes:

• Swapping key names like server.bootstrap
• Using incorrect keys like kafka.server
• Missing quotes or wrong format

3. What will be the output of the following Python code using confluent-kafka producer?

from confluent_kafka import Producer

p = Producer({'bootstrap.servers': 'localhost:9092'})
p.produce('my_topic', key='key1', value='hello')
p.flush()
print('Message sent')

medium

A. Error: No Kafka server found

B. hello

C. No output

D. Message sent

Solution

1. Step 1: Understand the code flow

   The code creates a producer, sends a message to 'my_topic', waits for delivery with flush(), then prints 'Message sent'.

2. Step 2: Identify the printed output

   The only print statement outputs the string 'Message sent'. The message sending is asynchronous but flush ensures completion before print.

3. Final Answer:

   Message sent -> Option D

4. Quick Check:

   flush() waits, then print runs [OK]

Hint: flush() waits, print runs after sending [OK]

Common Mistakes:

• Expecting message value to print
• Ignoring flush() effect
• Assuming error without Kafka running

4. Identify the error in this Python confluent-kafka producer code:

from confluent_kafka import Producer

p = Producer({'bootstrap.servers': 'localhost:9092'})
p.produce('my_topic', key='key1', 'hello')
p.flush()

medium

A. Missing key argument in produce()

B. Incorrect order of arguments in produce()

C. flush() called before produce()

D. Missing import statement

Solution

1. Step 1: Check produce() argument order

   The produce() method expects positional arguments first (topic, optionally value), followed only by keyword arguments. Here, key='key1' (keyword) is followed by 'hello' (positional), causing SyntaxError: positional argument follows keyword argument.

2. Step 2: Identify correct usage

   Correct usage: produce('my_topic', value='hello') or produce('my_topic', key='key1', value='hello').

3. Final Answer:

   Incorrect order of arguments in produce() -> Option B

4. Quick Check:

   Positional args before keywords [OK]

Hint: Positional args must precede keywords [OK]

Common Mistakes:

• Placing positional argument after keyword argument
• Mixing positional and keyword argument order
• Forgetting Python positional-before-keyword rule

5. You want to send multiple messages to Kafka and ensure all are delivered before your program exits. Which sequence of calls is correct using confluent-kafka producer in Python?

hard

A. Call produce() for each message, then call flush() once at the end

B. Call flush() after each produce() call

C. Call flush() before any produce() calls

D. Call produce() once, then call flush() multiple times

Solution

1. Step 1: Understand message sending behavior

   produce() queues messages asynchronously. Calling it multiple times queues multiple messages.

2. Step 2: Use flush() to wait for delivery

   Calling flush() once after all produce() calls waits for all messages to be sent before program exit.

3. Final Answer:

   Call produce() for each message, then call flush() once at the end -> Option A

4. Quick Check:

   Produce many, flush once after all [OK]

Hint: Flush once after all produce calls to wait [OK]

Common Mistakes:

• Flushing after every produce (slow)
• Flushing before producing messages
• Not flushing at all causing lost messages