Как избежать задержки при получении сообщений Kafka с помощью KafaConsumer API ⇐ JAVA

[Anonymous]

Мне нужно быстро получить 1000 сообщений из темы Kafka, но первоначальное получение происходит медленно kafka-clients 3.6.1 KafkaConsumer API.
Мы переходим со старого клиента Kafka (версия 0.8.1) на новый клиент Kafka (версия 3.6.1). В старой реализации Kafka использовался следующий подход, который был быстрым и эффективным.
Замеченная проблема:
Медленность при первом опросе: первая выборка 500 записи занимают значительно больше времени (например, 1284 мс) по сравнению с последующими выборками (например, 4 мс).
Инициализация потребителя Kafka: такие операции, как назначение(), поиск() и получение самого раннего смещения. отнимают время, и это влияет на производительность.
Цель:
Нам нужно сократить время, затрачиваемое на первый опрос (первые 500 записей ).
Мы также ищем способы эффективного выполнения метода Assign(), Seek() и получения самого раннего смещения в Kafka без ненужных задержек.
Пожалуйста, дайте рекомендации. в этих областях, чтобы воспроизвести производительность старой реализации Kafka с использованием новой версии Kafka. Потребительский API.
Например, если я указываю «0» в качестве смещения ввода, и это занимает время, указанное ниже (6 секунд)

Код: Выделить всё

Poll Records Count :292 diff :1227
Poll Records Count :292 diff :1181
Poll Records Count :296 diff :1234
Poll Records Count :292 diff :1133

Если я укажу смещение 110999, оно будет увеличиваться быстро.

Код: Выделить всё

Poll Records Count :500 diff :1278
Poll Records Count :500 diff :3

• Я устанавливаю пустой идентификатор группы, чтобы избежать перебалансировки, и я также разогреваю потребителя, но проблема с первоначальным опросом не решена
• Почему стартовые рекорды требуют времени. как этого избежать?

Мне нужно собрать 1000 записей и передать пользователю в качестве ответа. я выхожу из цикла после сбора 1000 записей.
Это мои обновленные свойства и код:

Код: Выделить всё

List msglist = new ArrayList();
try (KafkaConsumer consumer = KafkaConsumerFactory.createConsumer(clientName,fetchSize)){
TopicPartition topicPartition = new TopicPartition(topicName, 0);
consumer.assign(Collections.singletonList(topicPartition));
Map offsets = consumer.beginningOffsets(Collections.singletonList(topicPartition));
long kafkaEarliestOffset = offsets.get(topicPartition);
consumer.seek(topicPartition, readOffset);
consumer.poll(Duration.ofMillis(0));
do {
ConsumerRecords < String, String > records = consumer.poll(Duration.ofMillis(1500));
for (ConsumerRecord< String, String > consumerRecord: records) {
obj = new JSONObject(consumerRecord.value());
msglist.add(obj);
}
if (msglist.size() >= Math.round(limit/ inputReq.getApplicationArea().getReqInfo().size())
|| (endTime - startTime) >= waitTime) {
end = true;
consumer.commitSync();
}
}while (!end);
`
props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "");
props.setProperty(ConsumerConfig.CLIENT_ID_CONFIG,clientName);
props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "1024");
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,"500");
props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "500");
return new KafkaConsumer(props);
`

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

Подробнее здесь: https://stackoverflow.com/questions/792 ... nsumer-api