From c6e7829a2af39e7a96b496c630815633af14dabb Mon Sep 17 00:00:00 2001 From: Massimiliano Mirelli Date: Thu, 2 Mar 2023 10:17:22 +0200 Subject: [PATCH] Simplify handling NPE thrown on empty message sequence number header --- .../kafka/ops/OpTimeTrackKafkaConsumer.java | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/adapter-kafka/src/main/java/io/nosqlbench/adapter/kafka/ops/OpTimeTrackKafkaConsumer.java b/adapter-kafka/src/main/java/io/nosqlbench/adapter/kafka/ops/OpTimeTrackKafkaConsumer.java index 53bba79c7..9492ff331 100644 --- a/adapter-kafka/src/main/java/io/nosqlbench/adapter/kafka/ops/OpTimeTrackKafkaConsumer.java +++ b/adapter-kafka/src/main/java/io/nosqlbench/adapter/kafka/ops/OpTimeTrackKafkaConsumer.java @@ -211,16 +211,14 @@ public class OpTimeTrackKafkaConsumer extends OpTimeTrackKafkaClient { private void checkAndUpdateMessageErrorCounter(ConsumerRecord record) { Header msg_seq_number_header = record.headers().lastHeader(PulsarAdapterUtil.MSG_SEQUENCE_NUMBER); - if (msg_seq_number_header == null) { - logger.warn("Message sequence number header is null, skipping e2e message error metrics generation."); + String msgSeqIdStr = msg_seq_number_header != null ? new String(msg_seq_number_header.value()) : StringUtils.EMPTY; + if (!StringUtils.isBlank(msgSeqIdStr)) { + long sequenceNumber = Long.parseLong(msgSeqIdStr); + ReceivedMessageSequenceTracker receivedMessageSequenceTracker = + receivedMessageSequenceTrackerForTopic.apply(record.topic()); + receivedMessageSequenceTracker.sequenceNumberReceived(sequenceNumber); } else { - String msgSeqIdStr = new String(msg_seq_number_header.value()); - if (!StringUtils.isBlank(msgSeqIdStr)) { - long sequenceNumber = Long.parseLong(msgSeqIdStr); - ReceivedMessageSequenceTracker receivedMessageSequenceTracker = - receivedMessageSequenceTrackerForTopic.apply(record.topic()); - receivedMessageSequenceTracker.sequenceNumberReceived(sequenceNumber); - } + logger.warn("Message sequence number header is null, skipping e2e message error metrics generation."); } }