Schema Registry usage example

import requests
import json

init_string = 'data: '
source_url = 'https://stream.wikimedia.org/v2/stream/recentchange'

def avro_producer(source_url):
    s = requests.Session()

    with s.get(source_url, headers=None, stream=True) as resp:
        times = 10
        for line in resp.iter_lines():
            if line:
                decoded_line = line.decode()
                if decoded_line.find(init_string) >= 0:
                    print(str(11 - times) + ")")
                    # remove data: to create a valid json
                    decoded_line = decoded_line.replace(init_string, "")
                    # convert to json
                    decoded_json = json.loads(decoded_line)
                    print(decoded_json)
                    times -= 1
            if times <= 0: break

avro_producer(source_url)

import requests
import json

# 3rd party library imported
from confluent_kafka import SerializingProducer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka.schema_registry import Schema

# import from constants
from constants import SCHEMA_STR

init_string = 'data: '
source_url = 'https://stream.wikimedia.org/v2/stream/recentchange'
kafka_url = 'localhost:9092'
schema_registry_url = 'http://localhost:8081'
kafka_topic = 'wikimedia'
schema_registry_subject = f"{kafka_topic}-value"

def delivery_report(errmsg, msg):
    if errmsg is not None:
        print("Delivery failed for Message: {} : {}".format(msg.key(), errmsg))
        return
    print('Message: {} unsuccessfully produced to Topic: {} Partition: [{}] at offset {}'.format(msg.key(), msg.topic(), msg.partition(), msg.offset()))

def avro_producer(source_url, kafka_url, schema_registry_url, schema_registry_subject):
    # schema registry
    sr, latest_version = get_schema_from_schema_registry(schema_registry_url, schema_registry_subject)


    value_avro_serializer = AvroSerializer(schema_registry_client = sr,
                                          schema_str = latest_version.schema.schema_str,
                                          conf={
                                              'auto.register.schemas': False
                                            }
                                          )

    # Kafka Producer
    producer = SerializingProducer({
        'bootstrap.servers': kafka_url,
        'security.protocol': 'plaintext',
        'value.serializer': value_avro_serializer,
        'delivery.timeout.ms': 120000, # set it to 2 mins
        'enable.idempotence': 'true'
    })

    s = requests.Session()

    with s.get(source_url, headers=None, stream=True) as resp:
        for line in resp.iter_lines():
            if line:
                decoded_line = line.decode()
                if decoded_line.find(init_string) >= 0:
                    # remove data: to create a valid json
                    decoded_line = decoded_line.replace(init_string, "")
                    # convert to json
                    decoded_json = json.loads(decoded_line)

                    try:
                        # print(decoded_line + '\n')
                        producer.produce(topic=kafka_topic, value=decoded_json, on_delivery=delivery_report)

                        # Trigger any available delivery report callbacks from previous produce() calls
                        events_processed = producer.poll(1)
                        print(f"events_processed: {events_processed}")

                        messages_in_queue = producer.flush(1)
                        print(f"messages_in_queue: {messages_in_queue}")
                    except Exception as e:
                        print(e)


def get_schema_from_schema_registry(schema_registry_url, schema_registry_subject):
    sr = SchemaRegistryClient({'url': schema_registry_url})
    latest_version = sr.get_latest_version(schema_registry_subject)

    return sr, latest_version

def register_schema(schema_registry_url, schema_registry_subject, schema_str):
    sr = SchemaRegistryClient({'url': schema_registry_url})
    schema = Schema(schema_str, schema_type="AVRO")
    schema_id = sr.register_schema(subject_name=schema_registry_subject, schema=schema)

    return schema_id

def update_schema(schema_registry_url, schema_registry_subject, schema_str):
    sr = SchemaRegistryClient({'url': schema_registry_url})
    versions_deleted_list = sr.delete_subject(schema_registry_subject)
    print(f"versions of schema deleted list: {versions_deleted_list}")

    schema_id = register_schema(schema_registry_url, schema_registry_subject, schema_str)
    return schema_id


avro_producer(source_url, kafka_url, schema_registry_url, schema_registry_subject)

import json
from confluent_kafka import Producer, KafkaException
from confluent_kafka.avro import AvroConsumer
from confluent_kafka import avro

def avro_consumer():
    consumer_config = {
        "bootstrap.servers": "localhost:9092",
        "schema.registry.url": "http://localhost:8081",
        "group.id": "1",
        "auto.offset.reset": "earliest"
    }

    #print(value_schema)

    consumer = AvroConsumer(consumer_config)
    consumer.subscribe(['wikimedia'])

    while True:
      try:
        msg = consumer.poll(1)

        if msg is None:
          continue

        print("Value is :" + json.dumps(msg.value()))
        print("-------------------------")

      except KafkaException as e:
        print('Kafka failure ' + e)

    consumer.close()

def main():
    avro_consumer()

if __name__ == "__main__":
    main()