flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Satish Chandra Gupta <scgupt...@gmail.com>
Subject How to debug why Flink makes and executes only partial plan
Date Sat, 15 Oct 2016 03:25:52 GMT
Hi,

In my Flink program, after a couple of map, union and connect, I have a
final filter and a sink. Something like this (after abstracting out
details):

val filteredEvents: DataStream[NotificationEvent]
  = allThisStuffWorking
      .name("filtered_users")

filteredEvents
  *.filter(x => check(x.f1, x.f2, someStuff)) //BUG*
  .addSink(new NotificationSinkFunction(notifier))
  .name("send_notification")

The check function returns a Boolean and does not access anything other
than parameters passed. Here is relevant part of Notification Sink Function:

class NotificationSinkFunction(notifier: Notifier)
      extends SinkFunction[NotificationEvent] {

  val LOG: Logger = LoggerFactory.getLogger(this.getClass)

  def invoke(event: NotificationEvent): Unit = {
    LOG.info("Log this notification detail")
    *notifier.send(**event.f1, event.f2) //BUG*
  }
}

If I comment out the lines highlighted and marked with //BUG, the Flink
pipeline works and print the log messages, and Flink shows this execution
plan at the end:

filtered_users -> Sink: send_notification

[image: Inline image 1]


But with either of those two lines marked as BUG above, Flink makes and
executes plan only till filtered_user and does not print the log message.

[image: Inline image 2]

How can I figure out what is wrong with the check function or notifier send
function that prevents Flink from making the full plan. What are the
typical mistakes leading to this?

Thanks,
+satish

Mime
View raw message