Oracle incremental sync with Kestra trigger

id: oracle-incremental-sync-trigger
namespace: company.team

variables:
  oracle_url: jdbc:oracle:thin:@//oracle.internal:1521/ORCLPDB1
  batch_size: 500

concurrency:
  limit: 1

labels:
  team: data
  pipeline: orders-incremental-sync

pluginDefaults:
  - type: io.kestra.plugin.jdbc.oracle
    values:
      url: "{{ vars.oracle_url }}"
      username: "{{ secret('ORACLE_USERNAME') }}"
      password: "{{ secret('ORACLE_PASSWORD') }}"

tasks:
  - id: stage_batch
    type: io.kestra.plugin.jdbc.oracle.Batch
    description: Bulk load the batch the trigger captured into a transient
      per-execution staging table.
    from: "{{ trigger.uri }}"
    chunk: "{{ vars.batch_size }}"
    retry:
      type: constant
      interval: PT10S
      maxAttempt: 3
    columns:
      - order_id
      - customer_id
      - customer_name
      - region
      - amount
      - created_at
    sql: "INSERT INTO orders_enriched_stg (order_id, customer_id, customer_name,
      region, amount, created_at) VALUES (?, ?, ?, ?, ?, ?)"

  - id: merge_into_mart
    type: io.kestra.plugin.jdbc.oracle.Queries
    description: Upsert the staged rows into the mart keyed on order_id, then clear
      the staging table in one transaction. The MERGE makes a replay idempotent
      even on the at-least-once path.
    fetchType: NONE
    transaction: true
    retry:
      type: constant
      interval: PT10S
      maxAttempt: 3
    sql: |
      MERGE INTO orders_enriched t
      USING orders_enriched_stg s
        ON (t.order_id = s.order_id)
      WHEN MATCHED THEN UPDATE SET
        t.customer_id   = s.customer_id,
        t.customer_name = s.customer_name,
        t.region        = s.region,
        t.amount        = s.amount,
        t.created_at    = s.created_at
      WHEN NOT MATCHED THEN INSERT
        (order_id, customer_id, customer_name, region, amount, created_at)
      VALUES
        (s.order_id, s.customer_id, s.customer_name, s.region, s.amount, s.created_at);
      DELETE FROM orders_enriched_stg;

  - id: notify_batch
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Post the size of the synced batch to Slack so the data team can
      track sync volume.
    url: "{{ secret('SLACK_WEBHOOK') }}"
    retry:
      type: constant
      interval: PT10S
      maxAttempt: 3
    messageText: "Oracle incremental sync upserted {{ trigger.size }} enriched
      order(s) into orders_enriched. Source batch: {{ trigger.uri }}"

triggers:
  - id: poll_new_orders
    type: io.kestra.plugin.jdbc.oracle.Trigger
    description: Poll Oracle every 5 minutes for unclaimed orders, enrich them, and
      claim exactly that id window.
    interval: PT5M
    url: "{{ vars.oracle_url }}"
    username: "{{ secret('ORACLE_USERNAME') }}"
    password: "{{ secret('ORACLE_PASSWORD') }}"
    fetchType: STORE
    sql: |
      SELECT
        o.order_id,
        o.customer_id,
        c.customer_name,
        c.region,
        o.amount,
        o.created_at
      FROM orders o
      LEFT JOIN customers c ON c.customer_id = o.customer_id
      WHERE o.processed_flag IS NULL
      ORDER BY o.order_id
      FETCH FIRST {{ vars.batch_size }} ROWS ONLY
    afterSQL: |
      UPDATE orders
      SET processed_flag = 'Y'
      WHERE processed_flag IS NULL
        AND order_id <= (
          SELECT MAX(order_id) FROM (
            SELECT order_id
            FROM orders
            WHERE processed_flag IS NULL
            ORDER BY order_id
            FETCH FIRST {{ vars.batch_size }} ROWS ONLY
          )
        )

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack when a sync execution fails so the operator knows which
      already-claimed batch to replay.
    url: "{{ secret('SLACK_WEBHOOK') }}"
    messageText: ":rotating_light: {{ flow.namespace }}.{{ flow.id }} failed
      (execution {{ execution.id }}). The source rows for this batch are already
      claimed; replay batch {{ trigger.uri }} from this execution."

Run a change-data-capture style incremental sync from an Oracle Database with Kestra. A polling trigger captures a bounded batch of new orders, enriches them with customer attributes, claims exactly that id window race-safely, and upserts the result into a reporting mart with a MERGE so a replay never duplicates. This is the idempotent, at-least-once pattern teams reach for when an Oracle scheduler job and a plain INSERT keep producing duplicate rows or skipping records under concurrency.

