Блог

Я хочу иметь возможность использовать данные один раз от 2 разных потребителей (каждый потребитель будет получать разные сообщения) при использовании одинаковых topic и одинаковых group-id .

Производитель:

 producer = KafkaProducer(bootstrap_servers=['localhost:9092'],
                             value_serializer=lambda x:
                             dumps(x).encode('utf-8'))
    for e in range(50):
        data = {'number' : e}
        print('Producer {}'.format(data))
        producer.send('test', value=data)
        sleep(2)
  

Первый потребительский код:

 consumer = KafkaConsumer(
        'test',
        bootstrap_servers=['localhost:9092'],
        auto_offset_reset='earliest',
        enable_auto_commit=True,
        #auto_commit_interval_ms=100,
        group_id='my-group',
        value_deserializer=lambda x: loads(x.decode('utf-8')))

for message in consumer:
    message = message.value
    print('[1] Consume {}'.format(message))
    sleep(3)
  

Второй потребительский код:

 consumer = KafkaConsumer(
        'test',
        bootstrap_servers=['localhost:9092'],
        auto_offset_reset='earliest',
        enable_auto_commit=True,
        #auto_commit_interval_ms=100,
        group_id='my-group',
        value_deserializer=lambda x: loads(x.decode('utf-8')))

 for message in consumer:
        message = message.value
        print('[2] Consume {}'.format(message))
        sleep(5)
  

Я ожидаю увидеть, что некоторые сообщения используются, consumer-1 а другие сообщения consumer-2 (в соответствии с командой sleep в consumer коде)

Но кажется, что работает только один потребитель и получает все сообщения. (Первый потребитель застрял, а второй потребитель получает сообщения).

чего мне не хватает?

Из описания проблемы я предполагаю, что в вашей теме есть только один раздел.

Для вашей темы требуется более одного раздела, если вы хотите запустить более одного пользователя в группе