使用StreamBridge.send报错：SimpleFunctionRegistry返回空求助

问题背景 #

使用 Spring Boot 3.4.3 + Spring Cloud Stream Kafka Binder 4.3.1，调用StreamBridge.send()时出现以下错误，此前无此问题，日志未提供额外有效信息：

java.lang.RuntimeException: org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry$FunctionInvocationWrapper returned null
    at org.springframework.cloud.stream.function.StreamBridge.send(StreamBridge.java:228)
    at org.springframework.cloud.stream.function.StreamBridge.send(StreamBridge.java:178)

根据搜索到的信息，该错误通常因函数未正确注册或调用时函数未初始化导致：

The java.lang.RuntimeException: org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry$FunctionInvocationWrapper returned null error during StreamBridge.send calls typically arises when the function to invoke is not properly registered or is null at the time of invocation. This can occur in scenarios where a bean implementing SmartInitializingSingleton attempts to use StreamBridge.send before the function catalog is fully initialized, leading to a null function being cached and subsequently causing a NullPointerException. Similarly, if a consumer method attempts to use StreamBridge during application startup and the functionCatalog is not yet populated, the FunctionInvocationWrapper may be null, resulting in the same exception. The issue is exacerbated when the function registration happens after the afterSingletonsInstantiated phase, making the null state irrecoverable without a restart. This behavior is particularly problematic in environments like Azure Storage where message handlers are initialized early in the bootstrapping process.

已尝试延迟调用send()方法，但问题仍未解决。

排查与解决方案 #

1. 清除函数注册表缓存 #

首次调用StreamBridge.send()时若函数未就绪，会缓存null值，后续即使函数初始化完成也不会重新获取。可通过显式检查并刷新函数注册表解决：

@Autowired
private FunctionCatalog functionCatalog;
@Autowired
private StreamBridge streamBridge;

public void safeSend(String bindingName, Object payload) {
    // 检查函数是否存在，不存在则触发注册表刷新
    if (!functionCatalog.lookup(bindingName).isPresent()) {
        ((SimpleFunctionRegistry) functionCatalog).refresh();
    }
    streamBridge.send(bindingName, payload);
}

2. 验证绑定名称配置 #

确保调用StreamBridge.send()时传入的绑定名称与配置文件中的生产者绑定名称完全一致。例如配置：

spring:
  cloud:
    stream:
      bindings:
        my-out-0:
          destination: my-kafka-topic
          binder: kafka

调用时必须使用my-out-0作为绑定名称，而非目标topic名称my-kafka-topic。

3. 禁用函数缓存 #

添加配置禁用StreamBridge的函数缓存，确保每次调用都会重新查找函数：

spring.cloud.stream.function.cache.enabled=false

4. 调整Bean初始化顺序 #

确保使用StreamBridge的Bean在函数注册表初始化完成后再初始化：

• 使用@DependsOn指定依赖于函数注册表Bean：

@Service
@DependsOn("functionCatalog")
public class MessageSenderService {
    @Autowired
    private StreamBridge streamBridge;
    // 业务方法
}

• 或使用@Lazy延迟加载使用StreamBridge的Bean：

@Service
public class MessageSenderService {
    @Autowired
    @Lazy
    private StreamBridge streamBridge;
    // 业务方法
}

5. 检查Kafka Binder初始化状态 #

确认Kafka Binder是否正常初始化，排查集群连接配置是否正确：

@EventListener(ApplicationReadyEvent.class)
public void verifyBinderInitialization() {
    BinderFactory binderFactory = applicationContext.getBean(BinderFactory.class);
    Binder<?, ?, ?> kafkaBinder = binderFactory.getBinder("kafka", MessageChannel.class);
    log.info("Kafka Binder initialized status: {}", kafkaBinder != null);
}

内容的提问来源于stack exchange，提问作者Tim Dalsing