Python producer (confluent-kafka) - Step-by-Step Execution

Process Flow - Python producer (confluent-kafka)

Create Producer Config

↓

Initialize Producer

↓

Prepare Message

↓

Call produce()

↓

Message Queued

↓

Call flush() to send

↓

Message Sent to Kafka

↓

Exit

This flow shows how a Python Kafka producer is configured, sends a message, and ensures delivery.

Execution Sample

Kafka

from confluent_kafka import Producer

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

This code creates a Kafka producer, sends one message with key and value, then waits for delivery.

Process Table

Step	Action	Internal State	Result/Output
1	Create Producer with config	Producer object created, connected to localhost:9092	Ready to send messages
2	Call produce() with topic='my_topic', key='key1', value='hello'	Message queued in producer buffer	No immediate output
3	Call flush()	Producer sends all queued messages to Kafka broker	Message delivered to 'my_topic'
4	Exit	No pending messages	Program ends

💡 All messages flushed and sent, producer cleanly closed

Status Tracker

Variable	Start	After produce()	After flush()	Final
p	None	Producer object with 1 message queued	Producer object with empty queue	Producer object closed
message_queue	Empty	1 message queued	Empty	Empty

Key Moments - 3 Insights

Why do we need to call flush() after produce()?

produce() only queues the message; flush() actually sends all queued messages to Kafka. See execution_table step 3.

What happens if we call produce() without flush()?

Messages may stay in the buffer and not be sent, so flush() ensures delivery before program ends (execution_table step 4).

Why do we pass 'bootstrap.servers' in the config?

It tells the producer where the Kafka broker is, so it can connect and send messages (execution_table step 1).

Visual Quiz - 3 Questions

Look at the execution_table, what is the state of the producer after calling produce()?

A Producer closed

B Message sent to Kafka broker

C Message queued in producer buffer

D No messages queued

💡 Hint

Check execution_table row 2 under 'Internal State'

At which step does the message actually get sent to Kafka?

A Step 3

B Step 1

C Step 2

D Step 4

💡 Hint

Look at execution_table row 3 'Result/Output'

If flush() is not called, what is the likely outcome?

A Message is sent immediately

B Message stays queued and may not be sent

C Producer throws an error

D Message is lost before queuing

💡 Hint

Refer to key_moments about flush() importance and execution_table step 4

Concept Snapshot

Python Kafka Producer (confluent-kafka):
- Create Producer with config {'bootstrap.servers': 'host:port'}
- Use produce(topic, key, value) to queue messages
- Call flush() to send all queued messages
- flush() ensures delivery before program ends
- Without flush(), messages may not be sent

Full Transcript

This visual trace shows how to use the Python confluent-kafka Producer. First, you create a Producer object with the Kafka broker address. Then, you call produce() to queue a message with a topic, key, and value. The message is not sent immediately but stored in a buffer. To actually send the message to Kafka, you call flush(), which sends all queued messages and waits for confirmation. Finally, the program ends after all messages are delivered. Remember, without flush(), messages may remain unsent. This step-by-step helps beginners see how the producer works internally and why flush() is important.

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