我有一个简单的 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 主题。
PS 以防万一,GitHub 上的完整项目代码:https ://github.com/mvmn/scdftest-transformer/tree/scfunc
为了在本地运行启动 Kafka、Skipper、SCDF 和 SCDF 控制台,mvn clean install
在 app 文件夹中执行,然后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