When listeners are provided for DefaultKafkaConsumerFactory, the target KafkaConsumer instance is proxied.
The java.lang.reflect.Proxy is Serializable,
but the value it is wrapping is not.
When the MessageHeaders is serialized (e.g. into persistent MessageStore), it checks for Serializable type only on top-level object of the header. Therefore, the Proxy is passing condition, but eventually we fail with NotSerializableException, since the proxied object is not like that
Remove kafka_consumer from a message before it reaches an aggregator with its logic to serialize message into the store
Fixes https://github.com/spring-cloud/stream-applications/issues/500
When
listenersare provided forDefaultKafkaConsumerFactory, the targetKafkaConsumerinstance is proxied. Thejava.lang.reflect.ProxyisSerializable, but the value it is wrapping is not. When theMessageHeadersis serialized (e.g. into persistentMessageStore), it checks forSerializabletype only on top-level object of the header. Therefore, theProxyis passing condition, but eventually we fail withNotSerializableException, since the proxied object is not like thatkafka_consumerfrom a message before it reaches an aggregator with its logic to serialize message into the storeThis is a workaround until Spring for Apache Kafka is released with the fix: https://github.com/spring-projects/spring-kafka/pull/2822