Amazon MSK in Kafka - Step-by-Step Execution

Process Flow - Amazon MSK

Create MSK Cluster

↓

Configure Topics

↓

Produce Messages

↓

MSK Brokers Receive

↓

Store Messages in Partitions

↓

Consume Messages

↓

Process Data

↓

End

This flow shows how Amazon MSK manages Kafka clusters: create cluster, configure topics, produce messages, store them, and consume for processing.

Execution Sample

Kafka

from kafka import KafkaProducer, KafkaConsumer

producer = KafkaProducer(bootstrap_servers='b-1.msk-cluster:9092')
producer.send('test-topic', b'Hello MSK')
producer.flush()

consumer = KafkaConsumer('test-topic', bootstrap_servers='b-1.msk-cluster:9092', value_deserializer=lambda m: m.decode('utf-8'))
for msg in consumer:
    print(msg.value)

This code sends a message to an MSK topic and then reads messages from it.

Process Table

Step	Action	Evaluation	Result
1	Create KafkaProducer with MSK bootstrap server	Connect to MSK brokers	Producer ready to send messages
2	Send message 'Hello MSK' to 'test-topic'	Message serialized and sent	Message queued in MSK topic partition
3	Create KafkaConsumer for 'test-topic'	Connect to MSK brokers	Consumer ready to receive messages
4	Consumer polls messages	Receives 'Hello MSK' message	Message available for processing
5	Print message value	Decode bytes to string	Output: Hello MSK
6	No more messages	Consumer waits or exits	End of execution

💡 No more messages to consume, consumer stops polling

Status Tracker

Variable	Start	After Step 2	After Step 4	Final
producer	None	KafkaProducer instance connected	KafkaProducer instance connected	KafkaProducer instance connected
consumer	None	None	KafkaConsumer instance connected	KafkaConsumer instance connected
message_sent	None	b'Hello MSK'	b'Hello MSK'	b'Hello MSK'
message_received	None	None	Hello MSK	Hello MSK

Key Moments - 3 Insights

Why does the consumer receive the message after the producer sends it?

Because the producer sends the message to the MSK topic partition, which stores it until the consumer polls and receives it, as shown in steps 2 and 4 of the execution_table.

What happens if the consumer tries to read before the producer sends a message?

The consumer will wait or poll repeatedly but receive no messages until the producer sends one, so no output appears until step 4 in the execution_table.

Why do we need to specify bootstrap_servers in both producer and consumer?

Because both need to connect to the MSK cluster brokers to send or receive messages, as shown in steps 1 and 3 where connections are established.

Visual Quiz - 3 Questions

Look at the execution_table, what is the output printed at step 5?

A b'Hello MSK'

B Hello MSK

C test-topic

D No output

💡 Hint

Check the 'Result' column at step 5 in the execution_table.

At which step does the consumer connect to the MSK brokers?

A Step 3

B Step 2

C Step 1

D Step 4

💡 Hint

Look at the 'Action' and 'Evaluation' columns for consumer creation in the execution_table.

If the producer did not send any message, what would the consumer receive at step 4?

A The last message sent

B An error

C No messages

D A default message

💡 Hint

Refer to the key_moments explanation about consumer behavior before messages are sent.

Concept Snapshot

Amazon MSK manages Kafka clusters in the cloud.
Create clusters, configure topics, produce and consume messages.
Producer sends messages to topic partitions.
Consumer reads messages from topics.
Both connect via bootstrap servers.
MSK handles storage and delivery reliably.

Full Transcript

Amazon MSK is a managed Kafka service that lets you create Kafka clusters easily. You configure topics where messages are sent and received. A producer connects to the MSK brokers and sends messages to a topic. These messages are stored in partitions inside the cluster. A consumer connects to the same brokers and polls messages from the topic partitions. When the consumer receives a message, it can process or print it. This flow ensures reliable message delivery and easy scaling. Both producer and consumer need to know the bootstrap servers to connect to the cluster. If no messages are sent, the consumer waits without output. This example shows sending 'Hello MSK' and receiving it back.

Practice

1. What is the main benefit of using Amazon MSK for Kafka users?

easy

A. It manages Kafka servers for you, so you don't have to.

B. It replaces Kafka with a different messaging system.

C. It requires you to manually configure all Kafka brokers.

D. It only works with AWS Lambda functions.

Solution

1. Step 1: Understand Amazon MSK's purpose

   Amazon MSK is a managed service that runs Kafka clusters for you.

2. Step 2: Compare options with MSK features

   Only It manages Kafka servers for you, so you don't have to. correctly states that MSK manages Kafka servers, removing manual server management.

3. Final Answer:

   It manages Kafka servers for you, so you don't have to. -> Option A

4. Quick Check:

   Managed Kafka service = It manages servers [OK]

Hint: Remember MSK means managed Kafka servers [OK]

Common Mistakes:

• Thinking MSK replaces Kafka
• Assuming manual server setup is needed
• Believing MSK only works with Lambda

2. Which of the following is the correct way to specify the bootstrap servers when connecting to an Amazon MSK cluster in Kafka client configuration?

easy

A. "localhost:9092"

B. "msk://cluster-endpoint"

C. "127.0.0.1:2181"

D. "b-1.mskcluster.abcd1234.kafka.us-east-1.amazonaws.com:9092,b-2.mskcluster.abcd1234.kafka.us-east-1.amazonaws.com:9092"

Solution

1. Step 1: Identify MSK cluster endpoints format

   Amazon MSK provides bootstrap servers as DNS names with ports, like in "b-1.mskcluster.abcd1234.kafka.us-east-1.amazonaws.com:9092,b-2.mskcluster.abcd1234.kafka.us-east-1.amazonaws.com:9092".

2. Step 2: Eliminate incorrect formats

   "localhost:9092" is localhost, not MSK. "127.0.0.1:2181" is Zookeeper port, not Kafka. "msk://cluster-endpoint" is not a valid Kafka bootstrap server format.

3. Final Answer:

   "b-1.mskcluster.abcd1234.kafka.us-east-1.amazonaws.com:9092,b-2.mskcluster.abcd1234.kafka.us-east-1.amazonaws.com:9092" -> Option D

4. Quick Check:

   MSK bootstrap servers = DNS names with ports [OK]

Hint: MSK bootstrap servers are DNS names with port 9092 [OK]

Common Mistakes:

• Using localhost instead of MSK endpoints
• Confusing Zookeeper port with Kafka port
• Using invalid URI schemes like msk://

3. Given the following Kafka producer code snippet connecting to Amazon MSK, what will be the output if the message is sent successfully?

producer.send(new ProducerRecord<>("my-topic", "key1", "Hello MSK"));
System.out.println("Message sent successfully");

medium

A. Error: Connection refused

B. Message sent successfully

C. No output

D. Message failed to send

Solution

1. Step 1: Understand producer.send behavior

   The send method is asynchronous but the print statement runs immediately after.

2. Step 2: Check output after send call

   Since no error handling is shown and send is called, the print statement will output "Message sent successfully".

3. Final Answer:

   Message sent successfully -> Option B

4. Quick Check:

   Producer send + print = Message sent successfully [OK]

Hint: Print after send shows success unless exception thrown [OK]

Common Mistakes:

• Assuming send blocks until confirmed
• Expecting error without exception handling
• Thinking no output means success

4. You have this Kafka consumer code connecting to Amazon MSK:

Properties props = new Properties();
props.put("bootstrap.servers", "b-1.mskcluster.kafka:9092");
props.put("group.id", "test-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("my-topic"));
consumer.poll(Duration.ofMillis(100));

The consumer throws a runtime error about deserializer class not found. What is the likely cause?

medium

A. The topic name is invalid.

B. The bootstrap server address is wrong.

C. The deserializer class names are incorrect or misspelled.

D. The group.id property is missing.

Solution

1. Step 1: Analyze the error message

   The error says deserializer class not found, so the problem is with deserializer class names.

2. Step 2: Check deserializer class names in code

   They must be fully qualified and spelled exactly. Any typo causes class not found error.

3. Final Answer:

   The deserializer class names are incorrect or misspelled. -> Option C

4. Quick Check:

   Deserializer class error = wrong class name [OK]

Hint: Check spelling of deserializer class names first [OK]

Common Mistakes:

• Ignoring class name typos
• Assuming bootstrap server causes deserializer error
• Missing group.id but error is about deserializer

5. You want to create a Kafka consumer in Java that connects to an Amazon MSK cluster and only processes messages with keys starting with "user-". Which approach correctly applies this filter after receiving messages?

hard

A. Use a Kafka Streams application with a filter on the key prefix.

B. Set the consumer property "key.filter.prefix" to "user-".

C. Subscribe only to topics named "user-*".

D. Configure the producer to send only keys starting with "user-".

Solution

1. Step 1: Understand Kafka consumer filtering options

   Kafka consumers do not support filtering by key prefix via configuration properties.

2. Step 2: Identify correct filtering method

   Kafka Streams allows processing and filtering messages by key prefix programmatically.

3. Step 3: Evaluate other options

   There is no Kafka consumer property like "key.filter.prefix". Subscribing to topics named "user-*" filters by topic name, not message keys. Producer configuration controls what is sent, not what the consumer processes.

4. Final Answer:

   Use a Kafka Streams application with a filter on the key prefix. -> Option A

5. Quick Check:

   Filter keys after consuming = Kafka Streams filter [OK]

Hint: Filter keys in Kafka Streams, not consumer config [OK]

Common Mistakes:

• Expecting consumer config to filter keys
• Filtering by topic name instead of message key
• Trying to filter on producer side only