每次启动应用程序时都会出现“创建名为‘functionBindingRegistrar’的bean错误”异常

Question

我有一个普通的spring云流应用程序，它简单地从Kafka主题读取数据并生成消息到另一个Kafka主题，请参阅下面的配置：

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.5.6</version>
    <relativePath/>
</parent>

<properties>
    <spring-cloud.version>2020.0.4</spring-cloud.version>
    <spring-boot-maven-plugin.version>2.3.0.RELEASE</spring-boot-maven-plugin.version>
</properties>
<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>
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-kafka</artifactId>
        <exclusions>
            <exclusion>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-tomcat</artifactId>
            </exclusion>
        </exclusions>
    </dependency>
</dependencies>

和下面的application.proeprties

#Kafka Configurations
spring.kafka.bootstrap-servers=localhost:9092
spring.cloud.stream.kafka.binder.configuration.auto.offset.reset=latest

spring.cloud.function.definition=merchantCredentials;validatedProducts;validateImages;retryUnprocessedItems
#Input topics
#Merchants
spring.cloud.stream.bindings.merchantCredentials-in-0.destination=mis.merchantCtpCredentials
spring.cloud.stream.kafka.bindings.merchantCredentials-in-0.consumer.ack-mode=manual_immediate
spring.cloud.stream.bindings.merchantCredentials-in-0.contentType=application/json
spring.cloud.stream.bindings.merchantCredentials-in-0.consumer.header-mode=headers
spring.cloud.stream.bindings.merchantCredentials-in-0.consumer.partitioned=true
spring.cloud.stream.bindings.merchantCredentials-in-0.consumer.max-attempts=1
spring.cloud.stream.bindings.merchantCredentials-in-0.group=tuevGroup

#kfc.notifications.product
spring.cloud.stream.bindings.validatedProducts-in-0.destination=kfc.notifications.product
spring.cloud.stream.kafka.bindings.validatedProducts-in-0.consumer.ack-mode=manual_immediate
spring.cloud.stream.bindings.validatedProducts-in-0.contentType=application/json
spring.cloud.stream.bindings.validatedProducts-in-0.consumer.header-mode=headers
spring.cloud.stream.bindings.validatedProducts-in-0.consumer.concurrency=5
spring.cloud.stream.bindings.validatedProducts-in-0.consumer.partitioned=true
spring.cloud.stream.bindings.validatedProducts-in-0.consumer.max-attempts=1
spring.cloud.stream.bindings.validatedProducts-in-0.group=tuevGroup

#marketplace.products
spring.cloud.stream.bindings.validateImages-in-0.destination=marketplace.products
spring.cloud.stream.kafka.bindings.validateImages-in-0.consumer.ack-mode=manual_immediate
spring.cloud.stream.bindings.validateImages-in-0.contentType=application/json
spring.cloud.stream.bindings.validateImages-in-0.consumer.header-mode=headers
spring.cloud.stream.bindings.validateImages-in-0.consumer.partitioned=true
spring.cloud.stream.bindings.validateImages-in-0.consumer.max-attempts=1
spring.cloud.stream.bindings.validateImages-in-0.group=tuevGroup

#Output topics
#productValidated
spring.cloud.stream.bindings.validatedProducts-out-0.destination=marketplace.validated.products
spring.cloud.stream.bindings.validatedProducts-out-0.contentType=application/json
spring.cloud.stream.bindings.validatedProducts-out-0.producer.partition-count=10
spring.cloud.stream.bindings.validatedProducts-out-0.producer.header-mode=headers

spring.cloud.stream.bindings.retryUnprocessedItems-out-0.destination=marketplace.validated.products
spring.cloud.stream.bindings.retryUnprocessedItems-out-0.contentType=application/json
spring.cloud.stream.bindings.retryUnprocessedItems-out-0.producer.partition-count=10
spring.cloud.stream.bindings.retryUnprocessedItems-out-0.producer.header-mode=headers
spring.cloud.stream.poller.cron=0 0/10 * * * *
spring.cloud.stream.poller.initial-delay=10000

下面是所有定义的spring云函数的签名。

@Bean
public Consumer<Flux<Message<JsonNode>>> merchantCredentials() {

@Bean
public Function<Message<NotificationMessage>, Message<ProductValidatedEvent>> validatedProducts() {

@Bean
public Consumer<Message<ProductImportMessage>> validateImages() {

@PollableBean
@SchedulerLock(name = "retryProcess_scheduledTask", lockAtMostFor = "${retry.job.lock.atMost}", lockAtLeastFor = "${retry.job.lock.atLeast}")
public Supplier<Flux<Message<ProductValidatedEvent>>> retryUnprocessedItems() {

一切正常工作，应用程序启动并正常工作，不过，在日志中，我多次遇到这个异常，特别是在应用程序的启动阶段。

org.springframework.boot.SpringApplication - Application run failed org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'functionBindingRegistrar' defined in class path resource [org/springframework/cloud/stream/function/FunctionConfiguration.class]: Invocation of init method failed; nested exception is java.lang.IllegalArgumentException: Type must be one of Supplier, Function or Consumer

我已经检查了所有的配置，我仍然不知道如何防止这个问题发生。为什么会出现这种情况？这是可以忽略的吗？

更新1

我在spring框架( FunctionTypeUtils )中跟踪了这个bug。

    public static Type discoverFunctionTypeFromClass(Class<?> functionalClass) {
    Assert.isTrue(isFunctional(functionalClass), "Type must be one of Supplier, Function or Consumer");

此函数在FunctionConfiguration中由该函数调用。

private String[] filterEligibleFunctionDefinitions() {
...
                for (int i = 0; i < functionNames.length && eligibleDefinition; i++) {
                    String functionName = functionNames[i];
                    if (this.applicationContext.containsBean(functionName)) {

当我将调试点添加到这个和前一个调试点时，我得到了以下输出

functionName: merchantCredentials
functionalClass: com.rewedigital.services.tuev.marketplace.merchant.flow.MerchantFlowManger$$Lambda$1323/0x00000008008fc040
functionName: validatedProducts
functionalClass: com.rewedigital.services.tuev.marketplace.validator.listener.ProductChangedListener$$Lambda$1331/0x00000008008fa040
functionName: validateImages
functionalClass: com.rewedigital.services.tuev.marketplace.sieve.listener.ProductImagesListener$$Lambda$1324/0x00000008008fc440
functionName: retryUnprocessedItems
functionalClass: org.springframework.beans.factory.support.NullBean

表明retryUnprocessedItems是罪魁祸首，但不知道原因何在？

Answer 1

经过调查，发现问题主要是与@SchedulerLock注释有关。

我注意到，当shedLock表对方法添加了锁时，就会发生这个问题，因此它阻止了FunctionBeanRegistrar添加方法，因此出现了异常。

当然，这在另一方面意味着注释现在被认为是不可用的，因为@PollableBean注释真正运行的不是函数本身，而是函数返回的供应者lambda表达式，实际上使ShedLock变得无用。

一旦我删除了注释，所有的例外都消失了，太阳再次照耀，鸟儿歌唱等等。

下一个要回答的问题是如何以分布式的方式使用可滚动bean，但这不在这个问题的范围之内。