{"action":"closed","issue":{"url":"https://api.github.com/repos/openzipkin/brave/issues/102","labels_url":"https://api.github.com/repos/openzipkin/brave/issues/102/labels{/name}","comments_url":"https://api.github.com/repos/openzipkin/brave/issues/102/comments","events_url":"https://api.github.com/repos/openzipkin/brave/issues/102/events","html_url":"https://github.com/openzipkin/brave/issues/102","id":112069342,"number":102,"title":"Kafka Span Transport","user":{"login":"adriancole","id":64215,"avatar_url":"https://avatars.githubusercontent.com/u/64215?v=3","gravatar_id":"","url":"https://api.github.com/users/adriancole","html_url":"https://github.com/adriancole","followers_url":"https://api.github.com/users/adriancole/followers","following_url":"https://api.github.com/users/adriancole/following{/other_user}","gists_url":"https://api.github.com/users/adriancole/gists{/gist_id}","starred_url":"https://api.github.com/users/adriancole/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/adriancole/subscriptions","organizations_url":"https://api.github.com/users/adriancole/orgs","repos_url":"https://api.github.com/users/adriancole/repos","events_url":"https://api.github.com/users/adriancole/events{/privacy}","received_events_url":"https://api.github.com/users/adriancole/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"milestone":null,"comments":1,"created_at":"2015-10-19T04:21:14Z","updated_at":"2015-12-21T19:06:13Z","closed_at":"2015-12-21T19:06:13Z","body":"We need a kafka span transport, similar to the one in [htrace](https://github.com/apache/incubator-htrace/commit/9ebdafb45df03a4b0d90c77af65e0bd21c09ec1a).\r\n\r\nHere's some example code that works against zipkin server infrastructure:\r\n\r\n```java\r\nimport com.github.kristofa.brave.SpanCollector;\r\nimport java.io.ByteArrayOutputStream;\r\nimport java.io.Flushable;\r\nimport java.util.ArrayList;\r\nimport java.util.List;\r\nimport java.util.Properties;\r\nimport java.util.concurrent.BlockingQueue;\r\nimport java.util.concurrent.LinkedBlockingQueue;\r\nimport kafka.javaapi.producer.Producer;\r\nimport kafka.producer.KeyedMessage;\r\nimport kafka.producer.ProducerConfig;\r\nimport org.apache.thrift.TException;\r\nimport org.apache.thrift.protocol.TBinaryProtocol;\r\nimport org.apache.thrift.protocol.TProtocol;\r\nimport org.apache.thrift.transport.TIOStreamTransport;\r\n\r\npublic class KafkaSpanCollector implements SpanCollector, Flushable {\r\n  private final Producer<byte[], byte[]> producer;\r\n  // in real life, this would be bounded.\r\n  private BlockingQueue<com.twitter.zipkin.gen.Span> queue = new LinkedBlockingQueue<>();\r\n  private int limit = 200;\r\n\r\n  public KafkaSpanCollector() {\r\n    Properties props = new Properties();\r\n    props.put(\"metadata.broker.list\", System.getenv(\"METADATA_BROKER_LIST\"));\r\n    props.put(\"request.required.acks\", \"0\");\r\n    props.put(\"producer.type\", \"async\");\r\n    props.put(\"serializer.class\", \"kafka.serializer.DefaultEncoder\");\r\n    props.put(\"compression.codec\", \"1\");\r\n    this.producer = new Producer<>(new ProducerConfig(props));\r\n  }\r\n\r\n  @Override\r\n  public void collect(com.twitter.zipkin.gen.Span span) {\r\n    queue.offer(span);\r\n    if (queue.size() >= limit) {\r\n      flush();\r\n    }\r\n  }\r\n\r\n  @Override\r\n  public void flush() {\r\n    ByteArrayOutputStream baos = new ByteArrayOutputStream();\r\n    final TProtocol streamProtocol = new TBinaryProtocol.Factory().getProtocol(new TIOStreamTransport(baos));\r\n\r\n    List<KeyedMessage<byte[], byte[]>> messsages = new ArrayList<>(queue.size());\r\n    while (!queue.isEmpty()) {\r\n      com.twitter.zipkin.gen.Span span = queue.poll();\r\n      if (span != null) {\r\n        baos.reset();\r\n        try {\r\n          span.write(streamProtocol);\r\n          messsages.add(new KeyedMessage<>(\"zipkin\", baos.toByteArray()));\r\n        } catch (TException e) {\r\n          throw new AssertionError(e);\r\n        }\r\n\r\n      }\r\n    }\r\n    if (!messsages.isEmpty()) {\r\n      producer.send(messsages);\r\n    }\r\n  }\r\n\r\n  @Override\r\n  public void addDefaultAnnotation(String key, String value) {\r\n  }\r\n\r\n  @Override\r\n  public void close() {\r\n    producer.close();\r\n  }\r\n}\r\n```"}}