IntenseFinance/nosqlbench
3
0
Fork 0
mirror of https://github.com/nosqlbench/nosqlbench.git synced 2025-02-25 18:55:28 -06:00
Code Issues Packages Projects Releases Wiki Activity

Simplify handling NPE thrown on empty message sequence number header

Browse Source
This commit is contained in:
Massimiliano Mirelli
2023-03-02 10:17:22 +02:00
parent 97d2a01675
commit c6e7829a2a
1 changed files with 7 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
adapter-kafka/src/main/java/io/nosqlbench/adapter/kafka/ops/OpTimeTrackKafkaConsumer.java
View File

@@ -211,16 +211,14 @@ public class OpTimeTrackKafkaConsumer extends OpTimeTrackKafkaClient {
private void checkAndUpdateMessageErrorCounter(ConsumerRecord<String, String> record) { private void checkAndUpdateMessageErrorCounter(ConsumerRecord<String, String> record) {
Header msg_seq_number_header = record.headers().lastHeader(PulsarAdapterUtil.MSG_SEQUENCE_NUMBER); Header msg_seq_number_header = record.headers().lastHeader(PulsarAdapterUtil.MSG_SEQUENCE_NUMBER);
if (msg_seq_number_header == null) { String msgSeqIdStr = msg_seq_number_header != null ? new String(msg_seq_number_header.value()) : StringUtils.EMPTY;
logger.warn("Message sequence number header is null, skipping e2e message error metrics generation."); if (!StringUtils.isBlank(msgSeqIdStr)) {
long sequenceNumber = Long.parseLong(msgSeqIdStr);
ReceivedMessageSequenceTracker receivedMessageSequenceTracker =
receivedMessageSequenceTrackerForTopic.apply(record.topic());
receivedMessageSequenceTracker.sequenceNumberReceived(sequenceNumber);
} else { } else {
String msgSeqIdStr = new String(msg_seq_number_header.value()); logger.warn("Message sequence number header is null, skipping e2e message error metrics generation.");
if (!StringUtils.isBlank(msgSeqIdStr)) {
long sequenceNumber = Long.parseLong(msgSeqIdStr);
ReceivedMessageSequenceTracker receivedMessageSequenceTracker =
receivedMessageSequenceTrackerForTopic.apply(record.topic());
receivedMessageSequenceTracker.sequenceNumberReceived(sequenceNumber);
}
} }
} }