相关疑难解决方法(0)

使用Apache Kafka Java客户端(0.9),我正在尝试使用Kafka Producer类向代理发送一长串记录.

异步发送方法立即返回一段时间,然后在短时间内开始阻塞每次调用.大约三十秒后,客户端开始抛出异常(TimeoutException),并显示消息"Batch expired".

什么情况导致抛出此异常？

我正在使用Kafka和Spring-boot:

卡夫卡制片人班:

@Service
public class MyKafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    private static Logger LOGGER = LoggerFactory.getLogger(NotificationDispatcherSender.class);

    // Send Message
    public void sendMessage(String topicName, String message) throws Exception {
        LOGGER.debug("========topic Name===== " + topicName + "=========message=======" + message);
        ListenableFuture<SendResult<String, String>> result = kafkaTemplate.send(topicName, message);
        result.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                LOGGER.debug("sent message='{}' with offset={}", message, result.getRecordMetadata().offset());
            }

            @Override
            public void onFailure(Throwable ex) {
                LOGGER.error(Constants.PRODUCER_MESSAGE_EXCEPTION.getValue() + " : " + ex.getMessage());
            }
        }); …