Repository navigation
PubSubLite Not Advancing Watermarks in Dataflow/Beam Job #542
Description
In the job below a relatively low volume, for testing purposes, PubSubLite topic is consumed and intended to be written to a GCS bucket in Windowed files. It appears that all activities are performed but the Windows/Watermarks never advance or close so the actual files are never written.
Environment details
- OS type and version: Windows -> Dataflow GCP
- Java version: 1.8
- pubsublite version(s):
<dependency>
<groupId>com.google.cloud</groupId>
<artifactId>pubsublite-beam-io</artifactId>
<version>0.11.1</version>
</dependency>
Steps to reproduce
- Create PubSub Subscriber in DataFlow Job
- Adding Windowing, Trigger, etc.
- Convert SequencedMessage
- Attempt to write out to Cloud Storage
Code example
public static void main(String[] args) {
/******* some other setup, etc. *************/
Pipeline pipeline = Pipeline.create(options);
SubscriptionPath subPath = SubscriptionPath.newBuilder()
.setLocation(CloudZone.parse(options.getTargetProjectZone()))
.setProject(ProjectNumber.of(options.getTargetProjectNumber()))
.setName(SubscriptionName.of(options.getTargetPubsubSubscription()))
.build();
/* unused */
FlowControlSettings flowControlSettings = FlowControlSettings.builder()
// 10 MiB. Must be greater than the allowed size of the largest message (1 MiB)
.setBytesOutstanding(10 * 1024 * 1024L)
// 1,000 outstanding messages. Must be >0
.setMessagesOutstanding(1000L)
.build();
SubscriberOptions subOptions = SubscriberOptions.newBuilder()
.setSubscriptionPath(subPath)
//.setFlowControlSettings(flowControlSettings)
.build();
Window<SequencedMessage> window = Window.<SequencedMessage>into(FixedWindows.of(Duration.standardMinutes(1)))
.triggering(Repeatedly.forever(AfterWatermark.pastEndOfWindow()
.withEarlyFirings(AfterProcessingTime.pastFirstElementInPane()
.alignedTo(Duration.standardSeconds(60)))))
.withAllowedLateness(Duration.standardMinutes(1), Window.ClosingBehavior.FIRE_ALWAYS)
.discardingFiredPanes();
pipeline
.apply("Read messages", PubsubLiteIO.read(subOptions))
.apply(window)
.apply(ParDo.of(new ConvertToStrings()))
.apply(TextIO.write().to(options.getOutputDirectory()).withWindowedWrites().withNumShards(10));
pipeline.run();
}
static class ConvertToStrings extends DoFn<SequencedMessage, String> {
/**
*
*/
private static final long serialVersionUID = -7457075599298968462L;
@ProcessElement
public void processElement(@Element SequencedMessage word, OutputReceiver<String> out) {
String messageString = word.getMessage().getData().toStringUtf8();
LOG.info(messageString);
// Use OutputReceiver.output to emit the output element.
out.output(messageString);
}
}
Stack trace
No indications of errors are written to the logs at all. Other than:
"*~*~*~ Channel ManagedChannelImpl{logId=1017, target=us-east1-pubsublite.googleapis.com:443} was not shutdown properly!!! ~*~*~*
Make sure to call shutdown()/shutdownNow() and wait until awaitTermination() returns true."
exception: "java.lang.RuntimeException: ManagedChannel allocation site
at io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference.<init>(ManagedChannelOrphanWrapper.java:93)
at io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:53)
at io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:44)
at io.grpc.internal.AbstractManagedChannelImplBuilder.build(AbstractManagedChannelImplBuilder.java:518)
at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.createSingleChannel(InstantiatingGrpcChannelProvider.java:314)
at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.access$1600(InstantiatingGrpcChannelProvider.java:71)
at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider$1.createSingleChannel(InstantiatingGrpcChannelProvider.java:210)
at com.google.api.gax.grpc.ChannelPool.create(ChannelPool.java:72)
at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.createChannel(InstantiatingGrpcChannelProvider.java:217)
at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.getTransportChannel(InstantiatingGrpcChannelProvider.java:200)
at com.google.api.gax.rpc.ClientContext.create(ClientContext.java:169)
at com.google.cloud.pubsublite.v1.stub.GrpcTopicStatsServiceStub.create(GrpcTopicStatsServiceStub.java:80)
at com.google.cloud.pubsublite.v1.stub.TopicStatsServiceStubSettings.createStub(TopicStatsServiceStubSettings.java:108)
at com.google.cloud.pubsublite.v1.TopicStatsServiceClient.<init>(TopicStatsServiceClient.java:136)
at com.google.cloud.pubsublite.v1.TopicStatsServiceClient.create(TopicStatsServiceClient.java:117)
at com.google.cloud.pubsublite.internal.TopicStatsClientSettings.instantiate(TopicStatsClientSettings.java:60)
at com.google.cloud.pubsublite.internal.TopicStatsClient.create(TopicStatsClient.java:29)
at com.google.cloud.pubsublite.beam.TopicBacklogReaderSettings.instantiate(TopicBacklogReaderSettings.java:85)
at com.google.cloud.pubsublite.beam.SubscriberOptions.getBacklogReader(SubscriberOptions.java:164)
at com.google.cloud.pubsublite.beam.SubscribeTransform.newRestrictionTracker(SubscribeTransform.java:91)
at com.google.cloud.pubsublite.beam.PerSubscriptionPartitionSdf.newTracker(PerSubscriptionPartitionSdf.java:108)
at com.google.cloud.pubsublite.beam.PerSubscriptionPartitionSdf$DoFnInvoker.invokeNewTracker(Unknown Source)
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SplittableParDoViaKeyedWorkItems$ProcessFn.processElement(SplittableParDoViaKeyedWorkItems.java:482)
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SplittableParDoViaKeyedWorkItems$ProcessFn$DoFnInvoker.invokeProcessElement(Unknown Source)
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:232)
at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:188)
at org.apache.beam.runners.dataflow.worker.DataflowProcessFnRunner.processElement(DataflowProcessFnRunner.java:63)
at org.apache.beam.runners.dataflow.worker.SimpleParDoFn.processElement(SimpleParDoFn.java:339)
at org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process(ParDoOperation.java:44)
at org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:49)
at org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.runReadLoop(ReadOperation.java:212)
at org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.start(ReadOperation.java:163)
at org.apache.beam.runners.dataflow.worker.util.common.worker.MapTaskExecutor.execute(MapTaskExecutor.java:92)
at org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.process(StreamingDataflowWorker.java:1426)
at org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.access$1100(StreamingDataflowWorker.java:163)
at org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker$7.run(StreamingDataflowWorker.java:1105)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
External references such as API reference guides
- We've looked at a lot of documentation. Tried many many variations. The above is the simplest variant we could get down to.
- This is reminiscent of items mentioned in here https://stackoverflow.com/questions/58221372/what-actually-manages-watermarks-in-beam
Any additional information below
For context, we currently process 100's of billions of monthly messages via PubSub with a fundamentally similar dataflow job. This was our attempt at a POC for PubSubLite. It seems like the Watermark and Window closing behavior is incorrect.
The following image shows the "WriteShardedBundlesToTempFiles" step which is clearing getting elements and data, but the follow on steps never fire and the GCS file is never written.
Activity
GLStephen commentedon Mar 7, 2021
Update: the config above did write files after about 10 minutes. The addition of "withEarlyFirings" was the change that led to it.
This was removed from our PubSub job when carried over to the PubSubLite POC because certain triggering firing and window combinations warn about data loss.
As mentioned in the linked stack overflow post the watermark behavior seems unpredictable for testing purposes. With the job left running it predictably handled later messages pumped in. It just seems to effect the initial state. Is there anything to do here config wise to get a more reliable initial state for testing purposes?
dpcollins-google commentedon Mar 17, 2021
Hi Stephen- sorry I missed this. Pub/Sub Lite uses a monotonically increasing publish time watermark which will not advance in the absence of messages. In addition, we restrict bundle closing until having received 10x the byte limit of the flow control on a given partition. Either of these behaviors could be causing your initial window closing slowness, but the first is for correctness, and the second for performance.
For testing use, I'll look into adding a parameter to disable the bundle closing restriction, although this will decrease your throughput.
GLStephen commentedon Mar 17, 2021
@dpcollins-google thanks for the info, if I set flow control settings very low in the client can I force it to trigger, or is that a factor of the service backend side configuration? if messages to an endpoint are bursty, could messages end up trapped? In other words, very high rates only occasionally needed. Could you end up with "dangling" messages in Dataflow, never closed, but never enough volume to close for a long time?
dpcollins-google commentedon Mar 17, 2021
You cannot. Setting flow control bytes lower than the size of an individual message will cause the service to not send you any messages.
The bundles will be allowed to close every 45 seconds, even if there is low volume. This ensures no messages will become stuck in the way you described, at least in the PubsubLiteIO implementation. I don't know if windowing could potentially be stalled if the watermark did not advance (if there were no new messages for a while)- that would be a question for beam/dataflow people.