Message headers are incompatable

[rjj940221]

Spring integration messages use a MessageHeaders which implements Map<String, Object> this project converts message headers to a map of type Map<String, String> which causes issues for channels such as PriorityChannel which require the priority header to be an integer

[pavolloffay]

cc @gytis

[gytis]

@pavolloffay, I see that RabbitMQ integration does the same conversion to string. Do you think it would be worth considering adding a new Format e.g. ObjectMap to be used instead of TextMap in such situations?

[gytis]

@pavolloffay ?

[pavolloffay]

@rjj940221 what issues is the OT instrumentation causing?

Inject/Extract adapters are used to add tracing headers or read the headers and not to change headers globally. I didn't find where the conversion to string happens in this project.

[ttqin]

@pavolloffay we're having this issue with the Kafka binder too. issue is in this class even though the MessageTextMap's message is generic, the headers are <String,Object> and should not be .toString'd.

https://github.com/opentracing-contrib/java-spring-messaging/blob/master/opentracing-spring-messaging/src/main/java/io/opentracing/contrib/spring/integration/messaging/MessageTextMap.java#L42

[pavolloffay]

@ttqin what issue is it causing exactly?

That method is called only by tracer when extracting the context. Tracer looks only for tracing headers which can be converted string. The original headers are not changed.

[asanitckii]

we face the same issue.

The OpenTracingChannelInterceptor.preSend method creates a MessageTextMap MessageTextMap<?> carrier = new MessageTextMap<>(message);

and finally returns the message with transformed headers

return carrier.getMessage();

Is there a way to inject a custom Interceptor instead of the OpenTracingChannelInterceptor ?

[pavolloffay]

@Aleksander2016, @ttqin, @rjj940221 the headers are converted to string only when tracer reads them so there should not be a problem.

I have submitted #19 which tests types of the returned headers. Could you please have a look?

[asanitckii]

Ok, got it, thank you for help. It was changed 2018-05-23.

The type of the headers field was Map<String, String> now it's MutableMessageHeaders.

Need to check with spring uses the old version. May be I need update the project to newer spring version.

[asanitckii]

The latest version in Maven Central repository is 0.0.5 on Feb, 2018.

Do you use another repository?

[pavolloffay]

The last commit wasn't released I will cut 0.1.0 with the fix. I am closing this.

[asanitckii]

How long will it take for the new release?

[pavolloffay]

@gytis @geoand could you please release new version?

[geoand]

Sure thing. 0.1.0 right @pavolloffay ?

[pavolloffay]

Whichever is suitable :)

[pavolloffay]

thanks

[geoand]

On it :)

[dreaderxxx]

I have simular issue with header of type byte[]

Loooks like wrong converting in MessageTextMap line 42 headers.forEach((k, v) -> stringHeaders.put(k, String.valueOf(v)));

[pavolloffay]

@dreaderxxx did you try 0.1.0?

[dreaderxxx]

I did, maybe root issue is not in this library. The thing is that MessageTextMap is used by tracer to extract SpanContext but it fails

Error when extracting SpanContext from carrier. Handling gracefully.

io.jaegertracing.exceptions.MalformedTracerStateStringException: String does not match tracer state format: [B@6ef3781d
    at io.jaegertracing.SpanContext.contextFromString(SpanContext.java:117)
    at io.jaegertracing.propagation.TextMapCodec.extract(TextMapCodec.java:73)
    at io.jaegertracing.propagation.TextMapCodec.extract(TextMapCodec.java:27)
    at io.jaegertracing.PropagationRegistry$ExceptionCatchingExtractorDecorator.extract(PropagationRegistry.java:57)
    at io.jaegertracing.Tracer.extract(Tracer.java:193)
    at io.opentracing.contrib.spring.integration.messaging.OpenTracingChannelInterceptor.preSend(OpenTracingChannelInterceptor.java:60)
    at org.springframework.integration.channel.AbstractMessageChannel$ChannelInterceptorList.preSend(AbstractMessageChannel.java:611)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:453)
    at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:407)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:181)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:160)
    at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:108)
    at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:203)
    at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.access$300(KafkaMessageDrivenChannelAdapter.java:70)
    at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:387)
    at org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationRecordMessageListener.onMessage(KafkaMessageDrivenChannelAdapter.java:364)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1001)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:981)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:932)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:801)
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:689)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266)
    at java.util.concurrent.FutureTask.run(FutureTask.java)

"[B@6ef3781d" in this case is byte[] which is converted by String.valueOf(..) in MessageTextMap

[geoand]

@gytis Does this ring a bell? I seem to remember you mentioning something similar in the past, but I could be mistaken...

[gytis]

@geoand I don't remember seeing this. But doesn't it look like an issue with Jaeger?

[geoand]

@geoand I don't remember seeing this. But doesn't it look like an issue with Jaeger?

That would be my guess as well by looking at the stack trace

[pavolloffay]

I have created #24 to track this separately. It can be that context is corrupted or String.valueOf is corrupting it.