Как избежать задержки при получении сообщений 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 записей).
Мы также ищем способы эффективного выполнения метода присваивания(), поиска() и получения самого раннего смещения в Kafka без ненужных задержек.
Дайте, пожалуйста, рекомендации по этим областям, чтобы воспроизвести производительность старой реализации Kafka с использованием новой Kafka Consumer API.
Например, если я указываю «0» в качестве смещения ввода, и это занимает время, указанное ниже (6 секунд).
Количество записей опроса: 292 различий: 1227
Количество записей опросов: 292 различий: 1181
Количество записей опросов: 296 различий: 1234
Количество записей опросов: 292 различий :1133
Если я укажу смещение как 110999, и оно будет увеличиваться быстро.
Количество записей опроса: 500 diff: 1278
Опрос Количество записей: 500, разница: 3

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

Мое требование - я нужно собрать 1000 записей и передать пользователю ответ.iam вырвется из цикла после сбора 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, но проблема с медленностью не решена. Я ожидаю скорости получения записи из темы kafa с использованием API KafkaConsumer. Я также упомянул свойства здесь. Я пробовал использовать различные потребительские свойства kafka, но проблема с медлительностью все еще не решена. Я ожидаю скорости получения записей из темы kafa с использованием KafkaConsumer API. Я также упомянул свойства здесь.``

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