配置 Spring Cloud Task 使用 Spring Cloud Data Flow server 的 Kafa

Question

我有一个Spring Cloud Data Flow (SCDF) 服务器在Kubernetes cluster 上运行，Kafka 作为消息代理。现在我正在尝试启动一个Spring Cloud Task (SCT) 写入Kafka 中的主题。我希望 SCT 使用与 SCDF 相同的 Kafka。这提出了我的两个问题，希望他们能得到解答：

1. 如何配置 SCT 以使用与 SCDF 相同的 Kafka？
2. 是否可以配置 SCT，以便 Kafka 服务器 uri 在启动时可以自动传递给 SCT，类似于 在启动时传递给 SCT 的数据源属性？

由于我找不到任何有关如何实现此目的的示例，非常感谢您的帮助。

编辑：我自己的答案

这就是我如何让它适用于我的案例。我的 SCT 要求提供 spring.kafka.bootstrap-servers。在 SCDF 的 shell 中，我将其作为参数 --spring.kafka.bootstrap-servers=${KAFKA_SERVICE_HOST}:${KAFKA_SERVICE_PORT} 提供，其中 KAFKA_SERVICE_HOST 和 KAFKA_SERVICE_PORT 是由 SCDF 的 k8s 设置脚本创建的环境变量。

这是在 SCDF 的 shell 中启动任务的方法

dataflow:>task launch --name sample-task --arguments "--spring.kafka.bootstrap-servers=${KAFKA_SERVICE_HOST}:${KAFKA_SERVICE_PORT}"

Answer 1

您可能需要查看参考指南中的Spring Cloud Task Events 部分。

期望您选择选择的活页夹并将该库打包到任务应用程序的类路径中。使用该依赖项，您将使用 Spring Cloud Stream 的 Kafka binder properties（例如 spring.cloud.stream.kafka.binder.brokers）和其他与连接到现有 Kafka 集群相关的应用程序来配置应用程序。

使用这些配置启动任务应用程序（从 SCDF）后，您就可以在任务应用程序中发布或接收事件。

或者，使用 Task 应用程序的类路径中的 Kafka-binder，您可以通过全局配置为 SCDF 启动的所有 Task 定义 Kafka binder 属性。请参阅参考文献中的Common Application Properties。指南以获取更多信息。在此模型中，您不必为每个任务应用程序显式配置 Kafka 属性，而是 SCDF 会在启动任务时自动传播它们。请记住，这些属性将提供给所有任务启动。