Producer API basics in Kafka - Step-by-Step Execution

Process Flow - Producer API basics

Create Producer

↓

Prepare Message

↓

Send Message

↓

Broker Receives

↓

Acknowledge Success/Failure

↓

Close Producer

This flow shows how a Kafka Producer is created, sends a message to the broker, waits for acknowledgment, and then closes.

Execution Sample

Kafka

from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers='localhost:9092')
producer.send('my_topic', b'Hello Kafka')
producer.flush()
producer.close()

This code creates a Kafka producer, sends a message 'Hello Kafka' to 'my_topic', waits for sending to complete, and closes the producer.

Process Table

Step	Action	Evaluation	Result
1	Create KafkaProducer instance	Connect to localhost:9092	Producer ready to send messages
2	Prepare message 'Hello Kafka' for 'my_topic'	Message encoded as bytes	Message ready to send
3	Send message	Send request sent to broker	Message in transit
4	Broker receives message	Broker stores message in 'my_topic'	Message stored successfully
5	Producer flush	Wait for all messages to be acknowledged	All messages confirmed sent
6	Close producer	Release resources	Producer closed
7	End	No more messages to send	Execution stops

💡 Producer closed and no more messages to send

Status Tracker

Variable	Start	After Step 1	After Step 2	After Step 3	After Step 5	Final
producer	None	KafkaProducer instance	Same instance	Same instance	Same instance	Closed
message	None	None	b'Hello Kafka'	Sent	Sent and acknowledged	Sent and acknowledged

Key Moments - 3 Insights

Why do we call producer.flush() before closing?

flush() waits for all messages to be sent and acknowledged by the broker, ensuring no message is lost before closing, as shown in step 5 of the execution_table.

What happens if we send a message without encoding it as bytes?

KafkaProducer expects messages as bytes; if not encoded, sending will fail or raise an error, as seen in step 2 where message is prepared as bytes.

Why do we need to close the producer?

Closing releases network and system resources properly, preventing leaks. This is the final step (6) in the execution_table.

Visual Quiz - 3 Questions

Look at the execution_table, what is the state of 'producer' after step 3?

A Producer not yet created

B KafkaProducer instance ready to send more messages

C Producer closed

D Producer flushed and closed

💡 Hint

Check variable_tracker column 'After Step 3' for 'producer'

At which step does the broker confirm the message is stored?

A Step 4

B Step 2

C Step 5

D Step 6

💡 Hint

See execution_table row describing 'Broker receives message'

If we skip producer.flush(), what might happen?

A All messages are guaranteed sent

B Producer will automatically flush on close

C Messages might not be sent before closing

D Producer will throw an error immediately

💡 Hint

Refer to key_moments about flush importance and execution_table step 5

Concept Snapshot

Kafka Producer API basics:
- Create KafkaProducer with bootstrap_servers
- Send messages as bytes with producer.send(topic, message)
- Use producer.flush() to wait for all sends
- Close producer to free resources
- Messages are sent asynchronously and acknowledged by broker

Full Transcript

This visual execution shows how a Kafka Producer works step-by-step. First, the producer is created connecting to the Kafka broker. Then, a message is prepared by encoding it as bytes. The message is sent asynchronously to the broker, which stores it in the topic. The producer.flush() call waits until all messages are confirmed sent to avoid loss. Finally, the producer is closed to release resources. Variables like 'producer' and 'message' change state through these steps. Key points include the need to flush before closing and encoding messages as bytes. The quizzes test understanding of these steps and states.

Practice

1. What is the main purpose of the Kafka Producer API?

easy

A. To send messages to Kafka topics

B. To read messages from Kafka topics

C. To manage Kafka cluster configurations

D. To monitor Kafka broker health

Solution

1. Step 1: Understand Kafka Producer role

   The Producer API is designed to send data to Kafka topics, acting as the message sender.

2. Step 2: Differentiate from other Kafka APIs

   Reading messages is done by the Consumer API, not the Producer API.

3. Final Answer:

   To send messages to Kafka topics -> Option A

