No converter found capable of converting from type [java.lang.String] to type [org.springframework.expression.Expression]

[luamas]

spring.cloud.stream.kafka.default.producer.message-key-expression: headers['messageKey']

When you configure the default

Failed to bind properties under 'spring.cloud.stream.kafka.default.producer.message-key-expression' to org.springframework.expression.Expression:

Property: spring.cloud.stream.kafka.default.producer.message-key-expression
Value: headers['messageKey']
Origin: class path resource [application.yml] - 11:37
Reason: org.springframework.core.convert.ConverterNotFoundException: No converter found capable of converting from type [java.lang.String] to type [org.springframework.expression.Expression]

[luamas]

version 3.2.1

[olegz]

It appears to be some type of miss-configuration. The converter is registered with core. Are you saying it worked with previous versions?

[luamas]

spring:
cloud:
stream:
kafka:
bindings:
test-out-0:
producer:
message-key-expression: headers['messageKey']

There is no problem with this configuration

doc

https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/3.2.1/reference/html/spring-cloud-stream-binder-kafka.html#kafka-consumer-properties

To avoid repetition, Spring Cloud Stream supports setting values for all channels, in the format of spring.cloud.stream.kafka.default.consumer.=.

[olegz]

That is not what i asked you

[luamas]

This is a new application, configured in the default way. Otherwise, 'message-key-expression' needs to be configured separately for each one.

[luamas]

I'm starting to see what you mean. Spring Boot core bind has an error.

The following code

private Object bindObject(ConfigurationPropertyName name, Bindable target, BindHandler handler,
Context context, boolean allowRecursiveBinding) {
ConfigurationProperty property = findProperty(name, target, context);
if (property == null && context.depth != 0 && containsNoDescendantOf(context.getSources(), name)) {
return null;
}
AggregateBinder<?> aggregateBinder = getAggregateBinder(target, context);
if (aggregateBinder != null) {
return bindAggregate(name, target, handler, context, aggregateBinder);
}
if (property != null) {
try {
return bindProperty(target, context, property);
}
catch (ConverterNotFoundException ex) {
// We might still be able to bind it using the recursive binders
Object instance = bindDataObject(name, target, handler, context, allowRecursiveBinding);
if (instance != null) {
return instance;
}
throw ex;
}
}
return bindDataObject(name, target, handler, context, allowRecursiveBinding);
}

[luamas]

link spring-projects/spring-integration#3726

[olegz]

What version of spring cloud stream you are using. I do remember a long while back similar error in a multi-binder scenario, but then it was addressed, hence I am wondering what version

[luamas]

@olegz

spring cloud 2021.0.0
spring boot 2.6.3

This time it is a default configuration

org.springframework.cloud.stream.config.BindingServiceProperties

private void bindToDefault(String binding) {
		BindingProperties bindingPropertiesTarget = new BindingProperties();
		Binder binder = new Binder(
				ConfigurationPropertySources
						.get(this.applicationContext.getEnvironment()),
				new PropertySourcesPlaceholdersResolver(
						this.applicationContext.getEnvironment()),
				IntegrationUtils.getConversionService(
						this.applicationContext.getBeanFactory()),
				null);
		binder.bind("spring.cloud.stream.default",
				Bindable.ofInstance(bindingPropertiesTarget));
		this.bindings.put(binding, bindingPropertiesTarget);
	}

The IntegrationUtils.getConversionService(this.applicationContext.getBeanFactory()) method does not get SpelConverter .

@Bean
	@ConfigurationPropertiesBinding
	@IntegrationConverter
	public Converter<String, Expression> spelConverter() {
		return new SpelConverter();
	}

[olegz]

You don't need to register Spel converter, it is already registered by the framework

[olegz]

Any updates on this? Perhaps a reproducible sample?

[luamas]

However, the converter does not exist after debugging, so you can try it out

[luamas]

spring.cloud.stream.kafka.default.producer.message-key-expression is in need of this converter

[iguissouma]

I'm having similar issue in a project with multi binder while I'm trying to upgrade to upgrade to latest versions
spring-boot: 2.6.5
spring-cloud: 2021.0.1

        my-kafka:
          type: kafka
          inheritEnvironment: true
          environment:
            spring.cloud.stream:
              kafka:
                default:
                  producer:
                    messageKeyExpression: headers['messageKey']
                    headerPatterns: ta*
                binder:
                  producer-properties:
                    key.serializer: "org.apache.kafka.common.serialization.ByteArraySerializer"
                    value.serializer: "org.apache.kafka.common.serialization.ByteArraySerializer"

