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

Look for Payload RTT field in the Key as well

Browse Source
This commit is contained in:
Enrico Olivelli 2022-01-26 08:42:25 +01:00
parent e3ed838354
commit 614fb3bb51
1 changed files with 12 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/ops/PulsarConsumerOp.java
View File

@ -25,7 +25,8 @@ import org.apache.pulsar.client.api.schema.GenericObject;
import org.apache.pulsar.client.api.schema.GenericRecord;
import org.apache.pulsar.client.api.transaction.Transaction;
import org.apache.pulsar.common.schema.KeyValue;
import org.apache.pulsar.common.schema.SchemaType;
// this is from the Shaded version of Avro inside the Pulsar client jar
import org.apache.pulsar.shade.org.apache.avro.AvroRuntimeException;
public class PulsarConsumerOp implements PulsarOp {
@ -234,12 +235,20 @@ public class PulsarConsumerOp implements PulsarOp {
// look into the Key
if (keyValue.getKey() instanceof GenericRecord) {
GenericRecord keyPart = (GenericRecord) keyValue.getKey();
field = keyPart.getField(payloadRttTrackingField);
try {
field = keyPart.getField(payloadRttTrackingField);
} catch (AvroRuntimeException err) {
// field is not in the key
}
}
// look into the Value
if (keyValue.getValue() instanceof GenericRecord && field == null) {
GenericRecord valuePart = (GenericRecord) keyValue.getValue();
field = valuePart.getField(payloadRttTrackingField);
try {
field = valuePart.getField(payloadRttTrackingField);
} catch (AvroRuntimeException err) {
// field is not in the value
}
}
} else {
field = pulsarGenericRecord.getField(payloadRttTrackingField);