DefaultKafkaHeaderMapper$NonTrustedHeaderType encountered with a minimal SCS pipeline

[tjormola]

Please see https://github.com/tjormola/scs-problems-demo code demonstrating this problem. As the issue description, I will copy and paste the README here.

---

This project demonstrates a problem I was having when using the Stream Cloud Stream Kafka Binder with a project written in Kotlin.

The following chain of SCS handlers will throw

org.springframework.messaging.MessageDeliveryException: failed to send Message to channel 'input'; nested exception is java.lang.IllegalArgumentException: Unknown type for contentType header value: class org.springframework.kafka.support.DefaultKafkaHeaderMapper$NonTrustedHeaderType

when bound together using the SCS Kafka Binder. When using SCS Rabbit Binder, it'll work just fine.

Source configuration defined in demo.source.HttpSourceApplication:

@EnableBinding(Source::class)
@Configuration
class HttpSourceConfiguration {
    @Bean
    @StreamEmitter
    @Output(Source.OUTPUT)
    fun publishIncomingJsonPayload(): Publisher<Message<String>> {
        return IntegrationFlows
                .from(WebFlux.inboundChannelAdapter("/")
                        .requestMapping { it
                                .methods(HttpMethod.POST)
                                .consumes(MediaType.APPLICATION_JSON_VALUE)
                        }
                        .requestPayloadType(ResolvableType.forClass(String::class.java))
                        .statusCodeFunction { HttpStatus.OK })
                .toReactivePublisher()
    }
}

Processor configuration defined in demo.processor.ReactiveProcessorApplication:

@EnableBinding(TestProcessor::class)
@Configuration
class ReactiveProcessorConfiguration {
    @StreamListener
    @Output(TEST_PROCESSOR_OUTPUT_CHANNEL_NAME)
    fun processJsonMessage(@Input(Processor.INPUT) input: Flux<String>): Flux<String> {
        return input
                .map { it }
    }
}

Run the tests demo.test.RabbitReactiveIntegrationTest and demo.test.KafkaReactiveIntegrationTest under the integration-tests subproject to trigger the effect. The Kafka test will use an embedded Kafka instance and the Rabbit test expects the default configuration, i.e. Rabbit running on localhost with default guest credentials.

I wonder if Kotlin native types stuff is interfering here somehow. https://docs.spring.io/spring-kafka/docs/2.1.4.RELEASE/api/org/springframework/kafka/support/DefaultKafkaHeaderMapper.html#addTrustedPackages-java.lang.String...- refers to java.util, java.lang as being trusted packages by default. I haven't tried implementing this in plain Java, maybe it would work.
But as Kotlin is now very much liked in Spring world and the same code works with the other official binder implementation, I think this should work with Kafka out of the box, too. Thus I consider this behaviour a bug.

[garyrussell]

Try adding org.springframework.http (where MediaType lives) to the default header mapper's trustedPackages.

You can provide the bean name of the customized mapper in the binder's headerMapperBeanName property.

[garyrussell]

We should probably consider adding this to the default list of trusted packages, either in the binder, or in spring-kafka itself.

[tjormola]

Thanks a lot for the suggestion. I did this in tjormola/scs-problems-demo@2adb27f and the DefaultKafkaHeaderMapper$NonTrustedHeaderType exception is now sorted. Unfortunately, now we got a bunch of new exceptions related to header processing and JSON de-serialization in ObjectMapper. I don't fully understand what's happening here without further research, I might have time to debug this later, but here's the console log from the test run in case the problem cause is evident to you with a short glimpse..

https://gist.github.com/tjormola/2639373dbdca5d848f7261317a5da7d9

[olegz]

@tjormola I was able to reproduce it and I believe the issue is that we don't have deserializer for MediaType, only for MimeType. In the controlled environment passing org.springframework.http.MediaType triggers MimeTypeJsonDeserializer and your content type json string gets properly converted to MimeType.
While I didn't follow the discussion you and Gary had so I don't know the context of the conversation, but I am wondering if passing a org.springframework.http.MediaType is something you control.

[ivanprostran]

Hi, I've managed to get this to work with the following configuration/code:

1. Add Bean allocation (with related object mapper and support for MediaType)

	@Bean
	DefaultKafkaHeaderMapper mapper(){
		ObjectMapper objectMapper = new ObjectMapper();
		SimpleModule module = new SimpleModule();
		module.addDeserializer(MediaType.class, new JackDeserialization());
		objectMapper.registerModule(module);
		DefaultKafkaHeaderMapper defaultKafkaHeaderMapper = new DefaultKafkaHeaderMapper(objectMapper);
		defaultKafkaHeaderMapper.addTrustedPackages("org.springframework.http");
		defaultKafkaHeaderMapper.addTrustedPackages("org.springframework.util.MimeType");
		defaultKafkaHeaderMapper.addTrustedPackages("org.springframework.http.MediaType");
		LOG.info("**** Added trusted packages **** ");
		return defaultKafkaHeaderMapper;
	}

2. Create custom deserialization

public class JackDeserialization extends StdDeserializer<MediaType> {
	public JackDeserialization() {
		this(null);
	}

	public JackDeserialization(Class<?> vc) {
		super(vc);
	}

	@Override
	public MediaType deserialize(JsonParser jp, DeserializationContext ctxt)
			throws IOException, JsonProcessingException {
		JsonNode node = jp.getCodec().readTree(jp);

		MediaType mediaType = new MediaType("application", "json");
		return mediaType;
	}
}

3. Don't forget to add the following properties (important for headerMapper)

spring.cloud.stream.kafka.binder.headerMapperBeanName=mapper

[sobychacko]

Closing this issue.