gcp PubSub getting "processing delay was high"

Question

This is our set up, we have one topic and this topic has two subscriptions v1 and v2 with exact identical settings, both are pull subscription with 10 sec ack deadline.
both subscription v1 and v2 goes to separate dedicated dataflow where v2's data flow is more optimized but pretty much doing same thing.

Problem is that every now and then we see below warning messages and backlog starts build up in v2 subscription only and v1 shows little to no backlog.

08:53:56.000 ${MESSAGE_ID} Pubsub processing delay was high at 72 sec.

Dataflow log in v2 shows nothing obvious except above messages. In fact v2 dataflow cpu usage is lower than v1 so I cant make any sense out of this.

Questions:

What causes processing delay and how can I fix it?
Why isn't v1 subscription getting same warnings?

Updated at 2017/01/17

As suggested via @ben it seems that ParDo filtering operation we do right after PubSub read is hitting unexpectedly high latency. But considering getClassroomIds is a simple java list I'm not sure how I can tackle this problem. One question is that is coder we have applied to pubsub lazy? Is unzipping and deserializing we have defined in coder applied when ProcessContext#element() is called?

def processElement(c: DoFn[Entity, Entity]#ProcessContext) = {
  val startTime = System.currentTimeMillis()
  val entity = c.element()
  if (!entity.getClassroomIds.isEmpty) {
    c.output(entity)
  }

  val latencyMs = System.currentTimeMillis() - startTime
  if (latencyMs > 1000) {
    // We see this warning messages during the load spike
    log.warn(s"latency breached 1 second threshold: $latencyMs ms")
  }
}

Job IDs would be useful for to look at what was happening with these runs. Typically, that message means the ParDo steps immediately following the PubSub source are slow. Also looking at the "shuffler" logs may indicate if flow-control is introducing queueing delays. — Ben Chambers
@BenChambers ParDo right after should be very simple filtering logic so I dont' know if that is the case. However stream is an encoded zip that needs to be deserialized so it maybe slow there. But this part should be same between v1 and v2. And if you mean dataflow works logs by "shuffler" log, than I have looked at them and other than occasional GC logs and "process delay" log above I don't see anything around the event time line. — jk-kim
v1 stream is dataflow v1.6.1 and v2 is dataflow v1.9.0 but I don't see any obvious thing that might cause v1.6.1 to to perform better — jk-kim
We'll likely need Job IDs to get much further. Note that it isn't just the next ParDo -- it would be all of the ParDos after the source until the first group by key-like operation. — Ben Chambers
Our job id: 2017-01-12_17_40_22-2811560382049765631 and question is updated — jk-kim

Ben Chambers Ben Chambers · Accepted Answer · 2017-01-17T19:42:17

The timing you mention isn't accurately accounting for the time spent in that step. In particular, due to the fusion optimization it reflects all of the ParDos after the filtering operation.

If you have a pipeline which looks like:

ReadFromPubSub -> ParDo(Filter) -> ParDo(Expensive) -> ParDo(Write)

Both Expensive and Write execute on each element coming out of the Filter before the call to c.output returns. This is further a problem because it is fused into the elements coming out of PubSub.

The easiest solution is probably to perform a Reshuffle:

pipeline
  .apply(PubSubIO.read(...))
  .apply(keyEachElement by their value and Void)
  .apply(new Reshuffle<E, Void>())
  .apply(move the key back to the element)
  // rest of the pipeline

Note that using the Reshuffle instead of a regular GroupByKey has nice properties since it triggers faster than any normally trigger will fire.

gcp PubSub getting "processing delay was high"

1 Answers