Как получить kafka customer-id для регистрации

Question

В моем приложении я использую spring-kafka для получения сообщения от сервера kafka, но от потребителя консоли я получаю consumer-id всех активных потоков потребителя

TOPIC            PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                    HOST            CLIENT-ID
easytest-events    9          247367          247367          0             p3-S14-0-e6a1d3cb-8ab3-435f-9f53-5081a6e8f812 /10.66.56.129   p3-S14-0

Есть ли способ получить consumer-id через код, чтобы я мог сравнить их

Answer 1

У меня также был похожий вариант использования, когда я хотел получить идентификатор потребителя для ведения журнала, поэтому я использовал имя текущего потока , и, похоже, он работает нормально.

LOGGER.info("Current thread: {}",Thread.currentThread().getName());

и это печатает: -

Current thread: org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1
Current thread: org.springframework.kafka.KafkaListenerEndpointContainer#0-2-C-1
Current thread: org.springframework.kafka.KafkaListenerEndpointContainer#0-1-C-1

это также ясно показывает, что сообщения каждый раз используются различными потоками (может быть в циклическом режиме)

Answer 2

Идентификатор потребителя выглядит как идентификатор клиента, к которому добавлен UUID - так что вы можете просто использовать идентификатор клиента (который вы можете установить по своему усмотрению). Spring добавит -0, -1 и т. Д.

Вы можете увидеть количество потоков в журналах при назначении разделов ...

2018-08-31 09:34:27.869  INFO 55748 --- [o52105744-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [so52105744-0]
2018-08-31 09:34:27.876  INFO 55748 --- [o52105744-2-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [so52105744-3]
2018-08-31 09:34:27.876  INFO 55748 --- [o52105744-1-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [so52105744-2]
2018-08-31 09:34:27.876  INFO 55748 --- [o52105744-9-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [so52105744-1]
2018-08-31 09:34:27.876  INFO 55748 --- [o52105744-3-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [so52105744-4]
2018-08-31 09:34:27.876  INFO 55748 --- [o52105744-6-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [so52105744-7]
2018-08-31 09:34:27.876  INFO 55748 --- [o52105744-5-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [so52105744-6]
2018-08-31 09:34:27.876  INFO 55748 --- [o52105744-4-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [so52105744-5]
2018-08-31 09:34:27.877  INFO 55748 --- [o52105744-7-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [so52105744-8]
2018-08-31 09:34:27.877  INFO 55748 --- [o52105744-8-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [so52105744-9]