Apache Beam Pipeline and Poison Pills

userLC4m 0
Viewed 65

I am running an Apache Beam pipeline using GCP Dataflow and got this error from worker:

Error message from worker: java.lang.RuntimeException: java.lang.IllegalStateException: Got poison pill or timeout but stream is not done

Actually got bunch of them in ~2 minutes.

I am using the pipeline to write messages from PubSub to BigQuery. Inside the pipeline, when converting PubSub messages to TableRows, I am using a FailsafeElement<PubsubMessage, String> and have also dead letter tables in BigQuery so I can write there failed converted messages. But from what I have read about Poison Pills it seems like the problem is before converting the PubSub messages, in the stage of consuming them from the PubSub subscription (I have looked into all my dead letter tables and didn't see any events from the time this error occurred).

Am I right with my conclusion? And if so, is there any way to catch these poison pills with Apache Beam and log/write them somewhere?

Update:

The pipeline is written in Java and I am using Streaming Engine to run the Dataflow pipeline.

This is a schema of the pipeline:

  1. Read messages in from Pub/Sub
  2. Transform the PubsubMessages into TableRows
  3. Do some local computations on the data
  4. Write successful records out to BigQuery
  5. Write failed records out to BigQuery (to dead letter tables)

and the first part looks like this:

PCollection<PubsubMessage> messages =
                pipeline.apply(
                        "ReadPubSubSubscription",
                        PubsubIO.readMessagesWithAttributes()
                                .withIdAttribute("Some Attribute Field Name")
                                .fromSubscription(options.getInputSubscription()));

so I guess I am reading PubSub messages with a PubsubMessageWithAttributesCoder that for the payload is using a ByteArrayCoder.

Full stack trace follows:

Error message from worker: java.lang.RuntimeException: java.lang.IllegalStateException: Got poison pill or timeout but stream is not done.
        org.apache.beam.runners.dataflow.worker.MetricTrackingWindmillServerStub.getStateData(MetricTrackingWindmillServerStub.java:233)
        org.apache.beam.runners.dataflow.worker.WindmillStateReader.startBatchAndBlock(WindmillStateReader.java:474)
        org.apache.beam.runners.dataflow.worker.WindmillStateReader$WrappedFuture.get(WindmillStateReader.java:365)
        org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.Futures$2.get(Futures.java:542)
        org.apache.beam.runners.dataflow.worker.WindmillStateInternals$WindmillBag.fetchData(WindmillStateInternals.java:1636)
        org.apache.beam.runners.dataflow.worker.WindmillStateInternals$WindmillBag.access$2400(WindmillStateInternals.java:1583)
        org.apache.beam.runners.dataflow.worker.WindmillStateInternals$WindmillBag$1.read(WindmillStateInternals.java:1683)
        org.apache.beam.runners.dataflow.worker.WindmillStateInternals$WindmillBag$1.read(WindmillStateInternals.java:1674)
        org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.ReduceFnRunner.onTrigger(ReduceFnRunner.java:999)
        org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.ReduceFnRunner.emit(ReduceFnRunner.java:932)
        org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.ReduceFnRunner.onTimers(ReduceFnRunner.java:793)
        org.apache.beam.runners.dataflow.worker.StreamingGroupAlsoByWindowViaWindowSetFn.processElement(StreamingGroupAlsoByWindowViaWindowSetFn.java:97)
        org.apache.beam.runners.dataflow.worker.StreamingGroupAlsoByWindowViaWindowSetFn.processElement(StreamingGroupAlsoByWindowViaWindowSetFn.java:43)
        org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner.invokeProcessElement(GroupAlsoByWindowFnRunner.java:121)
        org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner.processElement(GroupAlsoByWindowFnRunner.java:73)
        org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.LateDataDroppingDoFnRunner.processElement(LateDataDroppingDoFnRunner.java:80)
        org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowsParDoFn.processElement(GroupAlsoByWindowsParDoFn.java:137)
        org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process(ParDoOperation.java:44)
        org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:49)
        org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.runReadLoop(ReadOperation.java:212)
        org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.start(ReadOperation.java:163)
        org.apache.beam.runners.dataflow.worker.util.common.worker.MapTaskExecutor.execute(MapTaskExecutor.java:92)
        org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.process(StreamingDataflowWorker.java:1437)
        org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.access$1100(StreamingDataflowWorker.java:165)
        org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker$7.run(StreamingDataflowWorker.java:1113)
        java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.IllegalStateException: Got poison pill or timeout but stream is not done.
        org.apache.beam.runners.dataflow.worker.windmill.GrpcWindmillServer$AppendableInputStream$1.hasMoreElements(GrpcWindmillServer.java:1538)
        java.base/java.io.SequenceInputStream.peekNextStream(SequenceInputStream.java:101)
        java.base/java.io.SequenceInputStream.nextStream(SequenceInputStream.java:97)
        java.base/java.io.SequenceInputStream.read(SequenceInputStream.java:203)
        org.apache.beam.runners.dataflow.worker.windmill.GrpcWindmillServer$AppendableInputStream.read(GrpcWindmillServer.java:1605)
        org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.CodedInputStream$StreamDecoder.tryRefillBuffer(CodedInputStream.java:2786)
        org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.CodedInputStream$StreamDecoder.isAtEnd(CodedInputStream.java:2709)
        org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.CodedInputStream$StreamDecoder.readTag(CodedInputStream.java:2060)
        org.apache.beam.runners.dataflow.worker.windmill.Windmill$KeyedGetDataResponse.<init>(Windmill.java:37549)
        org.apache.beam.runners.dataflow.worker.windmill.Windmill$KeyedGetDataResponse.<init>(Windmill.java:37505)
        org.apache.beam.runners.dataflow.worker.windmill.Windmill$KeyedGetDataResponse$1.parsePartialFrom(Windmill.java:40068)
        org.apache.beam.runners.dataflow.worker.windmill.Windmill$KeyedGetDataResponse$1.parsePartialFrom(Windmill.java:40062)
        org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.AbstractParser.parsePartialFrom(AbstractParser.java:215)
        org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:232)
        org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:237)
        org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.AbstractParser.parseFrom(AbstractParser.java:48)
        org.apache.beam.vendor.grpc.v1p36p0.com.google.protobuf.GeneratedMessageV3.parseWithIOException(GeneratedMessageV3.java:339)
        org.apache.beam.runners.dataflow.worker.windmill.Windmill$KeyedGetDataResponse.parseFrom(Windmill.java:38173)
        org.apache.beam.runners.dataflow.worker.windmill.GrpcWindmillServer$GrpcGetDataStream.issueRequest(GrpcWindmillServer.java:1164)
        org.apache.beam.runners.dataflow.worker.windmill.GrpcWindmillServer$GrpcGetDataStream.requestKeyedData(GrpcWindmillServer.java:1117)
        org.apache.beam.runners.dataflow.worker.MetricTrackingWindmillServerStub.getStateData(MetricTrackingWindmillServerStub.java:220)
        org.apache.beam.runners.dataflow.worker.WindmillStateReader.startBatchAndBlock(WindmillStateReader.java:474)
        org.apache.beam.runners.dataflow.worker.WindmillStateReader$WrappedFuture.get(WindmillStateReader.java:365)
        org.apache.beam.vendor.guava.v26_0_jre.com.google.common.util.concurrent.Futures$2.get(Futures.java:542)
        org.apache.beam.runners.dataflow.worker.WindmillStateInternals$WindmillBag.fetchData(WindmillStateInternals.java:1636)
        org.apache.beam.runners.dataflow.worker.WindmillStateInternals$WindmillBag.access$2400(WindmillStateInternals.java:1583)
        org.apache.beam.runners.dataflow.worker.WindmillStateInternals$WindmillBag$1.read(WindmillStateInternals.java:1683)
        org.apache.beam.runners.dataflow.worker.WindmillStateInternals$WindmillBag$1.read(WindmillStateInternals.java:1674)
        org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.ReduceFnRunner.onTrigger(ReduceFnRunner.java:999)
        org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.ReduceFnRunner.emit(ReduceFnRunner.java:932)
        org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.ReduceFnRunner.onTimers(ReduceFnRunner.java:793)
        org.apache.beam.runners.dataflow.worker.StreamingGroupAlsoByWindowViaWindowSetFn.processElement(StreamingGroupAlsoByWindowViaWindowSetFn.java:97)
        org.apache.beam.runners.dataflow.worker.StreamingGroupAlsoByWindowViaWindowSetFn.processElement(StreamingGroupAlsoByWindowViaWindowSetFn.java:43)
        org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner.invokeProcessElement(GroupAlsoByWindowFnRunner.java:121)
        org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner.processElement(GroupAlsoByWindowFnRunner.java:73)
        org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.LateDataDroppingDoFnRunner.processElement(LateDataDroppingDoFnRunner.java:80)
        org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowsParDoFn.processElement(GroupAlsoByWindowsParDoFn.java:137)
        org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process(ParDoOperation.java:44)
        org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:49)
        org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.runReadLoop(ReadOperation.java:212)
        org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.start(ReadOperation.java:163)
        org.apache.beam.runners.dataflow.worker.util.common.worker.MapTaskExecutor.execute(MapTaskExecutor.java:92)
        org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.process(StreamingDataflowWorker.java:1437)
        org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.access$1100(StreamingDataflowWorker.java:165)
        org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker$7.run(StreamingDataflowWorker.java:1113)
        java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
        java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
        java.base/java.lang.Thread.run(Thread.java:834)
apache-beam google-cloud-dataflow google-cloud-pubsub java google-bigquery
1 Answers

Your basic idea is right. See How do I deal with 'poison pill' messages when using Google Pub/Sub? for discussion of just the Pubsub part.

Are you running your Dataflow pipeline using Streaming Engine? Do you have an example of the pipeline so we could understand better? Perhaps a stack trace as well?

About the error before converting messages, are you reading from Pub/Sub in what format?

PubsubIO allows you to specify a dead-letter topic when reading, which could cover this use case of malformed messages:

PubsubIO.readProtos(Proto.class)
    .fromSubscription("projects/{project_id}/subscriptions/{input-subscription}") 
    .withDeadLetterTopic("projects/{project_id}/topics/{output=dlq}"))
Bruno Volpato0
Related