我有一个正常工作的嵌入式作业，我想部署其他位于同一位置的作业。这些新添加的作业将接收来自旧作业的消息并将其发送给kafka topic。
代码如下
@AutoService(StatefulFunctionModule.class)
public final class CoLocatedModule implements StatefulFunctionModule {
@Override
public void configure(Map<String, String> globalConfiguration, Binder binder) {
FunctionProvider provider = new FunctionProvider();
binder.bindFunctionProvider( CoLocated.TYPE,provider );
binder.bindEgress(KafkaSpecs.TO_TRANSACTION_SPEC);
}
}我得到一个错误如下
The program finished with the following exception:
org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: There are no ingress defined.
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:335)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:205)
at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:138)
at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:664)
at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:213)
at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:895)
at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:968)
at org.apache.flink.runtime.security.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30)
at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:968)
Caused by: java.lang.IllegalStateException: There are no ingress defined.
at org.apache.flink.statefun.flink.core.StatefulFunctionsUniverseValidator.validate(StatefulFunctionsUniverseValidator.java:25)
at org.apache.flink.statefun.flink.core.StatefulFunctionsJob.main(StatefulFunctionsJob.java:66)
at org.apache.flink.statefun.flink.core.StatefulFunctionsJob.main(StatefulFunctionsJob.java:41)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:567)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:321)这个错误是不言自明的，我想让我定义入口。
链接中有一个类似定义的嵌入式模块->https://ci.apache.org/projects/flink/flink-statefun-docs-stable/sdk/modules.html#embedded-模块
新定义的模块将从另一个模块接收消息并将它们发送给Kafka。
我是否必须为每个位于同一位置的作业定义入口？如果不是，我怎么能让这工作？
我怎样才能找到同一地点的工作来沟通？使用相同的functiontype是否足够？
位于同一位置的功能是否通过入口/出口进行通信？
1条答案
按热度按时间6gpjuf901#
回复内联，仅供参考，您所问的任何问题都不是同一地点的具体问题。这些属性适用于包含同地和远程混合工作负载的远程模块和作业。
我是否必须为每个位于同一位置的作业定义入口？如果不是，我怎么能让这工作？
是的，每个作业（远程或同位）至少需要一个入口。入口是将来自外部世界的消息消费到statefun应用程序的通道。想想Kafka或者动觉。如果没有入口，作业将永远不会执行任何操作，因为不会有初始消息来开始处理。
对于每个入口，您将绑定1个或多个路由器，这些路由器从入口获取每条消息，并根据它们的函数类型将它们转发给0个或多个函数[1]。
我怎样才能找到同一地点的工作来沟通？使用相同的functiontype是否足够？
是的，函数只是使用它们的函数类型相互传递消息。
位于同一位置的功能是否通过入口/出口进行通信？
不，消息是使用ApacheFlink运行时在函数之间传递的，该运行时包含高度优化的网络堆栈。一旦从入口中提取消息，它就再也不会与该入口交互。如果您感兴趣，可以在社区撰写的一些博客文章中了解flink的网络堆栈是如何工作的，但这并不是在产品中成功使用statefun的必要条件[2]。
[1] https://ci.apache.org/projects/flink/flink-statefun-docs-release-2.0/io-module/index.html#router
[2]https://flink.apache.org/2019/06/05/flink-network-stack.html