作为 Spring-Cloud-Data-Flow 流的一部分部署的 Spring-Cloud-Function 应用程序的错误 Kafka 主题名称

Question

我有一个简单的 SCDF 流，如下所示：

http --port=12346 | mvmn-transform | file --name=tmp.txt --directory=/tmp

mvmn-transform 是一个简单的自定义转换器，如下所示：

@SpringBootApplication
@EnableBinding(Processor.class)
@EnableConfigurationProperties(ScdfTestTransformerProperties.class)
@Configuration
public class ScdfTestTransformer {
    public static void main(String args[]) {
        SpringApplication.run(ScdfTestTransformer.class, args);
    }

    @Autowired
    protected ScdfTestTransformerProperties config;

    @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
    public Object transform(Message<?> message) {
        Object payload = message.getPayload();
        Map<String, Object> result = new HashMap<>();
        Map<String, String> headersStr = new HashMap<>();

        message.getHeaders().forEach((k, v) -> headersStr.put(k, v != null ? v.toString() : null));

        result.put("headers", headersStr);
        result.put("payload", payload);
        result.put("configProp", config.getSomeConfigProp());

        return result;
    }

    // See https://stackoverflow.com/questions/59155689/could-not-decode-json-type-for-key-file-name-in-a-spring-cloud-data-flow-stream
    @Bean("kafkaBinderHeaderMapper")
    public KafkaHeaderMapper kafkaBinderHeaderMapper() {
        BinderHeaderMapper mapper = new BinderHeaderMapper();
        mapper.setEncodeStrings(true);
        return mapper;
    }
}

这很好用。

但我读到 Spring Cloud Function 应该允许我实现这样的应用程序而无需指定绑定和转换器注释，所以我将其更改为：

@SpringBootApplication
// @EnableBinding(Processor.class)
@EnableConfigurationProperties(ScdfTestTransformerProperties.class)
@Configuration
public class ScdfTestTransformer {
    public static void main(String args[]) {
        SpringApplication.run(ScdfTestTransformer.class, args);
    }

    @Autowired
    protected ScdfTestTransformerProperties config;

    // @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
    @Bean
    public Function<Message<?>, Map<String, Object>> transform(
    // Message<?> message
    ) {
        return message -> {
            Object payload = message.getPayload();
            Map<String, Object> result = new HashMap<>();
            Map<String, String> headersStr = new HashMap<>();

            message.getHeaders().forEach((k, v) -> headersStr.put(k, v != null ? v.toString() : null));

            result.put("headers", headersStr);
            result.put("payload", payload);
            result.put("configProp", "Config prop val: " + config.getSomeConfigProp());

            return result;
        };
    }

    // See https://stackoverflow.com/questions/59155689/could-not-decode-json-type-for-key-file-name-in-a-spring-cloud-data-flow-stream
    @Bean("kafkaBinderHeaderMapper")
    public KafkaHeaderMapper kafkaBinderHeaderMapper() {
        BinderHeaderMapper mapper = new BinderHeaderMapper();
        mapper.setEncodeStrings(true);
        return mapper;
    }
}

现在我遇到了一个问题 - Spring-Cloud-Function 显然忽略了 SCDF 源和目标主题名称，而是创建了主题 transform-in-0 和 transform-out-0。

SCDF 创建名称类似于<stream-name>.<app-name> 的主题，例如TestStream123.http 和TestStream123.mvmn-transform

以前它们被用于转换器 - 应该如此，因为它是 SCDF 流的一部分。但是现在它们被 Spring-Cloud-Function 忽略了，而是创建了 transform-in-0 和 transform-out-0。

因此，我的转换器不再接收任何输入，因为它期望它在错误的 Kafka 主题上。并且可能也不会对流产生任何输出，因为它也会输出到错误的 Kafka 主题。

附：以防万一，GitHub上的完整项目代码：https://github.com/mvmn/scdftest-transformer/tree/scfunc

为了在本地运行启动 Kafka、Skipper、SCDF 和 SCDF 控制台，请在 app 文件夹中执行 mvn clean install，然后在控制台中执行 app register --name mvmn-transform-1 --type processor --uri maven://x.mvmn.study.scdf.scdftest:scdftest-transformer:0.1.1-SNAPSHOT --metadata-uri maven://x.mvmn.study.scdf.scdftest:scdftest-transformer:0.1.1-SNAPSHOT。然后你可以使用定义http --port=12346 | mvmn-transform | file --name=tmp.txt --directory=/tmp部署流

Answer 1

由于您使用的是编写 Spring Cloud Stream 应用程序的功能模型，因此在部署此应用程序时，您需要在自定义处理器上传递两个属性来恢复 Spring Cloud Data Flow 行为。

spring.cloud.stream.function.bindings.transform-in-0=input spring.cloud.stream.function.bindings.transform-out-0=output

你可以试试看，看看会不会有什么不同？