def METHOD_NAME():
    global n
    codec = MessageCodec()
    consumer = create_consumer()
    producer = create_producer()
    consumer.subscribe(topics=["raw", "raw_ios"])
    print("Kafka consumer subscribed")
    escape = 0
    for msg in consumer:
        messages = codec.decode_detailed(msg.value)
        session_id = codec.decode_key(msg.key)
        if messages is None:
            print('-')
        for message in messages:
            send = False
            if isinstance(message, Fetch) or isinstance(message, FetchEvent):
                producer.send('quickwit-kafka', value=transform_fetch(message))
                print(f'added message {n} type Fetch')
                sleep(5)
            if isinstance(message, GraphQL):
                producer.send('quickwit-kafka', value=transform_graphql(message))
                print(f'added message {n} type GraphQL')
                sleep(5)
            if isinstance(message, PageEvent):
                producer.send('quickwit-kafka', value=transform_pageevent(message))
                print(f'added message {n} type PageEvent')
                sleep(5)