也许我只是错过了smth，但我只是不知道该去哪里找。
我读了来自两个来源的消息，基于一个公共密钥建立了一个连接，然后把它全部交给Kafka。

val env = StreamExecutionEnvironment.getExecutionEnvironment 
env.setParallelism(3) 
... 
source1 
 .keyBy(_.searchId) 
 .connect(source2.keyBy(_.searchId)) 
 .process(new SearchResultsJoinFunction) 
 .addSink(KafkaSink.sink)

因此，当我在本地启动它时，它可以很好地工作，而且它也可以在并行度设置为1的集群上工作，但是在并行度设置为3的集群上不再工作。
当我将其部署到1个作业管理器和3个任务管理器并使每个任务处于“运行”状态时，2分钟后（当没有任何任务提交到接收器时），其中一个任务管理器将获得以下日志：
https://gist.github.com/zavalit/1b1bf6621bed2a3848a05c1ef84c689c#file-文件1-txt-l108
整件事都结束了。
任何暗示我都会感激的。tnx，提前。