Stopping a flow after processing all available data

This solution describes how to design a flow that stops automatically after it finishes processing all available data.

Let's say that your dataflow topology updates a database table daily at 4 am. Rather than have the flow process the data in a few minutes and sit idle for the rest of the day, you want to kick off the flow, have it process all data and then stop - just like old school batch processing.

To do this, simply configure the source to generate events and then route the no-more-data event record to the Pipeline Finisher executor.

The following sources generate no-more-data events:
  • Amazon S3 source
  • Azure Blob Storage source
  • Azure Data Lake Storage Gen2 source
  • Azure Data Lake Storage Gen2 (Legacy) source
  • Directory source
  • Google Cloud Storage source
  • JDBC Multitable Consumer source
  • JDBC Query Consumer source
  • MongoDB Atlas source
  • Oracle Multitable Consumer source
  • Salesforce source
  • Salesforce Bulk API 2.0 source
  • SAP HANA Query Consumer source
  • SFTP/FTP/FTPS Client source
  • SQL Server CDC Client source
  • SQL Server Change Tracking source
  • Web Client source

We'll use the JDBC Query Consumer to show a more complex scenario.

Here's the basic flow that reads from a database, performs some processing, and writes to HDFS:

Flow canvas

To configure the flow to stop after processing all available queried data:
  1. Configure the source to generate events:

    On the General tab of the JDBC Query Consumer source, select the Produce Events property.

    The event output stream becomes available:

    Flow canvas with a JDBC Query Consumer source that is configured to produce events

    The JDBC Query Consumer generates several types of events: query success, query failure, and no-more-data. We know this because you checked the Event Record section of the JDBC Query Consumer documentation. Every event-generating stage has event details in a similar section.

    The query success and failure events can be useful, so you might use a Stream Selector to route those records to a separate event stream. But let's say we don't care about those events, we just want the no-more-data event to pass to the Pipeline Finisher executor.

  2. Connect the event output stream to the Pipeline Finisher executor.

    At this point, all events that the source generates come to the executor. Since the JDBC Query Consumer source generates multiple event types, this setup might cause the executor to stop the flow too soon.

  3. To ensure that only the no-more-data event enters the executor, configure a precondition.

    With a precondition, only records that meet the specified condition can enter the stage.

    We know that each event record includes the event type in the sdc.event.type record header attribute. So to ensure that only no-more-data events enter the stage, we can use the following expression in the precondition:

    ${record:eventType() == 'no-more-data'}
  4. Records that don't meet the precondition go to the stage for error handling, so to avoid storing error records that we don't care about – that is, the query success and failure events – let's also set the On Record Error property to Discard.

    So here's the Pipeline Finisher:

    Flow canvas with a JDBC Query Consumer source that passes event records to a Pipeline Finisher executor

That's it!

With this setup, the JDBC Query Consumer passes a no-more-data event when it completes processing all data returned by the query, and the Pipeline Finisher executor stops the flow and transitions the flow to a Finished state. All other events generated by the source are discarded.