在 Kafka Send() 方法的 Onfailure() 方法调用中获取 Objects 值

Question

我想获取那些未发送到 Kafka 的 Person 对象，即 onFailure() 方法。

1)我创建了人员类型的临时数组并将其传递给 onFailure() 。但这不起作用。它总是显示相同的 temp[0] 不起作用。它总是打印列表中的最后一个值。 2）我试图创建本地人对象来解决这个问题。但是我得到了编译错误

"在封闭范围内定义的局部变量 pp 必须是 final 或有效 final"

我该如何解决这个问题？

class Sender {

        @Autowired
        private KafkaTemplate<String, Person> template;

        private static final Logger LOG = LoggerFactory.getLogger(Sender.class);



        Person temp= null;

        // @Transactional("ktm")
        public void sendThem(List<Person> toSend) throws InterruptedException {
            List<ListenableFuture<SendResult<String, Person>>> futures = new ArrayList<>();


            ListenableFutureCallback<SendResult<String, Person>> callback = new ListenableFutureCallback<SendResult<String, Person>>() {

                @Override
                public void onSuccess(SendResult<String, Person> result) {
                    //LOG.info(" message success 1: " + result.getProducerRecord().value());
                    LOG.info(" message success 2: " + temp);
                }

                @Override
                public void onFailure(Throwable ex) {
                    LOG.info(" message failed : "  + temp);

                }
            };

            for (Person p : toSend) {

                temp=p;
                ListenableFuture<SendResult<String, Person>> future = template.send("t_101", temp);


                future.addCallback(callback);
            }



        }
    }

Answer 1

失败的发送值在 Throwable 中可用 - 将其转换为 KafkaProducerException。您的Person 位于producerRecord.value()。

我已将答案更新为your other question。

您也可以在循环中创建一个新的回调，然后p 将可供您使用；但使用单个回调并从异常中获取Person 效率更高。