用Kafka处理Spring Cloud Stream中的NetworkException

Question

我在Spring Cloud Stream中有以下Listener - Producer：

@StreamListener(target = MultipleProcessor.DOTCONN_INPUT, condition= "headers['kafka_receivedTopic']=='dotconnectorissues'")
public void inputDot(Message<DotConnectorIssue> messageIn) {
    DotConnectorIssue data = messageIn.getPayload();
    ObjectMapper mapper = new ObjectMapper();
    DotConnectorUpdateDto dataMapped = new DotConnectorUpdateDto(data);
    if (dataMapped.getPlantCode().equals(plantCode)) {
        log.info("incoming dotConnectorIssue " + data);
        try {
            Message<String> messageOut = MessageBuilder
                    .withPayload(mapper.writeValueAsString(dataMapped))
                    .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                    .setHeader("type", "dotconnectorissue")
                    .build();

            boolean send = ehProcessor.outputAndon().send(messageOut, 15000L);
            log.info("sent message: "+ send);
            if (!send) messagePool.getPool().add(messageOut);
        } catch (JsonProcessingException e) {
            log.error("error during creating json", e);
        }
    }

}

这段代码可以工作，但有时消息会因为以下错误而无法发送：

[kafka-producer-network-thread | producer-2] ERROR o.s.k.s.LoggingProducerListener.onError - 
Exception thrown when sending a message with key='null' and payload='{123, 34, 116, 121, 112, 101, 34, 58, 34, 85, 80, 68, 65, 84, 69, 95, 68, 79, 84, 67, 79, 78, 78, 69...' to topic andon: 

org.apache.kafka.common.errors.NetworkException: The server disconnected before a response was received.

另外，如果send变量为true。

如何处理Spring Cloud Stream中的NetworkException错误？

Answer 1

设置sync property to true并捕获异常，或者，要异步获取错误，请设置生产者属性errorChannelEnabled。