Sample Kafka Producer for this plugin

[jital94]

Hi,
I'm trying to use this plugin but get following error
Unable to deserialize data java.lang.TypeNotPresentException: Type [unknown] not present
I'm sending JSON Strings
{"type":"Metrics","metric":"vmMemoryGB","timestamp":1500822865231,"value":7.0,"tags":{"host":"host-1"}}
Can you please specify how you are expecting messages from kafka stream.

[dheygere]

Hi @jital94,
You seem to have made a typo in the type field: "Metrics" instead of "Metric" (without the s).

[sjmittal]

Hi
I also get this error
KafkaRpcPluginThread: Unable to deserialize data java.lang.TypeNotPresentException: Type [unknown] not present

And my message in topic is
{"type":"Metric","metric":"fname.metric","tags":{"key1":"A2","key2":"B1"},"timestamp":1501079516693,"value":1755.0}

Any idea what maybe going wrong?

[manolama]

For folks seeing this can you try adding:

              final byte[] data = message.message();
              if (LOG.isDebugEnabled()) {
                LOG.debug(Bytes.pretty(data));
              }
              event = JSON.parseToObject(data, 
                  TypedIncomingData.class);

in place of Line 278 in the KafkaRpcPluginThread.java file please? Then enable debug logging and lets make sure the JSON is coming across properly.

[sjmittal]

Hi,
The issue was because this pulgin uses 2.4 and released version till data is 2.3.
I think you can mention in readme that this plugin would work with 2.4 version and that version can be build using next branch of opentsdb.

Also in this line
LOG.error("Unable to deserialize data " + ex);
If we can write as:
LOG.error("Unable to deserialize data ", ex);
It will give the cause of the exception.

Thanks
Sachin

[mikemeding]

@sjmittal have you been able to confirm that changing the version to 2.4 was able to fix the problem?

[mikemeding]

Nevermind. I just confirmed that this problem is NOT caused by version 2.3 vs 2.4 mismatch. @manolama I am seeing,
ERROR [KafkaRpcPluginThread.run] - Exception in kafkaReader or Tsdb Writer java.lang.NullPointerException: null at net.opentsdb.tsd.KafkaRpcPluginThread.run(KafkaRpcPluginThread.java:284) ~[opentsdb-rpc-kafka-2.3.2-SNAPSHOT.jar:] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_161] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_161] at java.lang.Thread.run(Thread.java:748) [na:1.8.0_161]

I can see my messages both entering and leaving the kafka queue correctly but for whatever reason they are not being parsed by JSONDeserializer correctly. Any idea what may be causing this?

[mikemeding]

@manolama Found the source of the error. Seems to be a concurrency issue as the threads are initialized before the group has finished loading the deserializer class. This can be fixed by replacing,
deserializer.deserialize(this, data);
with
group.getDeserializer().deserialize(this, data);
As the deserializer is defined as final in the worker thread and is set on initialization to null as the group class loader has not been run yet. But when the thread actually goes to run that pointer has now been corrected (because the class loader has finished) so just referencing it again fixes the problem.

Suggested bug fix?

[muffix]

This seems to be related to #10