Spring Cloud Stream 2.0.0 Release Notes

Key Features

• Polling Consumer
• Micrometer

Notable changes, enhancements and improvements

Content Type

• Default Content type is set to application/json which needs to be taken into consideration when migrating 1.3 application and/or operating in the mixed mode (i.e., 1.3 producer -> 2.0 consumer).
• Messages with textual payloads and "contentType" text/* or */json will be converted to Message<String> to maintain the behavioral compatibility with the previous version of the framework. Message's payload will still be converted to the appropriate argument type by the argument resolvers (MessageConverter's) if such argument is not a String (i.e., POJO) essentially resulting in secondary conversion which is not necessary for most cases. We are considering a flag to override this behavior to avoid secondary conversion.
• [TODO:Add to the DOC] @StreamMessageConverter - to define custom MessageConverter's used by argument resolvers. Added to the top of the list of existing MessageConverters
• List of MessageConverter's configured by default (in order):
  • TupleJsonMessageConverter - tbd
  • ByteArrayMessageConverter - tbd
  • ObjectStringMessageConverter - tbd
  • JavaSerializationMessageConverter (DEPRECATED) - tbd
  • KryoMessageConverter (DEPRECATED) - tbd
  • JsonUnmarshallingMessageConverter - tbd
• The contentType as a hint to help select the appropriate MessageConverter. For example, if payload is byte[] and argument is Foo which converter to use?
• Add note about the behavior of @Transformer around header propagation bug (unless SI fix is available before the release)

Notable Deprecations

• JavaSerializationMessageConverter (DEPRECATED) - tbd
• KryoMessageConverter (DEPRECATED) - tbd

Partitioning

• Note about partitionKeyExtractorClass deprecation in favor of Spring configured beans
• [TODO: ensure error is thrown at init] partitionCount must be accompanied by 'partitionKeyExtractor' otherwise it's an error

TODO 1.3 to 2.0 need migration pass for KryoConversion see #1142

• Actually consider section on overriding MessageCopnverters since by simply configuring @StreamMessageConverter and placing it ahead of existing one for the targeted content type one can accomplish override

@Bean
@StreamMessageConverter
public AlwaysStringKryoMessageConverter kryoOverrideMessageConverter() {
	return new AlwaysStringKryoMessageConverter(MimeType.valueOf("application/x-java-object"));
}

StreamListener Infrastructure enhancements

StreamLisener annotation post processor (StreamListenerAnnotationBeanPostProcessor) behavior is enhanced in 2.0 to address the needs of downstream implementations. This section is primarily applied to changes at the framework level (i.e. a new Binder requires a different behavior from the post processor). In a normal context, the users don't have to deal with these type of changes.

StreamListenerSetupMethodOrchestrator

StreamListenerMethodSetupOrchestrator is an API hook that allows downstream binder implementations or applications to inject custom strategies to alter the default StreamListener adapter method invocations.

Primary motivations for the new API

The default StreamListenerAnnotationBeanPostProcessor behaves in a strict manner enforcing various rules and validations. For example, it doesn't allow to have SendTo annotation with multiple destinations or multiple Output annotations present on a method annotated with StreamListener. There might be use cases in which a method needs to return a collection type or an array. Then, based on some rules it wants to send the data to multiple destinations through various bindings. If we rely on the default StreamListenerAnnotationBeanPostProcessor it is not possible to have this behavior for the StreamListener methods in a natural way. There is an extension mechanism already provided by the bean post processor to enhance the behavior, but this is not sufficient to satisfy this use case as the default validations still apply.

Contract for StreamListenerSetupMethodOrchestrator

Here is the contract of the StreamListenerSetupMethodOrchestrator interface

Implemenation Details

On the inbound side, the interface provides a default implementation which is equivalent to the current existing behavior, with the exception that this method is now available to be overridden by a potential downstream implementation. The main change though is introduced with the following methods.

boolean supports(Method method)
void orchestrateStreamListenerSetupMethod(StreamListener streamListener, Method method, Object bean)

The supports method takes a method and checks if the implementation can support orchestrating this method. The orchestrateStreamListenerSetupMethod allows an implementer to orchestrate the method by altering method invocation strategies. For instance, the implementation can allow having multiple bindings/destinations on the outbound side, change the way the method is invoked etc.

There is a default internal implementation provided for the framework that is part of the StreamListenerAnnotationBeanPostProcessor. This implementation is not for extension or used outside of this bean post processor. By default, this internal implementation supports any method that's annotated with StreamListener and therefore is used out of the box. However, if the user provides an implementation - be it either as part of a binder implementation or a new type of target type adapter (such as the Kafka Streams target type) - and register it as a Spring bean in the ApplicationContext, in that case, this bean is checked to see if the StreamListener method can be invoked using this implementation of the StreamListenerSetupMethodOrchestrator.

Structural changes in the Kafka Streams binder

• Binder will be called spring-cloud-stream-binder-kafka-streams starting with Elmhurst.RC1 (2.0.0.RC1 of the binder)
• Many classes those had KStream in its name are replaced with KafkaStreams(Details below)
• Many classes are removed from the public API to internal starting with 2.0.0.RC1 (Details below)
• All the properties that required kstream will need kafka.streams starting from 2.0.0.RC1. For example, earlier it was - spring.cloud.stream.kstream.binder.* or spring.cloud.stream.kstream.bindings.*, but now they are spring.cloud.stream.kafka.streams.binder.* and spring.cloud.stream.kafka.streams.bindings.* respectively.