Java - Kafka - Как получить сообщения от потребителя в режиме реального времени ⇐ JAVA

[Anonymous]

Я готовлю сценарии автоматизации тестов E2E для интерфейса Kafka. Поток заключается в том, что я создаю сообщение по теме x с телом, включая поле «ID», а затем прочитаю ответ ответа из темы y - Сообщение из темы ответа также содержит поле «ID», поэтому я могу соответствовать запросу с ответом. @Before Annotation (Junit 4) < /p>

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

@Before()
public void setUpConsumer() {
this.kafkaConsumer = new KafkaConsumer(setUpConsumerProperties());
kafkaConsumer.subscribe(topic);
}
< /code>
и здесь реализация setupconsumerproperties (): < /p>
protected Properties setUpConsumerProperties() {
Properties properties = new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "XXX);
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);
properties.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000);
properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "test-automation-client");
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test-automation-group");
properties.put("security.protocol", "SASL_SSL");
properties.put("sasl.mechanism", "SCRAM-SHA-512");
properties.put("ssl.endpoint.identification.algorithm", "");
properties.put("ssl.truststore.location", TRUSTSTORE_LOCATION);
properties.put("ssl.truststore.password", TRUSTSTORE_PASSWORD);
properties.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"\" password=\"
\";");
return properties;
}

А вот метод чтения сообщений - сначала я хочу назначить разделы клиенту, и как только это будет сделано, я начинаю читать новые сообщения - проблема заключается в том, что сообщения возвращаются через 20/30 секунды (или какое -то время не совсем ...), в то время как на инструменте GUI я могу видеть, что ответ в ответ опубликовано TP 2 секунды после публикации запроса в тему x, так что это определенно не с тем, что это не так.

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

 public TestClass readMessage() {
while (kafkaConsumer.assignment().isEmpty() && attemptsCounter < 10) {
kafkaConsumer.poll(Duration.ofMillis(100));
attemptsCounter++;
}

attemptsCounter = 0;
ConsumerRecords kafkaRecords;
while (attemptsCounter < DEFAULT_MAX_ATTEMPTS) {
kafkaRecords = kafkaConsumer.poll(Duration.ofMillis(2000));

if (!kafkaRecords.isEmpty()) {
String message = findMessageById(kafkaRecords); // find message by id matching request with response
if (message != null) {
return getThis();
}
}

attemptsCounter++;
log.info("Attempt {}/{}: Found {} records, continuing search...",
attemptsCounter, DEFAULT_MAX_ATTEMPTS, kafkaRecords.count());
}

throw new Exception("Cannot find message with ID " + this.id +
" after " + DEFAULT_MAX_ATTEMPTS + " attempts");
}

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


Подробнее здесь: https://stackoverflow.com/questions/797 ... -real-time