我有一个工作正常的嵌入式作业，我想部署更多的协同作业。这些新添加的作业将接收来自旧作业的消息，并将其发送到kafka主题。
代码如下
@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)
这个错误很容易解释，希望我定义ingress。
链接中有一个定义类似的嵌入式模块->https://ci.apache.org/projects/flink/flink-statefun-docs-stable/sdk/modules.html#embedded-模块
新定义的模块将接收来自另一个模块的消息，并将其发送给kafka。
- 我必须为每个共处一地的作业定义入口吗？如果没有，我该怎么做
- 如何让同一地点的工作进行沟通？使用相同的FunctionType是否足够
- 位于同一位置的功能是否通过入口/出口进行通信
内联响应，仅供参考，您所询问的内容都不是特定于同一位置的。这些属性适用于包含共存和远程混合工作负载的远程模块和作业。
我必须为每个共处一地的作业定义入口吗？如果没有，我该怎么做？
是的，每个作业(远程或共址(至少需要一个入口。入口是一个将来自外部世界的消息消耗到statefun应用程序中的通道。想想卡夫卡或Kinesis。如果没有入口，作业将永远不会做任何事情，因为没有开始处理的初始消息。
对于每个入口，您将绑定1个或多个路由器，这些路由器从入口接收每条消息，并根据它们的功能类型将它们转发给0个或更多个功能[1]。
如何让同一地点的工作进行沟通？使用相同的FunctionType是否足够？
是的，函数只是使用它们的函数类型相互发送消息。
位于同一位置的功能是否通过入口/出口进行通信？
否，使用Apache Flink运行时在函数之间传递消息，该运行时包含高度优化的网络堆栈。一旦从入口提取消息，它就再也不会与该入口交互。如果你感兴趣，你可以在社区写的一些博客文章中阅读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