import threading
from confluent_kafka import Consumer, KafkaException, KafkaError

def consume_messages():
    # Настройка конфигурации потребителя
    conf = {
        'bootstrap.servers': 'localhost:9092',
        'group.id': 'my_group',
        'auto.offset.reset': 'earliest'
    }

    # Создание потребителя
    consumer = Consumer(conf)

    # Подписка на топик
    consumer.subscribe(['my_topic'])

    try:
        while True:
            # Получение сообщения
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    # Достигнут конец раздела
                    continue
                else:
                    raise KafkaException(msg.error())
            # Обработка сообщения
            print(f'Received message: {msg.value().decode("utf-8")}')
    except KeyboardInterrupt:
        pass
    finally:
        # Закрытие потребителя
        consumer.close()

def start_consumer_thread():
    consumer_thread = threading.Thread(target=consume_messages)
    consumer_thread.daemon = True
    consumer_thread.start()

if __name__ == "__main__":
    start_consumer_thread()
    # Основной поток может выполнять другие задачи
    while True:
        pass