Excessive logging (Exception thrown while calling task.commitRecord())

[hrvbos]

Hi,

We have an issue with Mirus version 0.6.9. We are having excessive number of logs with the Exception thrown while calling task.commitRecord().

Exception thrown while calling task.commitRecord() (org.apache.kafka.connect.runtime.WorkerSourceTask)
java.lang.NullPointerException
at com.salesforce.mirus.metrics.MirrorJmxReporter.recordMirrorLatency(MirrorJmxReporter.java:162)
at com.salesforce.mirus.MirusSourceTask.commitRecord(MirusSourceTask.java:241)
at org.apache.kafka.connect.runtime.WorkerSourceTask.commitTaskRecord(WorkerSourceTask.java:470)
at org.apache.kafka.connect.runtime.WorkerSourceTask.lambda$sendRecords$4(WorkerSourceTask.java:387)
at org.apache.kafka.clients.producer.KafkaProducer$InterceptorCallback.onCompletion(KafkaProducer.java:1365)
at org.apache.kafka.clients.producer.internals.ProducerBatch.completeFutureAndFireCallbacks(ProducerBatch.java:228)
at org.apache.kafka.clients.producer.internals.ProducerBatch.done(ProducerBatch.java:197)
at org.apache.kafka.clients.producer.internals.Sender.completeBatch(Sender.java:684)
at org.apache.kafka.clients.producer.internals.Sender.completeBatch(Sender.java:655)
at org.apache.kafka.clients.producer.internals.Sender.lambda$null$1(Sender.java:574)
at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
at org.apache.kafka.clients.producer.internals.Sender.lambda$handleProduceResponse$2(Sender.java:561)
at java.base/java.lang.Iterable.forEach(Iterable.java:75)
at org.apache.kafka.clients.producer.internals.Sender.handleProduceResponse(Sender.java:561)
at org.apache.kafka.clients.producer.internals.Sender.lambda$sendProduceRequest$3(Sender.java:785)
at org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:109)
at org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:584)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:576)
at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:327)
at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:242)
at java.base/java.lang.Thread.run(Thread.java:829)

From what I see the actual issue is that topic which is initialized in mirrorJmxExporter and the topic which is passed to recordMirrorLatency method are not the same.
Topics for mirrorJmxExporter are initialized here

mirus/src/main/java/com/salesforce/mirus/MirusSourceTask.java

Line 127 in 2fbc34e

this.mirrorJmxReporter.addTopics(topicPartitionList);

and this will be topic which is source.

But topic which is sent to recordMirrorLatency

mirus/src/main/java/com/salesforce/mirus/MirusSourceTask.java

Line 235 in 2fbc34e

mirrorJmxReporter.recordMirrorLatency(sourceRecord.topic(), latency);

is taken from sourceRecord.

When SourceRecord is built topic is prefixed/suffixed so it becomes destination topic

mirus/src/main/java/com/salesforce/mirus/MirusSourceTask.java

Line 298 in 2fbc34e

String topic = destinationTopicNamePrefix + consumerRecord.topic() + destinationTopicNameSuffix;

So finally when record is passed to mirrorJmxExporter it tries to get something for that topic and breaks.

After compiling the Mirus with changing the topic for recordMirrorLatency with something like this
mirrorJmxReporter.recordMirrorLatency(sourceRecord.sourcePartition().get("topic").toString(), latency) the problem goes away (this would not be correct way to solve this problem, but tried it just to see if it helps).

From what I see the problem got introduced in Mirus in this PR #89

This should be fixed since there is no config option or something similar to disable this behavior.

BR,
Hrvoje