Я столкнулся с проблемой в моем проекте, связанной с потреблением сообщений Kafka. Когда сообщение публикуется в тему Кафки, Кафкалистенер успешно читает и совершает смещение для большинства сообщений. Тем не менее, представляется, что для определенных сообщений существует основная проблема - эти сообщения, кажется, исчезают при достижении слушателя, не оставляя никаких трассировков или не запуская какие -либо ошибки или исключения. Примерно в 90% случаев все работает так, как и ожидалось, и сообщения обрабатываются правильно. Однако примерно в 10% случаев слушатель вообще не обрабатывает сообщение - сообщение иногда читается, а иногда и нет, даже если оно было опубликовано. Это непоследовательное поведение специально наблюдается с определенными сообщениями, которые непоследовательно потребляются < /p>
public void sendCartablErrorMessage(String erroMessage, String errorCode, String cartableTrackingNo) {
String topicName = "CCM_COMMIT_ERROR_REQUEST";
CartableCommitMessage cartableCommitMessage = new CartableCommitMessage(
errorCode, erroMessage,
Long.valueOf(cartableTrackingNo));
sendCartablMessageInCntCcmTest(cartableCommitMessage);
log.error("Cartable Error Message sent to kafka :"+cartableTrackingNo);
}
< /code>
public void sendCartablMessageInCntCcmTest(CartableCommitMessage cartableCommitMessage) {
String topicName = Topics.CNT_CCM_TEST;
kafkaTemplate.send(topicName, cartableCommitMessage);
}
< /code>
@KafkaListener(topics = Topics.CNT_CCM_TEST,containerFactory = "kafkaListenerContainerFactoryAutoAck")
public void onConfirmPayCartableEventTest(CartableCommitMessage request) throws JsonProcessingException {
log.info("Start CNT_CCM_TEST Listener..............");
try {
log.info("**********Cartable Listener RAW Topics.CNT_CCM_TEST : {}", new ObjectMapper().writeValueAsString(request));
} catch (Exception e) {
e.printStackTrace();
}
}
Подробнее здесь: https://stackoverflow.com/questions/796 ... kalistener
