SCDF一对多处理器结合JPA流式输出返回null问题求助

问题根源 #

1. Stream被提前关闭：你用try-with-resources包裹了JPA返回的Stream，当方法return这个Stream时，try-with-resources会自动关闭它。Spring Cloud Stream处理时，Stream已经处于关闭状态，读不到任何数据，最终输出null。
2. Repository方法名不规范：FindByParentId首字母大写，不符合Spring Data JPA的查询方法命名规则（必须以小写find开头），Spring无法生成正确的查询实现，大概率返回空Stream。

修复步骤 #

1. 去掉try-with-resources，让Spring管理Stream生命周期 #

JPA的Stream和当前事务绑定，@Transactional注解会保证事务结束后自动关闭Stream，不需要手动关闭。修改处理器代码：

@Service
public class ProcessorFunction implements Function<Parent, Stream<Message<Child>>> {

    @Autowired
    private ChildRepository repo;

    @Override
    @Transactional(readOnly = true)
    public Stream<Message<Child>> apply(Parent parent) {
        return repo.findByParentId(parent.getId())
                   .map(GenericMessage::new);
    }
}

2. 修正Repository方法名 #

把方法名改成符合Spring Data规范的写法：

public interface ChildRepository extends JpaRepository<Child, Long> {
  
  Stream<Child> findByParentId(long id);
}

额外提醒 #

• 确认@Transactional(readOnly = true)注解正常生效，它是保证Stream在事务周期内可用的关键。
• 检查Parent的getId()返回值是否正确，确保能查询到对应的Child数据。

内容的提问来源于stack exchange，提问作者Juzer Ali