add kafka avro demo with python

kafka-demos/python-schema-registry/.gitignore

@@ -0,0 +1 @@
+.venv/

kafka-demos/python-schema-registry/client.py

@@ -0,0 +1,134 @@
+import argparse
+from confluent_kafka import Producer, Consumer, KafkaException
+from confluent_kafka.schema_registry import SchemaRegistryClient, header_schema_id_serializer
+from confluent_kafka.schema_registry.avro import AvroSerializer, AvroDeserializer
+from confluent_kafka.serialization import SerializationContext, MessageField
+
+DEFAULT_SCHEMA_REGISTRY_URL = 'http://localhost:8081'
+# For Confluent Cloud, add: 'basic.auth.user.info': '<sr-api-key>:<sr-api-secret>'
+# See: https://docs.confluent.io/cloud/current/sr/index.html
+
+USER_SCHEMA_STR = """
+{
+  "namespace": "com.example",
+  "type": "record",
+  "name": "User",
+  "fields": [
+    {"name": "name", "type": "string"},
+    {"name": "age", "type": "int"}
+  ]
+}
+"""
+
+class User:
+    def __init__(self, name, age):
+        self.name = name
+        self.age = age
+
+    def to_dict(self):
+        return {"name": self.name, "age": self.age}
+
+
+def run_producer(topic, bootstrap_servers, schema_registry_url):
+    schema_registry_client = SchemaRegistryClient({'url': schema_registry_url})
+    avro_serializer = AvroSerializer(
+        schema_registry_client,
+        USER_SCHEMA_STR,
+        lambda user, ctx: user.to_dict(),
+        conf={"schema.id.serializer": header_schema_id_serializer, 'subject.name.strategy.type': 'TOPIC'}
+    )
+    producer = Producer({'bootstrap.servers': bootstrap_servers })
+
+    print(f"Producing to topic '{topic}'. Enter 'name,age' per line. Press Ctrl+C to stop.")
+    try:
+        while True:
+            line = input("> ").strip()
+            if not line:
+                continue
+            parts = line.split(",", 1)
+            if len(parts) != 2:
+                print("Invalid input - expected format: name,age")
+                continue
+            name, raw_age = parts[0].strip(), parts[1].strip()
+            try:
+                age = int(raw_age)
+            except ValueError:
+                print(f"Invalid age '{raw_age}' - must be an integer")
+                continue
+            user = User(name=name, age=age)
+            headers = []
+            value = avro_serializer(user, SerializationContext(topic, MessageField.VALUE, headers=headers))
+            producer.produce(topic, key=name, value=value, headers=headers)
+            producer.flush()
+            print(f"Produced: key={name} value={user.to_dict()}")
+    except KeyboardInterrupt:
+        print("\nStopped.")
+
+
+def run_consumer(topic, bootstrap_servers, schema_registry_url):
+    schema_registry_client = SchemaRegistryClient({'url': schema_registry_url})
+    avro_deserializer = AvroDeserializer(
+        schema_registry_client,
+        USER_SCHEMA_STR,
+        lambda obj, ctx: User(name=obj["name"], age=obj["age"]),
+    )
+    consumer = Consumer({
+        'bootstrap.servers': bootstrap_servers,
+        'group.id': 'python-schema-registry-demo',
+        'auto.offset.reset': 'earliest',
+    })
+    consumer.subscribe([topic])
+    print(f"Consuming from topic '{topic}'. Press Ctrl+C to stop.")
+
+    try:
+        while True:
+            msg = consumer.poll(timeout=1.0)
+            if msg is None:
+                continue
+            if msg.error():
+                raise KafkaException(msg.error())
+            headers = msg.headers() or []
+            user = avro_deserializer(msg.value(), SerializationContext(topic, MessageField.VALUE, headers=headers))
+            print(f"Consumed: key={msg.key().decode()} value={user.to_dict()}")
+    except KeyboardInterrupt:
+        print("\nStopped.")
+    finally:
+        consumer.close()
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Avro Kafka client with Schema Registry")
+    parser.add_argument(
+        "--role",
+        required=True,
+        choices=["PRODUCER", "CONSUMER"],
+        help="Run as PRODUCER or CONSUMER",
+    )
+    parser.add_argument(
+        "--topic",
+        default="users",
+        help="Kafka topic (default: users)",
+    )
+    parser.add_argument(
+        "--bootstrap-server",
+        default="localhost:9092",
+        dest="bootstrap_server",
+        help="Kafka bootstrap server (default: localhost:9092)",
+    )
+    parser.add_argument(
+        "--schema-registry",
+        default=DEFAULT_SCHEMA_REGISTRY_URL,
+        dest="schema_registry",
+        help=f"Schema Registry URL (default: {DEFAULT_SCHEMA_REGISTRY_URL})",
+    )
+
+    args = parser.parse_args()
+
+    if args.role == "PRODUCER":
+        run_producer(args.topic, args.bootstrap_server, args.schema_registry)
+    else:
+        run_consumer(args.topic, args.bootstrap_server, args.schema_registry)
+
+
+if __name__ == "__main__":
+    main()