[iguissouma]

A sample conf like this reproduce the problem:

spring:
  application:
    name: "spring-cloud-stream-multibinder"
  cloud:
    stream:
      bindings:
        input:
          destination: test1
          binder: kafka1
        output:
          destination: test2
          binder: kafka1

      binders:
        kafka1:
          type: kafka
          environment:
            spring.cloud.stream.kafka.binder:
              brokers: localhost:9092
        kafka2:
          type: kafka
          environment:
            spring.cloud.stream.kafka.binder:
              brokers: localhost:9092
      kafka:
        bindings:
          output:
            # this conf is working
            #producer:
            #  messageKeyExpression: headers['myKey'].getBytes()
        default:
          producer:
            messageKeyExpression: headers['myKey']

import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.support.MessageBuilder;

@SpringBootApplication
@EnableBinding(
	Processor.class
)
public class DemoApplication {

	public static void main(String[] args) {
		SpringApplication.run(DemoApplication.class, args);
	}

}

@Configuration
class Test implements CommandLineRunner{

	private final Processor processor;

	Test(Processor processor) {
		this.processor = processor;
	}

	@Override
	public void run(String... args) throws Exception {
		processor.output().send(
				MessageBuilder.withPayload("hello")
						.setHeader("myKey", "toto")
						.build()
		);
	}
}

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.6.6</version>
		<relativePath/> <!-- lookup parent from repository -->
	</parent>
	<groupId>com.example</groupId>
	<artifactId>demo</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<name>demo</name>
	<description>Demo project for Spring Boot</description>
	<properties>
		<java.version>11</java.version>
		<spring-cloud.version>2021.0.1</spring-cloud.version>
	</properties>
	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-amqp</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.cloud</groupId>
			<artifactId>spring-cloud-stream</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.cloud</groupId>
			<artifactId>spring-cloud-stream-binder-kafka</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.cloud</groupId>
			<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.kafka</groupId>
			<artifactId>spring-kafka</artifactId>
		</dependency>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework.amqp</groupId>
			<artifactId>spring-rabbit-test</artifactId>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework.cloud</groupId>
			<artifactId>spring-cloud-stream</artifactId>
			<scope>test</scope>
			<classifier>test-binder</classifier>
			<type>test-jar</type>
		</dependency>
		<dependency>
			<groupId>org.springframework.kafka</groupId>
			<artifactId>spring-kafka-test</artifactId>
			<scope>test</scope>
		</dependency>
	</dependencies>
	<dependencyManagement>
		<dependencies>
			<dependency>
				<groupId>org.springframework.cloud</groupId>
				<artifactId>spring-cloud-dependencies</artifactId>
				<version>${spring-cloud.version}</version>
				<type>pom</type>
				<scope>import</scope>
			</dependency>
		</dependencies>
	</dependencyManagement>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>

</project>

I'm still using legacy annotations.

***************************
APPLICATION FAILED TO START
***************************

Description:

Failed to bind properties under 'spring.cloud.stream.kafka.bindings.output' to org.springframework.cloud.stream.binder.kafka.properties.KafkaBindingProperties:

    Reason: org.springframework.core.convert.ConverterNotFoundException: No converter found capable of converting from type [java.lang.String] to type [org.springframework.cloud.stream.binder.kafka.properties.KafkaBindingProperties]

Action:

Update your application's configuration


Process finished with exit code 1

[luamas]

@iguissouma Don't use the default variable for now, The default variable is faulty

[iguissouma]

I noticed when we don specify environment property the default property is properly taken and it works as expected but it's not a solution.
@olegz can you please advice on this issue?

spring:
  application:
    name: "spring-cloud-stream-multibinder"
  cloud:
    stream:
      bindings:
        input:
          destination: test1
          binder: kafka1
        output:
          destination: test2
          binder: kafka1

      binders:
        kafka1:
          type: kafka
        kafka2:
          type: kafka
      kafka:
        default:
          producer:
            messageKeyExpression: headers['myKey']