4. Quick Check:

   Producer API = send messages [OK]

Hint: Producer API sends data; Consumer API reads data [OK]

Common Mistakes:

• Confusing Producer with Consumer API
• Thinking Producer manages cluster
• Assuming Producer monitors brokers

2. Which of the following is the correct way to create a Kafka Producer in Java?

easy

A. Producer<String> producer = new Producer(props);

B. KafkaProducer producer = new KafkaProducer<>(props);

C. KafkaProducer producer = new KafkaProducer();

D. KafkaProducer producer = KafkaProducer.create(props);

Solution

1. Step 1: Recall KafkaProducer constructor usage

   The KafkaProducer class requires properties and type parameters for key and value types.

2. Step 2: Check each option's syntax

   KafkaProducer producer = new KafkaProducer<>(props); correctly uses generics and constructor with properties. Others miss generics, constructor, or use invalid methods.

3. Final Answer:

   KafkaProducer producer = new KafkaProducer<>(props); -> Option B

4. Quick Check:

   Correct constructor with generics = KafkaProducer producer = new KafkaProducer<>(props); [OK]

Hint: Use new KafkaProducer<>(props) with key and value types [OK]

Common Mistakes:

• Omitting generics for key and value
• Using wrong constructor or method
• Not passing properties object

3. What will be the output of the following Java code snippet?

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer producer = new KafkaProducer<>(props);
ProducerRecord record = new ProducerRecord<>("my-topic", "key1", "value1");
producer.send(record);
System.out.println("Message sent");
producer.close();

medium

A. No output

B. Compilation error

C. Message sent

D. Runtime exception due to missing topic

Solution

1. Step 1: Analyze producer setup and send call

   The code sets up properties correctly, creates a producer, and sends a message to "my-topic" with key and value.

2. Step 2: Check output statement and close

   After sending, it prints "Message sent" and closes the producer. No errors expected if Kafka broker is running.

3. Final Answer:

   Message sent -> Option C

4. Quick Check:

   Successful send prints message [OK]

Hint: If send() succeeds, print statement runs [OK]

Common Mistakes:

• Assuming send() is synchronous and blocks
• Thinking missing topic causes error here
• Ignoring need for running Kafka broker

4. Identify the error in this Kafka Producer code snippet:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "StringSerializer");
props.put("value.serializer", "StringSerializer");
KafkaProducer producer = new KafkaProducer<>(props);
ProducerRecord record = new ProducerRecord<>("topic", "key", "value");
producer.send(record);
producer.close();

medium

A. Missing topic name in ProducerRecord

B. Properties object not initialized

C. Producer not closed properly

D. Incorrect serializer class names

Solution

1. Step 1: Check serializer class names in properties

   The serializer values must be full class names like "org.apache.kafka.common.serialization.StringSerializer".

2. Step 2: Verify other code parts

   Topic name is present, producer is closed properly, and properties object is initialized correctly.

3. Final Answer:

   Incorrect serializer class names -> Option D

4. Quick Check:

   Serializer class names must be full package names [OK]

Hint: Use full serializer class names with package [OK]

Common Mistakes:

• Using short class names without package
• Forgetting to close producer
• Leaving topic name empty

5. You want to send messages with keys to ensure ordering per key in Kafka. Which ProducerRecord constructor should you use?

hard

A. ProducerRecord(String topic, K key, V value)

B. ProducerRecord(String topic, V value)

C. ProducerRecord(K key, V value)

D. ProducerRecord(String topic)

Solution

1. Step 1: Understand ordering by key in Kafka

   Kafka guarantees ordering within a partition, and keys determine partition assignment.

2. Step 2: Identify correct ProducerRecord constructor

   To send with a key, use the constructor that includes topic, key, and value.

3. Final Answer:

   ProducerRecord(String topic, K key, V value) -> Option A

4. Quick Check:

   Use constructor with topic, key, and value for ordering [OK]

Hint: Include key in ProducerRecord to maintain order [OK]

Common Mistakes:

• Using constructor without key loses ordering
• Omitting topic causes errors
• Confusing key and value parameters