本地Pubsub仿真器不适用于Dataflow [英] Local Pubsub Emulator won't work with Dataflow
问题描述
我正在用Java开发Dataflow，输入来自Pubsub.后来，我在此处上看到了有关如何使用本地Pubsub模拟器的指南，因此我不会需要部署到GCP才能进行测试.
I am developing Dataflow in Java, the input comes from a Pubsub. Later, I saw a guide here on how to use local Pubsub emulator so I would not need to deploy to GCP in order to test.
这是我的简单代码:
private interface Options extends PipelineOptions, PubsubOptions, StreamingOptions {
@Description("Pub/Sub topic to read messages from")
String getTopic();
void setTopic(String topic);
@Description("Pub/Sub subscription to read messages from")
String getSubscription();
void setSubscription(String subscription);
@Description("Local file output")
String getOutput();
void setOutput(String output);
}
public static void main(String[] args) {
Options options = PipelineOptionsFactory
.fromArgs(args)
.withValidation()
.as(Options.class);
options.setStreaming(true);
options.setPubsubRootUrl("localhost:8085");
Pipeline pipeline = Pipeline.create(options);
pipeline
.apply("IngestFromPubsub", PubsubIO.readStrings().fromTopic(options.getTopic()))
// other .apply's
pipeline.run();
}
我能够按照指南进行操作，包括需要使用示例Python代码创建主题，订阅，发布者甚至发布消息的部分.当我使用Python代码与Pubsub仿真器进行交互时，我在运行仿真器的命令行中注意到消息Detected HTTP/2 connection:
I was able to follow the guide, including the part where I need to use the example Python code to create topic, subscription, publisher and even publish messages. When I use the Python code to interact with the Pubsub emulator, I notice the message Detected HTTP/2 connection in the command-line where I run the emulator:
Executing: cmd /c C:\...\google-cloud-sdk\platform\pubsub-emulator\bin\cloud-pubsub-emulator.bat --host=localhost --port=8085
[pubsub] This is the Google Pub/Sub fake.
[pubsub] Implementation may be incomplete or differ from the real system.
[pubsub] Apr 10, 2020 3:33:26 PM com.google.cloud.pubsub.testing.v1.Main main
[pubsub] INFO: IAM integration is disabled. IAM policy methods and ACL checks are not supported
[pubsub] Apr 10, 2020 3:33:26 PM io.gapi.emulators.netty.NettyUtil applyJava7LongHostnameWorkaround
[pubsub] INFO: Unable to apply Java 7 long hostname workaround.
[pubsub] Apr 10, 2020 3:33:27 PM com.google.cloud.pubsub.testing.v1.Main main
[pubsub] INFO: Server started, listening on 8085
[pubsub] Apr 10, 2020 3:34:38 PM io.gapi.emulators.grpc.GrpcServer$3 operationComplete
[pubsub] INFO: Adding handler(s) to newly registered Channel.
[pubsub] Apr 10, 2020 3:34:38 PM io.gapi.emulators.netty.HttpVersionRoutingHandler channelRead
[pubsub] INFO: Detected HTTP/2 connection.
[pubsub] Apr 10, 2020 3:34:52 PM io.gapi.emulators.grpc.GrpcServer$3 operationComplete
[pubsub] INFO: Adding handler(s) to newly registered Channel.
[pubsub] Apr 10, 2020 3:34:52 PM io.gapi.emulators.netty.HttpVersionRoutingHandler channelRead
[pubsub] INFO: Detected HTTP/2 connection.
我使用Dataflow Pipeline Run Configuration在Eclipse中编译/运行了代码，但是遇到了问题.
I compiled/run the code in Eclipse using Dataflow Pipeline Run Configuration, but I get a problem.
Exception in thread "main" org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.RuntimeException: Failed to create subscription:
...
Caused by: java.lang.RuntimeException: Failed to create subscription:
at org.apache.beam.sdk.io.gcp.pubsub.PubsubUnboundedSource.createRandomSubscription(PubsubUnboundedSource.java:1427)
...
Caused by: java.lang.IllegalArgumentException: java.net.MalformedURLException: unknown protocol: localhost
...
Caused by: java.net.MalformedURLException: unknown protocol: localhost
当我尝试在options.setPubsubRootUrl("localhost:8085")行中添加http时，出现了一个无限重复的异常:
When I try to add http in the line options.setPubsubRootUrl("localhost:8085"), I get an infinitely repeated exception:
com.google.api.client.http.HttpRequest execute
WARNING: exception thrown while executing request
java.net.ConnectException: Connection refused: connect
at java.net.DualStackPlainSocketImpl.waitForConnect(Native Method)
at java.net.DualStackPlainSocketImpl.socketConnect(Unknown Source)
at java.net.AbstractPlainSocketImpl.doConnect(Unknown Source)
at java.net.AbstractPlainSocketImpl.connectToAddress(Unknown Source)
at java.net.AbstractPlainSocketImpl.connect(Unknown Source)
at java.net.PlainSocketImpl.connect(Unknown Source)
at java.net.SocksSocketImpl.connect(Unknown Source)
它似乎已到达Pubsub模拟器，但无法连接，因为我运行该模拟器的命令行也会无限生成此内容:
It seems to reach the Pubsub emulator but can't connect as the command-line where I run the emulator generates this infinitely also:
[pubsub] Apr 10, 2020 3:49:30 PM io.gapi.emulators.grpc.GrpcServer$3 operationComplete
[pubsub] INFO: Adding handler(s) to newly registered Channel.
[pubsub] Apr 10, 2020 3:49:30 PM io.gapi.emulators.netty.HttpVersionRoutingHandler channelRead
[pubsub] INFO: Detected non-HTTP/2 connection.
如何使我的数据流与Pubsub模拟器一起使用?
How can I make my Dataflow work with Pubsub emulator?
推荐答案
您正尝试使用Beam 2.5 SDK的Dataflow分支从Beam Direct Runner连接到Pubsub仿真器.自2019年6月6日起已弃用Dataflow 2.5 SDK和Eclipse插件.但这应该可行.
You are attempting to connect to the Pubsub emulator from the Beam Direct Runner, using the Dataflow fork of the Beam 2.5 SDK. The Dataflow 2.5 SDK and Eclipse plugin were deprecated as of June 6, 2019. However this should work.
您已经发现，在Beam中，您的PubsubRootUrl必须以'http://'开头.您看到的第二个问题表明localhost:8085上没有任何内容在监听.这可能是因为实际上有2个本地主机:IPv4和IPv6. Pubsub仿真器仅在IPv4上侦听，而Windows首先尝试IPv6.尝试将localhost替换为127.0.0.1以强制使用IPv4.您应该以此结束:
You need to prefix your PubsubRootUrl with 'http://' in Beam, as you've discovered. The second problem you are seeing indicates that nothing is listening on localhost:8085. This is likely because there are actually 2 localhosts: IPv4 and IPv6. The Pubsub Emulator only listens on IPv4 and Windows tries IPv6 first. Try replacing localhost with 127.0.0.1 to force IPv4. You should end up with this:
options.setPubsubRootUrl("http://127.0.0.1:8085")
这篇关于本地Pubsub仿真器不适用于Dataflow的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
