Кафка Потребительская задержка

Question

У меня следующая проблема:

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

Я использую kafka-client версия 0.11.0.0.Смена версии возможна, но утомительна, поэтому я хочу избегать изменения версий в максимально возможной степени.

Мой потребительский конфиг выглядит следующим образом:

group.id= unique-group-d
bootstrap.servers = # kafka broker with version 0.11.0.0
auto.offset.reset = latest
# avoid client id collision with Kafka
client.id= some-example-gateway-client

И я опрашиваю сообщения вроде этого:

ConsumerRecords<String, String> tConsumerRecords = tempConsumer.poll(1000);
tempMessage.forEach(entry -> {
    [...]
});

Не существует 100% конфликтов имен сидентификатор клиента или группы.В теме, на которую я подписываюсь, есть только 1 раздел.

Я сомневаюсь, что производитель находится под большой нагрузкой, его отдельная среда со спорадическими сообщениями начинает создаваться этим производителем.

Это не помогаетчто сегодня пятница, и все уже ушли.

Я что-то упустил?Могу ли я переложить вину на сторону производителя?

Answer 1

Начиная с Kafka v0.10 каждое сообщение содержит атрибут метаданных метаданных времени, который устанавливается либо производителем во время создания сообщения, либо посредником во время вставки сообщения.

Имеется интерфейс org.apache.kafka.streams.processor.TimestampExtractor и метод extract, который извлекает метку времени из записи.

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

Источник: https://kafka.apache.org/10/javadoc/org/apache/kafka/streams/processor/TimestampExtractor.html

https://kafka.apache.org/documentation/#topicconfigs