Deserialise Jaeger spans in Java

[shivang16]

I am trying to get Jaeger span details from the Kafka plugin in Java. I'm using below code :

String bootstrapServers="127.0.0.1:9092"; String grp_id="tracing_app"; String topic="jaeger-spans"; //Creating consumer properties Properties properties=new Properties(); properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServers); properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer .class.getName()); properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName()); properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,grp_id); properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest"); //creating consumer org.apache.kafka.clients.consumer.KafkaConsumer<String,String> consumer= new org.apache.kafka.clients.consumer.KafkaConsumer<String,String>(properties); //Subscribing consumer.subscribe(Arrays.asList(topic)); //polling while(true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { System.out.println("Key: " + record.key() + ", Value:" + record.value()); System.out.println("Partition:" + record.partition() + ",Offset:" + record.offset()); } }

But due to some reason, I am getting output like this.

How can I properly de-serialize the span data?

[joe-elliott]

By default Jaeger pushes span data encoded using a protobuf encoding. You would need to use the protobuf definitions to generate classes to deserialize the data:
https://github.com/jaegertracing/jaeger-idl/blob/main/proto/api_v2/model.proto

Jaeger can also use a json format when sending to kafka which can be configured using --kafka.producer.encoding :
https://www.jaegertracing.io/docs/latest/cli/

[shivang16]

Even after having the encoding in JSONformat. I get the same output

Key: 04d8207846b2fc6f
Value:
� xF��o7�e�7��	readTrace”
� xF��o� xF��o(2�������:���B�
db.statement�
		SELECT trace_id, span_id, parent_id, operation_name, flags, start_time, duration, tags, logs, refs, process
		FROM traces
		WHERE trace_id = ?B
db.type	cassandraB
	componentgocqlB
internal.span.formatjaegerJ@
����꾁
event	searching
trace_id005c4ddef7b8f7eeRx
jaeger-query
jaeger.version	Go-2.30.0
hostname48db693dbdec
ip
172.18.0.8
�client-uuid46457659cebd7997
Partition:0,Offset:113

Is there any documentation available that shows how events are published to the Kafka stream?

[joe-elliott]

Not that I know of. Spans are written here:

jaeger/plugin/storage/kafka/writer.go

Lines 76 to 91 in 7872d1b

	func (w *SpanWriter) WriteSpan(ctx context.Context, span *model.Span) error {
	spanBytes, err := w.marshaller.Marshal(span)
	if err != nil {
	w.metrics.SpansWrittenFailure.Inc(1)
	return err
	}
	// The AsyncProducer accepts messages on a channel and produces them asynchronously
	// in the background as efficiently as possible
	w.producer.Input() <- &sarama.ProducerMessage{
	Topic: w.topic,
	Key: sarama.StringEncoder(span.TraceID.String()),
	Value: sarama.ByteEncoder(spanBytes),
	}
	return nil
	}

and the marshaller is chosen here:

jaeger/plugin/storage/kafka/factory.go

Lines 77 to 84 in 7872d1b

	switch f.options.Encoding {
	case EncodingProto:
	f.marshaller = newProtobufMarshaller()
	case EncodingJSON:
	f.marshaller = newJSONMarshaller()
	default:
	return errors.New("kafka encoding is not one of '" + EncodingJSON + "' or '" + EncodingProto + "'")
	}

Everything generally looks correct to me. Are you sure you're setting that flag value on the collector correctly?

[shivang16]

version: '3.8'

services:

  zookeeper:
    image: confluentinc/cp-zookeeper
    networks:
      - kafka-net
    container_name: zookeeper
    environment:
        ZOOKEEPER_CLIENT_PORT: 2181
    ports:
        - 2181:2181

  cassandra:
    hostname: cassandra
    image: cassandra
    networks:
      - kafka-net
    environment:
      MAX_HEAP_SIZE: 1G
      HEAP_NEWSIZE: 256M
    ports:
     - "9042:9042"

  cassandra-schema:
    image: jaegertracing/jaeger-cassandra-schema
    networks:
      - kafka-net
    depends_on:
      - cassandra

  kafka:
    image: confluentinc/cp-kafka:6.1.0
    networks:
      - kafka-net
    container_name: kafka
    environment:
        KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
        KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
        ALLOW_PLAINTEXT_LISTENER: "yes"
        KAFKA_LISTENERS-INTERNAL: //kafka:29092,EXTERNAL://localhost:9092
        KAFKA_ADVERTISED: INTERNAL://kafka:29092,EXTERNAL://localhost:9092
        KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:29092,EXTERNAL://localhost:9092
        KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
        KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
    ports:
        - 9092:9092
        - 29092:29092
    depends_on:
        - zookeeper
    restart: on-failure

  jaeger-collector:
    image: jaegertracing/jaeger-collector
    container_name: jaeger-collector
    networks:
      - kafka-net
    ports:
      - "14250:14250"
      - "14267:14267"
      - "14268:14268" # HTTP collector port to receive spans
      - "14269:14269" # HTTP health check port
    restart: on-failure
    environment:
      LOG_LEVEL: "debug"
      SPAN_STORAGE_TYPE: "kafka"
      KAFKA_BROKERS: "kafka:9092"
      KAFKA_PRODUCER_BROKERS: "kafka:29092"
      KAFKA_PRODUCER_ENCODING: "protobuf"
  jaeger-agent:
    image: jaegertracing/jaeger-agent
    container_name: jaeger-agent
    networks:
      - kafka-net
    command: ["--reporter.grpc.host-port=jaeger-collector:14250"]
    ports:
      - "5775:5775/udp"
      - "6831:6831/udp"
      - "6832:6832/udp"
      - "5778:5778"
    environment:
      LOG_LEVEL: "debug"
      SPAN_STORAGE_TYPE: "kafka"
    restart: on-failure
    depends_on:
      - jaeger-collector

  jaeger-ingester:
    image: jaegertracing/jaeger-ingester
    container_name: jaeger-ingester
    networks:
      - kafka-net
    ports:
      - "14270:14270" # HTTP health check port: http://localhost:14270/
      - "14271:14271" # Metrics port: http://localhost:14270/metrics
    restart: on-failure
    environment:
      LOG_LEVEL: debug
      INGESTER_PARALLELISM: 1
      INGESTER_DEADLOCKINTERVAL: ms
      SPAN_STORAGE_TYPE: cassandra
      CASSANDRA_SERVERS: cassandra
      CASSANDRA_KEYSPACE: jaeger_v1_dc1
      METRICS_BACKEND: expvar
      KAFKA_CONSUMER_BROKERS: kafka:29092
      KAFKA_CONSUMER_TOPIC: jaeger-spans
    depends_on:
      - cassandra-schema

  jaeger-query:
    image: jaegertracing/jaeger-query
    container_name: jaeger-query
    networks:
      - kafka-net
    ports:
      - "16686:16686" # Jaeger UI port
      - "16687:16687" # HTTP health check port: http://localhost:16687/
    restart: on-failure
    depends_on:
      - cassandra-schema
    environment:
      LOG_LEVEL: debug
      SPAN_STORAGE_TYPE: cassandra
      CASSANDRA_SERVERS: cassandra
      CASSANDRA_KEYSPACE: jaeger_v1_dc1
      JAEGER_ENDPOINT: http://jaeger-collector:14268/api/traces

networks:
  kafka-net:
    driver: bridge

This is the docker script I am using to run the services.