Kafka vs Kinesis vs Pub/Sub: how they differ and which to choose

import json
from confluent_kafka import Consumer, Producer
from confluent_kafka.admin import AdminClient, NewTopic
from orders import ORDERS

BROKERS = "localhost:9092"
admin = AdminClient({"bootstrap.servers": BROKERS})
admin.create_topics([NewTopic("orders", num_partitions=3, replication_factor=1)])["orders"].result()

# The key picks the partition: one country's orders stay in order.
producer = Producer({"bootstrap.servers": BROKERS, "enable.idempotence": True})
for order in ORDERS:
    producer.produce("orders", key=order["country"], value=json.dumps(order))
producer.flush()

# A consumer group: Kafka stores how far "revenue" has read, per partition.
consumer = Consumer({
    "bootstrap.servers": BROKERS,
    "group.id": "revenue",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,
})
consumer.subscribe(["orders"])
revenue, seen = {}, 0
while seen < len(ORDERS):
    msg = consumer.poll(5.0)
    if msg is None or msg.error():
        continue
    order = json.loads(msg.value())
    revenue[order["country"]] = revenue.get(order["country"], 0) + order["amount"]
    seen += 1
    consumer.commit(msg)  # after the work, so a crash re-reads rather than skips
consumer.close()
print(dict(sorted(revenue.items())))

The key decides the partition, so the 3 German orders landed in 1 partition, in order. The consumer group revenue commits its position after each order is counted, so a crash reads an order again rather than skipping it, and a second group could read the whole topic again from the start. Run with confluent-kafka 2.16.

import json
import boto3
from orders import ORDERS

kinesis = boto3.client("kinesis", region_name="eu-central-1")
kinesis.create_stream(StreamName="orders", StreamModeDetails={"StreamMode": "ON_DEMAND"})
kinesis.get_waiter("stream_exists").wait(StreamName="orders")

# The partition key picks the shard: one country's orders stay in order.
kinesis.put_records(StreamName="orders", Records=[
    {"PartitionKey": order["country"], "Data": json.dumps(order).encode()} for order in ORDERS
])

# Read every shard from the oldest record still kept. A real consumer uses the
# Kinesis Client Library, which keeps each shard's checkpoint in DynamoDB.
revenue = {}
for shard in kinesis.list_shards(StreamName="orders")["Shards"]:
    iterator = kinesis.get_shard_iterator(
        StreamName="orders", ShardId=shard["ShardId"], ShardIteratorType="TRIM_HORIZON"
    )["ShardIterator"]
    for record in kinesis.get_records(ShardIterator=iterator)["Records"]:
        order = json.loads(record["Data"])
        revenue[order["country"]] = revenue.get(order["country"], 0) + order["amount"]
print(dict(sorted(revenue.items())))

An on-demand stream needs no shard count; the mock started this one with 4. The partition key places each order on a shard. This reader walks every shard itself; in production the Kinesis Client Library does that, sharing shards between workers and keeping each shard's checkpoint in DynamoDB. Run with boto3 1.43.

import json
from google.cloud import pubsub_v1
from orders import ORDERS

PROJECT = "shop-local"
publisher = pubsub_v1.PublisherClient(
    publisher_options=pubsub_v1.types.PublisherOptions(enable_message_ordering=True)
)
subscriber = pubsub_v1.SubscriberClient()
topic = publisher.create_topic(name=publisher.topic_path(PROJECT, "orders")).name
# A subscription is a copy of the topic for one reader. Create it before publishing:
# it only receives messages published after it exists.
subscription = subscriber.create_subscription(request={
    "name": subscriber.subscription_path(PROJECT, "revenue"),
    "topic": topic,
    "enable_message_ordering": True,
}).name

# The ordering key keeps one country's orders in order; without one, there is no order.
for order in ORDERS:
    publisher.publish(topic, json.dumps(order).encode(), ordering_key=order["country"]).result()

revenue, seen = {}, 0
while seen < len(ORDERS):
    response = subscriber.pull(request={"subscription": subscription, "max_messages": 10})
    for received in response.received_messages:
        order = json.loads(received.message.data)
        revenue[order["country"]] = revenue.get(order["country"], 0) + order["amount"]
        seen += 1
    # Acknowledge after the work; a message not acknowledged in time is delivered again.
    if response.received_messages:
        subscriber.acknowledge(request={
            "subscription": subscription,
            "ack_ids": [m.ack_id for m in response.received_messages],
        })
print(dict(sorted(revenue.items())))

No partitions or shards anywhere. The subscription is made before publishing because it only receives messages published after it exists, and ordering is switched on twice, on the publisher and on the subscription. A message that is not acknowledged in time is delivered again. Run with google-cloud-pubsub 2.43 against the emulator (PUBSUB_EMULATOR_HOST).