Async Communication with SQS and Kafka — Message-Driven Microservices
In this tutorial, you will learn about Async Communication with SQS and Kafka. We cover key concepts, practical examples, and best practices to help you master this topic.
AWS SQS and Apache Kafka are leading message brokers for async communication, with SQS offering simple managed queues on AWS and Kafka providing high-throughput distributed event streaming with strong ordering guarantees.
What You'll Learn
By the end of this lesson you will compare SQS and Kafka, implement SQS producers and consumers, configure Kafka topics and partitions, design consumer groups for parallel processing, handle message failures, and choose the right broker for your use case.
Why It Matters
Choosing the right async messaging system directly impacts scalability, ordering guarantees, and operational complexity. SQS is simple and Serverless, ideal for straightforward task queues. Kafka handles millions of messages per second with strong ordering, suited for event streaming and Data Pipelines.
Real-World Use
DodaZIP uses both SQS and Kafka for different purposes. SQS handles file compression jobs (simple work queues), while Kafka powers the event-driven audit log, streaming file processing status to analytics, monitoring, and Compliance services simultaneously.
flowchart LR
A[SQS: Work Queue] -->|1 job to 1 worker| B[Worker Pool]
C[Kafka: Event Stream] -->|1 event to many consumers| D[Service A]
C --> E[Service B]
C --> F[Service C]
subgraph AWS
A
end
subgraph Self-Managed
C
end
style A fill:#2d3748,color:#fff
style C fill:#2d3748,color:#fff
SQS Implementation
Simple Queue Service for work queues.
# sqs_implementation.py
# AWS SQS producer and consumer
def sqs_impl():
print("AWS SQS Implementation")
print("=" * 40)
print()
producer_code = """
import boto3
import json
# Initialize SQS client
sqs = boto3.client('sqs', region_name='us-east-1')
def send_compression_job(file_id, format_type):
"""Send a message to the compression job queue."""
queue_url = sqs.get_queue_url(
QueueName='compression-jobs'
)['QueueUrl']
response = sqs.send_message(
QueueUrl=queue_url,
MessageBody=json.dumps({
'file_id': file_id,
'format': format_type,
'priority': 'normal'
}),
MessageAttributes={
'file_id': {
'DataType': 'String',
'StringValue': file_id
},
'priority': {
'DataType': 'String',
'StringValue': 'normal'
}
},
MessageGroupId='compression', # For FIFO queues
MessageDeduplicationId=file_id # For FIFO queues
)
print(f"Sent message: {response['MessageId']}")
return response['MessageId']
# Send a job
send_compression_job('file-123', 'zip')
"""
print("SQS Producer:")
print(producer_code)
consumer_code = """
import boto3
import json
import time
sqs = boto3.client('sqs', region_name='us-east-1')
queue_url = sqs.get_queue_url(QueueName='compression-jobs')['QueueUrl']
def process_messages():
"""Long-poll for messages and process them."""
while True:
response = sqs.receive_message(
QueueUrl=queue_url,
MaxNumberOfMessages=10, # Batch up to 10
WaitTimeSeconds=20, # Long polling
VisibilityTimeout=300, # 5 min to process
MessageAttributeNames=['All']
)
messages = response.get('Messages', [])
for message in messages:
try:
body = json.loads(message['Body'])
print(f"Processing: {body}")
# Simulate work
time.sleep(2)
# Delete after successful processing
sqs.delete_message(
QueueUrl=queue_url,
ReceiptHandle=message['ReceiptHandle']
)
print(f"Completed: {body['file_id']}")
except Exception as e:
print(f"Failed: {e}")
# Message becomes visible again after VisibilityTimeout
# Start consuming
process_messages()
"""
print("SQS Consumer:")
print(consumer_code)
sqs_impl()
Kafka Implementation
Apache Kafka for event streaming.
# kafka_implementation.py
# Kafka producer and consumer
def kafka_impl():
print("Apache Kafka Implementation")
print("=" * 40)
print()
producer_code = """
from kafka import KafkaProducer
import json
# Kafka producer
producer = KafkaProducer(
bootstrap_servers=['kafka-1:9092', 'kafka-2:9092'],
value_serializer=lambda v: json.dumps(v).encode('utf-8'),
acks='all', # Wait for all replicas to acknowledge
retries=3, # Retry on transient errors
batch_size=16384 # Batch messages for efficiency
)
def publish_file_event(event_type, file_data):
"""Publish a file event to Kafka."""
future = producer.send(
'file-events',
value={
'event_type': event_type,
'file_id': file_data['file_id'],
'user_id': file_data['user_id'],
'timestamp': file_data['timestamp'],
'metadata': file_data.get('metadata', {})
},
key=file_data['file_id'].encode(), # Same key = same partition
partition=None # Auto-select based on key
)
# Wait for acknowledgment
record_metadata = future.get(timeout=10)
print(f"Published to topic {record_metadata.topic} "
f"partition {record_metadata.partition} "
f"offset {record_metadata.offset}")
return record_metadata
# Publish events
publish_file_event('file.uploaded', {
'file_id': 'file-456',
'user_id': 'user-789',
'timestamp': '2026-06-28T10:30:00Z',
'metadata': {'size': 1024000, 'type': 'pdf'}
})
"""
print("Kafka Producer:")
print(producer_code)
consumer_code = """
from kafka import KafkaConsumer
import json
# Kafka consumer with consumer group
consumer = KafkaConsumer(
'file-events',
bootstrap_servers=['kafka-1:9092', 'kafka-2:9092'],
group_id='virus-scanner-group', # Consumer group
auto_offset_reset='earliest', # Start from beginning if new
enable_auto_commit=True, # Auto-commit offsets
auto_commit_interval_ms=5000, # Commit every 5s
value_deserializer=lambda m: json.loads(m.decode('utf-8')),
max_poll_records=100 # Max records per poll
)
def consume_events():
print("Listening for file events...")
for message in consumer:
event = message.value
print(f"Received: {event['event_type']} "
f"(partition: {message.partition}, "
f"offset: {message.offset})")
try:
process_event(event)
# Offset auto-committed after poll
except Exception as e:
print(f"Error processing event: {e}")
# Offset not committed, message will be redelivered
def process_event(event):
if event['event_type'] == 'file.uploaded':
scan_for_viruses(event['file_id'])
elif event['event_type'] == 'file.compressed':
update_compression_status(event['file_id'])
# Start consumer
consume_events()
"""
print("Kafka Consumer:")
print(consumer_code)
kafka_impl()
SQS vs Kafka
Detailed comparison for decision making.
# sqs_vs_kafka.py
# SQS vs Kafka comparison
def compare_sqs_kafka():
print("SQS vs Kafka Comparison")
print("=" * 40)
print()
comparisons = [
("Deployment", "Fully managed (AWS)", "Self-managed or MSK/Confluent"),
("Ordering", "FIFO queues (within group)", "Per-partition (strong)"),
("Throughput", "Unlimited (auto-scaling)", "Millions msg/sec (partitioned)"),
("Retention", "Up to 14 days", "Configurable (days to indefinite)"),
("Consumer Model", "Pull (1 msg to 1 consumer)", "Pull (1 msg to 1 consumer group)"),
("Broadcast", "SNS + SQS fan-out", "Multiple consumer groups"),
("Exactly-Once", "FIFO queues", "Idempotent producer + transactional"),
("Latency", "~100ms (standard)", "~10ms (producer ack=1)"),
("Message Size", "Up to 256KB", "Up to 1MB (default)"),
("Cost Model", "Pay per request", "Pay per broker hour + storage"),
]
print(f"{'Aspect':25s} {'SQS':40s} {'Kafka':40s}")
print("-" * 105)
for aspect, sqs, kafka in comparisons:
print(f"{aspect:25s} {sqs:40s} {kafka:40s}")
compare_sqs_kafka()
Consumer Group Pattern
Parallel processing with Kafka consumer groups.
# consumer_groups.py
# Kafka consumer group patterns
def consumer_groups():
print("Kafka Consumer Group Patterns")
print("=" * 40)
print()
code = """
# Consumer Group: order-events-group
# Topic: order-events (6 partitions)
# 3 consumers in the group
# Partition Assignment:
# Consumer 1: partitions 0, 3
# Consumer 2: partitions 1, 4
# Consumer 3: partitions 2, 5
# Each message goes to exactly one consumer in the group.
# Messages with the same key go to the same partition
# and thus the same consumer.
# Scaling:
# - Max 6 consumers (one per partition) for this topic
# - Adding consumers up to 6 increases parallelism
# - Adding more than 6: some consumers idle
# Rebalancing:
# When a consumer joins or leaves, Kafka rebalances
# partition assignments across remaining consumers.
# During rebalancing, no messages are processed.
def consumer_rebalance_listener():
from kafka import ConsumerRebalanceListener
class RebalanceListener(ConsumerRebalanceListener):
def on_partitions_revoked(self, revoked):
print(f"Partitions revoked: {revoked}")
# Commit offsets before rebalance
consumer.commit()
def on_partitions_assigned(self, assigned):
print(f"Partitions assigned: {assigned}")
consumer = KafkaConsumer(
'order-events',
group_id='order-processors',
bootstrap_servers=['kafka:9092'],
value_deserializer=lambda m: json.loads(m),
enable_auto_commit=False # Manual commit
)
consumer.subscribe(
['order-events'],
listener=RebalanceListener()
)
"""
print(code)
consumer_groups()
Common Mistakes
Not handling poison pills: A malformed message that crashes the consumer is retried infinitely. Implement dead letter queues (SQS) or send to a separate error topic (Kafka) after N retries.
Ignoring message ordering requirements: SQS Standard does not guarantee ordering. Use FIFO queues if order matters. Kafka guarantees per-partition ordering only.
Auto-commit without processing confirmation: Auto-committing offsets before processing completes can lose messages if the consumer crashes. Commit after processing.
Creating too many partitions in Kafka: More partitions means more parallelism but also more rebalancing overhead and file handles. Start with partition count = expected max consumers * 2.
Using SQS for event broadcast: SQS is a queue (one message, one consumer). Use SNS fan-out to multiple SQS queues, or use Kafka for native broadcast.
Practice Questions
What is the main difference between SQS and Kafka? SQS is a managed queue service for work distribution. Kafka is a distributed event streaming platform for high-throughput pub/sub.
How does Kafka achieve ordering? Kafka guarantees ordering within a partition. Messages with the same key go to the same partition, preserving their order.
What is a consumer group in Kafka? A group of consumers that coordinate to consume partitions of a topic. Each partition is consumed by exactly one consumer in the group.
What is long polling in SQS? Instead of immediately returning empty responses, SQS waits up to 20 seconds for messages to arrive, reducing empty responses and costs.
Challenge: Design an async messaging system for an e-commerce platform. Use SQS for order processing queue, Kafka for event-driven analytics and notification broadcast. Define the topics, queues, consumer groups, error handling, and ordering requirements.
FAQ
Mini Project
Design an async messaging architecture for a video processing pipeline. Use SQS for transcoding job queues (each job goes to one worker). Use Kafka for pipeline events (video.uploaded, transcoding.completed, thumbnail.generated) consumed by notification, analytics, and CDN invalidation services. Include dead letter handling.
def video_pipeline_async():
print("Video Processing - Async Messaging Design")
print("=" * 45)
print()
print("SQS Queues:")
print(" transcoding-jobs - Work queue for transcode workers")
print(" thumbnail-jobs - Work queue for thumbnail workers")
print(" transcoding-jobs-dlq - Dead letter queue")
print()
print("Kafka Topics:")
print(" video.events - All pipeline events")
print(" Partitions: 6")
print(" Retention: 7 days")
print()
print("Consumer Groups:")
print(" notification-svc - Sends email/SMS")
print(" analytics-svc - Tracks metrics")
print(" cdn-invalidation-svc - Purges CDN cache")
print(" audit-svc - Stores audit log")
print()
print("Flow:")
print(" Upload -> SQS transcode job -> Worker transcodes")
print(" -> Kafka video.events -> 4 consumer groups")
print(" SQS thumbnail job -> Worker generates thumb")
print(" -> Kafka video.events -> same 4 groups")
print()
print("Error Handling:")
print(" SQS: 5 retries then DLQ")
print(" Kafka: Retry topic, then error topic")
video_pipeline_async()
What's Next
Next: Event Bus Communication for event bus patterns.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro