diff --git a/kafka-demos/python-schema-registry/.gitignore b/kafka-demos/python-schema-registry/.gitignore new file mode 100644 index 00000000..21d0b898 --- /dev/null +++ b/kafka-demos/python-schema-registry/.gitignore @@ -0,0 +1 @@ +.venv/ diff --git a/kafka-demos/python-schema-registry/client.py b/kafka-demos/python-schema-registry/client.py new file mode 100644 index 00000000..66c1573c --- /dev/null +++ b/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': ':' +# 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() \ No newline at end of file diff --git a/kafka-demos/python-schema-registry/docker-compose.yml b/kafka-demos/python-schema-registry/docker-compose.yml new file mode 100644 index 00000000..97cb0016 --- /dev/null +++ b/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: +