Netflix Conductor type:KAFKA_PUBLISH - Failed to construct k

Question

我在我的工作流中创建了这个任务:

"tasks": [
{
"name": "get_users_list",
"taskReferenceName": "get_users_list",
"inputParameters": {
"http_request": {
"uri": "https://reqres.in/api/users?page=${workflow.input.pagenumber}",
"method": "GET",
"contentType": "application/json",
"connectionTimeOut": "36000",
"readTimeOut": "36000"
}
},
"type": "HTTP",
"decisionCases": {},
"defaultCase": [],
"forkTasks": [],
"startDelay": 0,
"joinOn": [],
"optional": false,
"defaultExclusiveJoinTask": [],
"asyncComplete": false,
"loopOver": []
},
{
"name": "get_user_details",
"taskReferenceName": "get_user_details",
"inputParameters": {
"http_request": {
"uri": "https://reqres.in/api/users/${get_users_list.output.response.body.data[0].id}",
"method": "GET",
"contentType": "application/json",
"connectionTimeOut": "36000",
"readTimeOut": "36000"
}
},
"type": "HTTP",
"decisionCases": {},
"defaultCase": [],
"forkTasks": [],
"startDelay": 0,
"joinOn": [],
"optional": false,
"defaultExclusiveJoinTask": [],
"asyncComplete": false,
"loopOver": []
},
{
"name": "call_kafka",
"taskReferenceName": "call_kafka",
"inputParameters": {
"kafka_request": {
"topic": "transaction-1",
"value": "${get_user_details.output.response.body.data.first_name}",
"bootStrapServers": "kafka:9092"
},
"key": "",
"keySerializer": "org.apache.kafka.common.serialization.StringSerializer"
},
"type": "KAFKA_PUBLISH",
"decisionCases": {},
"defaultCase": [],
"forkTasks": [],
"startDelay": 0,
"joinOn": [],
"optional": false,
"defaultExclusiveJoinTask": [],
"asyncComplete": false,
"loopOver": []
}

,

当前两个任务完成时，call_kafka"此错误导致任务失败，

"调用kafka任务失败，原因是:org.apache.kafka.common.KafkaException: Failed to construct kafkaproducer"

我是新来的Netflix导体，并试图通过工作流发布消息到Kafka。如果有任何错误，请纠正我，并建议解决这个问题。提前感谢

Answer 1

Kafka任务已移至社区repo。请查看:https://github.com/Netflix/conductor/discussions/2961

在conductor/server/build.gradle文件中添加以下依赖项:

实现com.netflix.conductor:conductor-kafka:3.8.0