我有一个看起来像这样的简单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主题。
P.S。以防万一,在GitHub上的完整项目代码:https://github.com/mvmn/scdftest-transformer/tree/scfunc
为了在本地运行Kafka,Skipper,SCDF和SCDF控制台,请在app文件夹中执行mvn clean install
,然后在Coonsole中执行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
由于您正在使用编写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
您可以尝试一下,看看是否有区别吗?