this happens with DF plugin -> KAFKA plugin (0.17.0 and 0.16.1-beta tested):

time="2016-11-09T20:35:05Z" level=error msg="error with publisher job" _module=scheduler-job block=run error="Publish call error: Cannot marshal metrics to JSON format, err=json: unsupported value: NaN" job-type=publisher plugin-config=map[] plugin-name=kafka plugin-version=-1 

this happens with 0.18.0:

time="2016-11-09T20:32:56Z" level=error msg="error with publisher job" _module=scheduler-job block=run error="Publish call error: Cannot initialize a new Sarama SyncProducer using the given broker addresses ([localhost:9092]), err=kafka: client has run out of available brokers to talk to (Is your cluster reachable?)" job-type=publisher plugin-config=map[brokers:{localhost:9092} topic:{snap}] plugin-name=kafka plugin-version=-1 

Same configuration, only snapd binary changed.

Hello,
I tested kafka plubsher using Snap V0.11 with Kafka pluign.
I user psutil (collection), passthrough (processor) and kafka (publisher).
I defined some task like this.

And, when I see messages using this command,
$ bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test5 --from-beginning

I saw this message.

How to can see messages correctly? (Formatiing Issues?)

There are some TODOs in README ("Documentation" and "Examples") that need to be resolved - at the very least an example of global config and sample task are needed.

Should allow to set key when sending data to kafka.

It means: all data now is sent to random partition. Which means - you do not have guarantee of proper ordering, nor you can properly use multiple readers (as data distribute randomly).
That makes any kind of "windows"/derivative calculation impossible.

Add two options: no key, key=plugin_running_on. Maybe some other keying options.
I would say, most of people would prefer partitioning by hostname/ip.

I am using the plugin binary for v0.15.0-beta. When data arrives in kafka it does not appear to be serialized as json. Have been using trifecta and kafka-console-consumer.sh to inspect data. Data for metrics is definitely arriving in kafka, but it is not in json format AFAICT, and is possibly corrupted. eg.

disk_totalfloat6�`�Asource10.1.0.90:5051plugin_running_onjenkins-1-node-0�m�F#2��intelmesosagentslavevalid_framework_messagesfloat6source10.1.0.90:5051plugin_running_onjenkins-1-node-0�m�F#2��
"��[]plugin.MetricType��������
MetricType��
Namespace_��LastAdvertisedTime_�Version_Config_��Data_Tags_��Unit_

Issue reported by smartx-jshan on main snap repo - intelsdi-x/snap#1057

•OS version : Ubuntu 14.04.4
•Snap version: Snap v0.14-beta
•Environment details: Virtual Machine based on KVM

When loading snap-plugin-publisher-kafka, I see the error message as below.

$ snapd -t 0 -l 1
$ snapctl plugin load snap-plugin-publisher-kafka

gob: name not registred for interface: "*cpolicy.ConfigPolicyNode"

However, when I use Snap v0.11-beta with kafka plugin, it works correctly.

Current data export option is quite BIIIIIG ONE. And having so much duplicated information does not help at all.
So - as for kafka, I would suggest to add new data format. Rather than exporting
{ [ {},{} ] } ... do little magic.
I would suggest drop most of fields, leaving only value and create tree structure on output.

Here is java code:

List<Metrics> metrics = mapper.readValue(record.value(), List.class, Metrics.class);
Map<String, Object> output = new HashMap<>();
for (Metrics m : metrics) {
    String[] split = m.namespace.split("/");
    List<String> path = new ArrayList<>(Arrays.asList(split));
    path.remove(path.size() - 1);
    path.remove(0);
    ensurePath(output, path).put(split[split.length - 1], m.data);
}


    private Map<String, Object> ensurePath(Map<String, Object> map, List<String> path) {
        if (!map.containsKey(path.get(0)))
            map.put(path.get(0), new HashMap<String, Object>());
        if (path.size() > 1) {
            String remove = path.remove(0);
            return ensurePath((Map<String, Object>) map.get(remove), path);
        } else
            return (Map<String, Object>) map.get(path.get(0));
    }

Here is partial JSON example on output:

{
  "intel":{
    "procfs":{
      "iface":{
        "lo":{
          "packets_recv":29227674,
          "compressed_recv":0,
          "multicast_sent":0,
          "frame_recv":0,
          "fifo_recv":0,
          "drop_sent":0,
          "bytes_sent":27885493282,
          "errs_recv":0,
          "compressed_sent":0,
          "errs_sent":0,
          "packets_sent":29227674,
          "drop_recv":0,
          "frame_sent":0,
          "multicast_recv":0,
          "bytes_recv":27885493282,
          "fifo_sent":0
        },
        "eth0":{
          "packets_recv":9660182,
          "compressed_recv":0,
          "multicast_sent":0,
          "frame_recv":0,
          "fifo_recv":0,
          "drop_sent":0,
          "bytes_sent":14853181981,
          "errs_recv":0,
          "compressed_sent":0,
          "errs_sent":0,
          "packets_sent":11323280,
          "drop_recv":0,
          "frame_sent":0,
          "multicast_recv":0,
          "bytes_recv":7701517374,
          "fifo_sent":0
        }
      },
      "processes":{
        "running":1,
        "stopped":0,
        "tracing":0,
        "wakekill":0,
        "parked":0,
        "waiting":0,
        "zombie":0,
        "waking":0,
          "pid":{
            "18399":{
            "snap-plugin-collector-interface":{
            "ps_disk_octets_wchar":174263636,

Only need to add
_timestamp and _tags from first Metrics[] element (or task) to make it complete.

{
  "_timestamp":"123123123",
  "_tags" : { ....}
  "intel" {
....
}

Explanation:
All plugins run at time, when task is scheduled - every 10 seconds, every minute etc. Only thing is important to have task scheduled/running time. If we run evey N secs, we do not care about exact time (nanoseconds :) ). did it run eta+0 or eta+1 or eta+whatever. +- it will run at our scheduled time.
Tags .. they are the same for each metrics reported - at least for all plugins I've worked with.
Adding this as optional "compressed, tree" output format would save a lot of space and that also allow better ways of processing.
Data size compared to original is ~10%.
I do not use logstash, but this format also would allow easy processing chain snap -> kafka -> logstash -> ES. As tree would be much simpler to consume and as you can see from example, you can effectively insert that data directly to ES.

The plugin opens a new connection to Kafka everytime when it produces a message on Kafka. However it does not close the connection after it is done. These not closed connections eventually floods the plugin and the snap task using the plugin stops.