How it works

The poll_new_orders trigger (io.kestra.plugin.jdbc.oracle.Trigger) polls every PT5M for orders rows where processed_flag IS NULL, ordered by order_id, fetching only the first batch_size rows (default 500). Its sql LEFT JOINs customers so each captured row already carries customer_name and region, and fetchType: STORE writes the batch to internal storage as trigger.uri.
The trigger afterSQL claims that exact window, marking only rows whose order_id is at or below the highest id in the batch. Because order_id is monotonically increasing, rows inserted concurrently get a higher id and are simply picked up next poll, avoiding the race a blanket UPDATE ... WHERE processed_flag IS NULL would create.
The stage_batch task (io.kestra.plugin.jdbc.oracle.Batch) bulk loads the enriched batch from trigger.uri into the transient orders_enriched_stg table in chunks of batch_size.
The merge_into_mart task (io.kestra.plugin.jdbc.oracle.Queries) runs a MERGE into orders_enriched keyed on order_id and clears the staging table in one transaction (transaction: true). The MERGE makes replays idempotent.
The notify_batch task (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) posts the synced row count to Slack.

What you get

Idempotent mart loads: the MERGE on order_id absorbs replays instead of inserting duplicates.
Race-safe claiming: only the exact emitted id window is marked processed.
Built-in resilience: stage_batch, merge_into_mart, and notify_batch each retry (constant, 3 attempts).
Operator visibility: a per-batch Slack summary plus an errors alert naming the execution id and trigger.uri to replay.
Back-pressure, not corruption: concurrency.limit is 1, so a slow mart load just delays the next poll.

Who it's for

Data engineers running incremental syncs out of Oracle into a reporting mart.
Analytics teams who need exactly-once-at-the-mart semantics without a full CDC stack.
Platform teams replacing brittle Oracle scheduler jobs with observable, replayable pipelines.

Why orchestrate this with Kestra

Oracle's own scheduler (DBMS_SCHEDULER) can run SQL on a clock, but it cannot store the captured batch as an artifact, retry individual steps, alert with the exact batch to replay, or give you execution-level lineage when a load half-fails. Kestra adds an event-driven polling trigger, per-task retries, a flow-level error handler, and declarative YAML you version in Git, turning a fragile "claim then load" job into a replayable, auditable pipeline.

Prerequisites

A reachable Oracle Database instance. The JDBC URL uses the thin driver service form (jdbc:oracle:thin:@//host:1521/service), set in the oracle_url variable.
An orders table with a numeric, sequence-generated order_id and a processed_flag column, a customers table to join against, an orders_enriched mart with a unique order_id, and an orders_enriched_stg staging table of the same shape.
Least privilege: SELECT on orders and customers, UPDATE on processed_flag, and INSERT/UPDATE/DELETE on the mart and staging tables.

Secrets

ORACLE_USERNAME: the Oracle database user used by the trigger and the mart load.
ORACLE_PASSWORD: the password for that Oracle user.
SLACK_WEBHOOK: the Slack Incoming Webhook URL the batch summary and failure alert post to.

Quick start

Create the ORACLE_USERNAME, ORACLE_PASSWORD, and SLACK_WEBHOOK secrets.
Update the oracle_url variable to point at your instance and service name.
Create orders_enriched_stg matching orders_enriched, with a unique constraint on orders_enriched.order_id.
Adjust the trigger sql, the batch_size variable, the staging INSERT, and the MERGE to match your schema.
Enable the flow; the trigger starts polling automatically.

How to extend

Catch up on a backlog by raising batch_size or lowering the trigger interval (default drains about 6,000 rows per hour).
Swap the watermark: use a last_modified timestamp instead of processed_flag for update-aware syncs.
Fan out to more destinations by adding tasks after the MERGE (for example a downstream warehouse load).
On Oracle RAC or CACHE/NOORDER sequences, switch to ORDER NOCACHE to keep the monotonic-id guarantee.

Incremental Oracle order sync with a polling trigger

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Incremental Oracle order sync with a polling trigger

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Build an ETL Pipeline with Airbyte Cloud and dbt Cloud for SaaS Analytics

Generate and Execute SQL Queries from Natural Language Using OpenAI GPT-4o

Run a Scheduled ETL Sync with Airbyte Cloud

Build an ETL Pipeline with Airbyte Cloud and dbt Core for SaaS Analytics

AI Document Data Extraction: Parse PDFs and Web Pages into Structured JSON with OpenAI

AI-Powered Inventory Demand Forecasting, Auto-Generate Purchase Orders with OpenAI

New to Kestra?