kafka-demos/python-schema-registry/docker-compose.yml

@@ -0,0 +1,97 @@
+---
+services:
+  broker:
+    image: apache/kafka:3.8.0
+    hostname: broker
+    user: "0:0"
+    volumes:
+      - kafka-data:/tmp/kraft-combined-logs
+    ports:
+      - 9092:9092
+      - 29092:29092
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CONTROLLER:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: true
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_PROCESS_ROLES: broker,controller
+      KAFKA_NODE_ID: 1
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:29093
+      KAFKA_LISTENERS: PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
+      # flink parallelism should not be higher than consumed topic partition count
+      # we want to test locally with flink parallelism >= 2, so let's bump default partition count
+      KAFKA_NUM_PARTITIONS: 3
+      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
+    healthcheck:
+      test: ["CMD","/opt/kafka/bin/kafka-broker-api-versions.sh", "--bootstrap-server", "localhost:9092"]
+      interval: 30s
+      timeout: 10s
+      retries: 5
+
+  kafkasetup:
+    depends_on:
+      - broker
+    image: apache/kafka:3.8.0
+    restart: on-failure
+    # Put your initialization (e.g. topic creation) in this script. You have access to all Kafka cli utilities.
+    entrypoint:
+      - /bin/bash
+      - -c
+      - |
+        # Topic creation example:
+        # /opt/kafka/bin/kafka-topics.sh --bootstrap-server broker:29092 --create --topic my-topic --partitions 1 --replication-factor 1 --config retention.ms=-1
+        echo "Kafka topics created successfully."
+
+  schema-registry:
+    image: confluentinc/cp-schema-registry:8.2.0
+    hostname: schema-registry
+    container_name: schema-registry
+    depends_on:
+      - broker
+    ports:
+      - "8081:8081"
+    environment:
+      SCHEMA_REGISTRY_HOST_NAME: schema-registry
+      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092'
+      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
+
+    # confluent compatibility api is accessible at /apis/ccompat/v8
+  apicurio-registry:
+    depends_on:
+      - broker
+    image: apicurio/apicurio-registry:3.2.3
+    environment:
+      QUARKUS_PROFILE: "prod"
+      APICURIO_KAFKA_COMMON_BOOTSTRAP_SERVERS: "broker:29092"
+      APICURIO_CCOMPAT_LEGACY_ID_MODE_ENABLED: "true"
+      APICURIO_STORAGE_KIND: "kafkasql"
+    ports:
+      - 9081:8080
+
+  kafbat-ui:
+    container_name: kafbat-ui
+    image: ghcr.io/kafbat/kafka-ui:latest
+    ports:
+      - 8082:8080
+    depends_on:
+      - broker
+      - schema-registry
+      - apicurio-registry
+    environment:
+      KAFKA_CLUSTERS_0_NAME: local-confluent
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: broker:29092
+      KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://schema-registry:8081
+      KAFKA_CLUSTERS_1_NAME: local-apicurio
+      KAFKA_CLUSTERS_1_BOOTSTRAPSERVERS: broker:29092
+      KAFKA_CLUSTERS_1_SCHEMAREGISTRY: http://apicurio-registry:8080/apis/ccompat/v8
+
+volumes:
+  kafka-data:
+