Spring Cloud数据流自定义Scala处理器无法从Starter Apps（SCDF 2.5.1）发送/接收数据

Question

我一直在为Spring云数据流在Scala中创建一个简单的自定义处理器，并且在向starter应用程序发送/接收数据时遇到了问题。我无法看到任何消息在流中传播。流的定义是时间触发器。时间单位=秒|传递日志|日志其中传递日志是我的自定义处理器。

我使用的是Spring Cloud Data Flow 2.5.1和Spring Boot 2.2.6。

这是处理器使用的代码-我使用的是功能模型。

@SpringBootApplication
class PassThroughLog {

  @Bean
  def passthroughlog(): Function[String, String] = {
    input: String => {
      println(s"Received input `$input`")
      input
    }
  }
}

object PassThroughLog {
  def main(args: Array[String]): Unit = SpringApplication.run(classOf[PassThroughLog], args: _ *)
}

应用yml公司

spring:
  cloud:
    stream:
      function:
        bindings:
          passthroughlog-in-0: input
          passthroughlog-out-0: output

build.gradle.kts

// scala
implementation("org.scala-lang:scala-library:2.12.10")

// spring
implementation(platform("org.springframework.cloud:spring-cloud-dependencies:Hoxton.SR5"))
implementation(platform("org.springframework.cloud:spring-cloud-stream-dependencies:Horsham.SR5"))
implementation("org.springframework.boot:spring-boot-starter")
implementation("org.springframework.boot:spring-boot-starter-actuator")
implementation("org.springframework.cloud:spring-cloud-starter-function-web:3.0.7.RELEASE")
implementation("org.springframework.cloud:spring-cloud-starter-stream-kafka:3.0.5.RELEASE")

如果这里缺少代码示例，我已经将整个项目发布到github。我还在那里发布了日志，因为它们很长。

当我引导本地Kafka集群并将任意数据推送到输入主题时，我能够看到数据流过处理器。然而，当我在Spring云数据流上部署应用程序时，情况并非如此。我正在Kubernetes通过Docker部署该应用程序。

此外，当我使用定义<代码>时间>触发器部署流时。时间单位=秒| log，我在日志接收器中看到消息。这让我确信问题出在定制处理器上。

我是否错过了一些简单的东西，例如依赖项或额外配置？非常感谢任何帮助。

Answer 1

原来问题出在我的Dockerfile上。为了便于配置，我有一个build参数来指定入口点中使用的jar文件。为了实现这一点，我使用了ENTRYPOINT的shell版本。将我的入口点更改为exec版本解决了我的问题。

shell版本的ENTRYPOINT不能很好地使用图像参数（docker run

更改Dockerfile的位置：

FROM openjdk:11.0.5-jdk-slim as build
ARG JAR
ENV JAR $JAR
ADD build/libs/$JAR .
ENTRYPOINT java -jar $JAR

到

FROM openjdk:11.0.5-jdk-slim as build
ARG JAR
ADD build/libs/$JAR program.jar
ENTRYPOINT ["java", "-jar", "program.jar"]

已修复问题。

Answer 2

在SCDF中使用Spring Cloud Stream 3. x版本时，您必须设置一个附加属性，以让SCDF知道哪些通道绑定配置为输入和输出通道。

请参见：功能应用程序

特别注意以下特性：

应用程序。时间来源。Spring云流动作用绑定。timeSupplier-out-0=输出

应用程序。原木水槽。Spring云流动作用绑定。logConsumer-in-0=输入

在您的情况下，必须将函数绑定分别映射到输入和输出。