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

Handle NPE thrown on empty message sequence number header

Browse Source
This commit is contained in:
Massimiliano Mirelli 2023-03-02 10:09:28 +02:00
parent 31fe98b402
commit 97d2a01675
1 changed files with 11 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -210,12 +210,17 @@ public class OpTimeTrackKafkaConsumer extends OpTimeTrackKafkaClient {
}
private void checkAndUpdateMessageErrorCounter(ConsumerRecord<String, String> record) {
String msgSeqIdStr = new String(record.headers().lastHeader(PulsarAdapterUtil.MSG_SEQUENCE_NUMBER).value());
if ( !StringUtils.isBlank(msgSeqIdStr) ) {
long sequenceNumber = Long.parseLong(msgSeqIdStr);
ReceivedMessageSequenceTracker receivedMessageSequenceTracker =
receivedMessageSequenceTrackerForTopic.apply(record.topic());
receivedMessageSequenceTracker.sequenceNumberReceived(sequenceNumber);
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.");
} 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);
}
}
}