spring-cloud-stream - 无法从 DLQ spring cloud stream kafka 读取异常标头
问题描述
使用 Spring Cloud Stream Kafka 侦听器从 kafka 主题读取消息,并在异常情况下将其发送到配置属性的死信队列
spring.cloud.stream.kafka.bindings.input.consumer.enable-dlq=true
spring.cloud.stream.kafka.bindings.input.consumer.dlq-name=book_error
我可以向 DLQ 发送消息。但是,当我尝试从 DLQ 读取以便我可以根据 Error Message 采取行动时。我无法读取嵌入在标题中的异常。
@StreamListener("dlqChannel")
public void error(Message<?> message) {
System.out.println("Handling ERROR: READING FROM DLQ");
logger.info("header :" +message.getHeaders());
logger.info("payload : " +message.getPayload());
//return message;
}
标头有效负载似乎具有我无法破译的对象 ID。如何根据异常消息解析错误和处理。下面是我尝试打印时得到的标题
header :{x-original-offset=[B@f82eb25, x-original-partition=[B@7a3b83c, deliveryAttempt=1, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedTopic=book_errors, kafka_offset=0, x-exception-message=[B@6dcc9872, x-exception-fqcn=[B@68079694, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@70449372, x-original-topic=[B@6a3ca71e, x-original-timestamp-type=[B@63baad23, kafka_receivedPartitionId=0, contentType=application/json, x-original-timestamp=[B@37dd34f6, kafka_receivedTimestamp=1579990310188, kafka_groupId=bkerrgrp, x-exception-stacktrace=[B@6356ee7c}
解决方案
异常标头是byte[]
- 这是 Kafka 支持的唯一类型。各种字符串值信息存储为String.getBytes(StandardCharsets.UTF_8))
.
利用
String exceptionMessage = new String(message.getHeaders().get("x-exception-message", byte[].class), StandardCharsets.UTF_8);
使用适当的类型(int、long)存储数值。
对那些使用ByteBuffer.wrap(header).getInt()
等。
这是存储标题的代码...
kafkaHeaders.add(
new RecordHeader(X_ORIGINAL_TOPIC, record.topic().getBytes(StandardCharsets.UTF_8)));
kafkaHeaders.add(new RecordHeader(X_ORIGINAL_PARTITION,
ByteBuffer.allocate(Integer.BYTES).putInt(record.partition()).array()));
kafkaHeaders.add(new RecordHeader(X_ORIGINAL_OFFSET,
ByteBuffer.allocate(Long.BYTES).putLong(record.offset()).array()));
kafkaHeaders.add(new RecordHeader(X_ORIGINAL_TIMESTAMP,
ByteBuffer.allocate(Long.BYTES).putLong(record.timestamp()).array()));
kafkaHeaders.add(new RecordHeader(X_ORIGINAL_TIMESTAMP_TYPE,
record.timestampType().toString().getBytes(StandardCharsets.UTF_8)));
kafkaHeaders.add(new RecordHeader(X_EXCEPTION_FQCN,
throwable.getClass().getName().getBytes(StandardCharsets.UTF_8)));
kafkaHeaders.add(new RecordHeader(X_EXCEPTION_MESSAGE,
throwable.getMessage().getBytes(StandardCharsets.UTF_8)));
kafkaHeaders.add(new RecordHeader(X_EXCEPTION_STACKTRACE,
getStackTraceAsString(throwable).getBytes(StandardCharsets.UTF_8)));