Airbyte Connections Sync

Airbyte Connections Sync

Certified

Run an Airbyte connection sync

Starts a sync for an Airbyte connection and, by default, waits for the job to finish. Polling runs every second for up to 60 minutes unless you change pollFrequency or maxDuration. If a sync is already running for the connection, the task adopts it and polls it to completion instead of failing or blindly re-triggering — see onActiveSync.

When the job ends in any state other than SUCCEEDED, the task fails but keeps its outputs (job, attempts, metadata, ...), so errors and finally branches can read them. Set onPartialSuccess: WARNING to end in WARNING instead when the job failed after committing records.

The task also exposes the connection that ran the job, a connectionSnapshot file (connector configurations are never included) and a jobUrl to the job logs in the Airbyte web app. When Kestra runs with assets enabled (Enterprise Edition), each replicated stream that committed records is emitted as a table asset with the source stream as its input. Assets are emitted only when the task ends in SUCCESS or WARNING, because Kestra drops the assets of a failed task run, and their counts are the same per-stream figures as metadata, so a job that ended in WARNING emits no asset

yaml
type: io.kestra.plugin.airbyte.connections.Sync

Trigger an Airbyte sync and wait for completion

yaml
id: airbyte_sync
namespace: company.team

tasks:
  - id: sync
    type: io.kestra.plugin.airbyte.connections.Sync
    url: http://localhost:8080
    connectionId: e3b1ce92-547c-436f-b1e8-23b6936c12cd

  - id: log_rows_synced
    type: io.kestra.plugin.core.log.Log
    message: "Rows synced: {{ outputs.sync.metadata.rowsSynced }}"

Trigger a single Airbyte sync on schedule

yaml
id: airbyte_sync
namespace: company.team

tasks:
  - id: data_ingestion
    type: io.kestra.plugin.airbyte.connections.Sync
    connectionId: e3b1ce92-547c-436f-b1e8-23b6936c12ab
    url: http://host.docker.internal:8000/
    username: "{{ secret('AIRBYTE_USERNAME') }}"
    password: "{{ secret('AIRBYTE_PASSWORD') }}"

triggers:
  - id: every_minute
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/1 * * * *"

Fail immediately instead of adopting an already running sync

yaml
id: airbyte_sync_fail_on_active
namespace: company.team

tasks:
  - id: sync
    type: io.kestra.plugin.airbyte.connections.Sync
    url: http://localhost:8080
    connectionId: e3b1ce92-547c-436f-b1e8-23b6936c12cd
    onActiveSync: FAIL

Handle a failed sync and tolerate partial success

yaml
id: airbyte_sync_with_error_handling
namespace: company.team

tasks:
  - id: sync
    type: io.kestra.plugin.airbyte.connections.Sync
    url: http://localhost:8080
    connectionId: e3b1ce92-547c-436f-b1e8-23b6936c12cd
    onPartialSuccess: WARNING

errors:
  - id: alert
    type: io.kestra.plugin.core.log.Log
    level: ERROR
    message: |
      Airbyte job {{ outputs.sync.jobId }} ended {{ outputs.sync.job.status }}:
      {{ outputs.sync.job.failures[0].externalMessage ?? 'no failure message reported' }}
      Logs: {{ outputs.sync.jobUrl }}
Installation

Our standard Docker images already include this plugin. If you use a -slim image or another setup, install it with the following command: 

kestractl plugins get io.kestra.plugin:plugin-airbyte --compatible-for $KESTRA_VERSION

For Kestra CLI, replace LATEST with a specific version or version range to pin the plugin version. For kestractl, use your Kestra version with --compatible-for, or specify an exact plugin version.

Properties

Connection ID

Airbyte connection ID to sync

Airbyte API URL

Base URL of the Airbyte instance to call. This value is rendered from the current run context

Application credentials

Client credentials used to request an Airbyte application access token from /api/v1/applications/token. Use this instead of a static token when Airbyte application authentication is enabled

Definitions
clientId*string

Client ID

Airbyte application client ID

clientSecret*string

Client secret

Airbyte application client secret. Store this value in a Secret

Assets this task consumes as inputs or produces as outputs, for lineage tracking and the asset graph (Enterprise Edition). A flow declaring this property on a task is rejected in the open-source edition.

Definitions
assetFailureBehaviorstring
Possible Values
IGNOREFAILWARN

Asset failure behavior

Behavior applied to the task state when a declared asset fails to render, emit, or be persisted (e.g. a lock conflict): FAIL escalates it to FAILED, WARN (default) warns it if it would otherwise succeed, IGNORE leaves the state untouched.

enableAutobooleanstring

Whether to auto-register assets referenced dynamically at runtime that are not statically declared in inputs or outputs.

inputsarray

The assets consumed as inputs.

id*string
Min length1
typestring
outputs

The assets produced as outputs.

id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*string
Min length1

Custom asset type

descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
DefaultPT1H

Maximum wait duration

Maximum total time to wait when wait is enabled. Defaults to 60 minutes. When a job is adopted from an already running sync, this duration is counted from the moment it is adopted, not from the job's actual start time

DefaultADOPT
Possible Values
ADOPTFAILSKIP

Behavior when a sync is already running

Controls what happens when a sync is already running for this connection. A job that is pending, queued, running or incomplete counts as an active sync:

  • ADOPT (default): attach to the in-flight job and poll it to completion instead of triggering a new sync.
  • FAIL: fail the task immediately.
  • SKIP: succeed immediately without waiting, with alreadyRunning set to true and no jobId.
DefaultFAIL
Possible Values
FAILWARNING

Behavior on partial success

Controls what happens when the job ends FAILED but at least one attempt committed records (Airbyte's partialSuccess). Only applies when wait is enabled:

  • FAIL (default): fail the task.
  • WARNING: end the task in WARNING and return the outputs. Any other non-successful outcome always fails the task.

HTTP client options

HTTP client configuration applied to Airbyte API requests

Definitions
allowFailedbooleanstring
Defaultfalse

If true, allow a failed response code (response code >= 400)

allowedResponseCodesarray
SubTypeinteger

List of response code allowed for this request

auth

The authentication to use.

type*object
passwordstring

The password for HTTP basic authentication.

usernamestring

The username for HTTP basic authentication.

type*object
tokenstring

The token for bearer token authentication.

type*object
passwordstring

The password for HTTP Digest authentication.

usernamestring

The username for HTTP Digest authentication.

defaultCharsetstring
DefaultUTF-8

The default charset for the request.

enabledTcpExtendedKeepAlivebooleanstring
Defaulttrue

Whether to enable TCP Keep-Alive extended socket options (TCP_KEEPIDLE, TCP_KEEPINTERVAL, TCP_KEEPCOUNT).

Set to false when running on Windows workers, as these extended socket options are not supported by the Windows JDK and will cause connection failures.

followRedirectsbooleanstring
Defaulttrue

Whether redirects should be followed automatically.

logsarray
SubTypestring
Possible Values
REQUEST_HEADERSREQUEST_BODYRESPONSE_HEADERSRESPONSE_BODY

The enabled log.

proxy

The proxy configuration.

addressstring

The address of the proxy server.

passwordstring

The password for proxy authentication.

portintegerstring

The port of the proxy server.

typestring
DefaultDIRECT
Possible Values
DIRECTHTTPSOCKS

The type of proxy to use.

usernamestring

The username for proxy authentication.

retry

Retry strategy for HTTP requests.

retry is disabled by default.

Retry with a fixed delay between attempts.

interval*string
Formatduration
type*object
behaviorstring
DefaultRETRY_FAILED_TASK
Possible Values
RETRY_FAILED_TASKCREATE_NEW_EXECUTION
maxAttemptsinteger
Minimum>= 1
maxDurationstring
Formatduration
warningOnRetryboolean
Defaultfalse

Retry with exponentially increasing delays between attempts.

interval*string
Formatduration
maxInterval*string
Formatduration
type*object
behaviorstring
DefaultRETRY_FAILED_TASK
Possible Values
RETRY_FAILED_TASKCREATE_NEW_EXECUTION
delayFactornumber
maxAttemptsinteger
Minimum>= 1
maxDurationstring
Formatduration
warningOnRetryboolean
Defaultfalse

Retry with a random delay within a configurable range between attempts.

maxInterval*string
Formatduration
minInterval*string
Formatduration
type*object
behaviorstring
DefaultRETRY_FAILED_TASK
Possible Values
RETRY_FAILED_TASKCREATE_NEW_EXECUTION
maxAttemptsinteger
Minimum>= 1
maxDurationstring
Formatduration
warningOnRetryboolean
Defaultfalse
retryOnStatusCodesarray
SubTypeinteger
Default[502,503,504]

HTTP status codes that should be retried.

Used for idempotent methods (GET, HEAD) without their own entry in retryOnStatusCodesByMethod. Non-idempotent methods are never retried on a status code unless listed in retryOnStatusCodesByMethod. Defaults to 502, 503 and 504 for idempotent methods (GET, HEAD) only.

retryOnStatusCodesByMethodobject
SubTypearray

Per-HTTP-method overrides for which status codes should be retried.

Keys are HTTP methods (e.g. GET, POST). A method with no entry here falls back to retryOnStatusCodes. This lets a caller retry more broadly for safe, read-only methods (for example, every 5xx and a 429 for GET/HEAD) while keeping a narrower list for methods that write, where retrying a request that may have already been processed by the origin risks duplicating the effect. Example: {GET: [429, 500, 502, 503, 504], HEAD: [429, 500, 502, 503, 504]}.

retryableTransportFailureMethodsarray
SubTypestring
Default["GET","HEAD"]

HTTP methods for which any transport-level failure may be retried.

A transport-level failure is a connection or timeout error rather than an HTTP response (e.g. a read timeout or a connection reset mid-request). For a method listed here, any such failure is retried. Defaults to GET and HEAD HTTP methods.

ssl

The SSL request options

insecureTrustAllCertificatesbooleanstring

Whether to disable checking of the remote SSL certificate.

Only applies if no trust store is configured. Note: This makes the SSL connection insecure and should only be used for testing. If you are using a self-signed certificate, set up a trust store instead.

timeout

The timeout configuration.

connectTimeoutstring

The time allowed to establish a connection to the server before failing.

readIdleTimeoutstring
DefaultPT5M

The time allowed for a read connection to remain idle before closing it.

Basic auth password

Password for Airbyte basic authentication. Store this value in a Secret

DefaultPT1S

Poll frequency

Interval between status checks while waiting for the job to finish. Defaults to 1 second

Bearer token

Bearer token used for Airbyte API requests. This value is rendered from the current run context

Basic auth username

Username for Airbyte basic authentication. If both basic auth and a bearer token are configured, basic auth is sent last and takes precedence

Defaulttrue

Wait for completion

If true, wait for the Airbyte job to reach a terminal state before the task completes. Defaults to true

Airbyte web app URL

Base URL of the Airbyte web app, used to build the jobUrl output. Defaults to url. Set it when the web app is served from a different address than the API

Adopted

Whether jobId refers to a sync that was already running and got adopted, rather than a sync newly triggered by this task

Already running

Whether a sync was already running for the connection, either adopted (onActiveSync: ADOPT) or skipped (onActiveSync: SKIP)

Formaturi

Attempts file

Internal storage URI of an Ion file with one row per Airbyte attempt. Only available when wait is enabled

Connection

Summary of the connection, source and destination that ran the job. null when their configuration could not be read

Definitions

Short, non-secret description of the connection that ran the job

connectionIdstring

Connection ID

destinationIdstring

Destination ID

destinationNamestring

Destination name

Name given to the destination in Airbyte

destinationTypestring

Destination type

Destination connector type, for example Snowflake

namestring

Name

scheduleTypestring

Schedule type

sourceIdstring

Source ID

sourceNamestring

Source name

Name given to the source in Airbyte

sourceTypestring

Source type

Source connector type, for example Postgres

statusstring

Status

streamCountinteger

Selected stream count

Number of streams replicated by the connection

workspaceIdstring

Workspace ID

Formaturi

Connection snapshot file

Internal storage URI of a JSON file with the connection, source and destination configuration at the time of the job. Connector configurations are never included. null when it could not be read

Job summary

Status, timing, attempt counts and sanitized failure reasons of the Airbyte job. Only available when wait is enabled

Definitions

Final state of the Airbyte job. The job ID is exposed by the jobId output of Sync

attemptCountinteger

Attempt count

configTypestring
Possible Values
CHECK_CONNECTION_SOURCECHECK_CONNECTION_DESTINATIONDISCOVER_SCHEMAGET_SPECSYNCRESET_CONNECTIONREFRESHCLEAR

Config type

Kind of job, for example sync

createdAtstring
Formatdate-time

Created at

durationstring
Formatduration

Duration

Time between the job start and the end of its last attempt. Templates render it as a number of seconds, for example 31.0

endedAtstring
Formatdate-time

Ended at

End time of the last attempt

failedAttemptCountinteger

Failed attempt count

failuresarray

Failures

Sanitized failure reasons of every attempt

Sanitized failure reason of an Airbyte attempt. Internal messages and stack traces are never kept

externalMessagestring

External message

User-facing failure message, truncated to 2 KiB

originstring
Possible Values
SOURCEDESTINATIONREPLICATIONPERSISTENCENORMALIZATIONDBTAIRBYTE_PLATFORMUNKNOWN

Origin

Airbyte component that failed, for example source or destination

retryableboolean

Retryable

Whether Airbyte considers the failure retryable

streamstring

Stream

Stream affected by the failure, formatted as namespace.name or name. Absent when the failure is not tied to a stream

typestring
Possible Values
CONFIG_ERRORSYSTEM_ERRORMANUAL_CANCELLATIONREFRESH_SCHEMAHEARTBEAT_TIMEOUTDESTINATION_TIMEOUTTRANSIENT_ERROR

Type

Failure category reported by Airbyte, for example config_error or system_error

partialSuccessboolean

Partial success

true when at least one attempt committed records before failing

startedAtstring
Formatdate-time

Started at

statusstring
Possible Values
PENDINGQUEUEDRUNNINGINCOMPLETEFAILEDSUCCEEDEDCANCELLED

Status

Terminal job status

Job ID

Airbyte job ID for the sync. Set for every outcome except onActiveSync: SKIP, whether the job was newly triggered or adopted from an already running sync

Job URL

Link to the job logs in the Airbyte web app, built from webappUrl or url. null when the workspace could not be resolved

Sync metadata

Rows synced and source and destination tables affected by the completed Airbyte job. Only available when wait is enabled

Definitions
bytesSyncedinteger

Bytes synced

Total number of bytes committed to the destination. Absent when Airbyte does not report per-stream committed bytes

destinationarray

Destination tables

Destination streams affected by the sync and the number of records committed to each one. Contains one entry per stream returned by Airbyte, so its size follows the connection's stream count

bytesinteger

Bytes processed

Bytes emitted by the source, or committed to the destination

namestring

Table name

namespacestring

Table namespace

recordsRejectedinteger

Records rejected

Records rejected by the destination. Only set on destination tables

rowsinteger

Rows processed

recordsRejectedinteger

Records rejected

Total number of records the destination rejected. Absent when Airbyte does not report rejected records

rowsSyncedinteger

Rows synced

Total number of records committed to the destination

sourcearray

Source tables

Source streams affected by the sync and the number of records emitted from each one. Contains one entry per stream returned by Airbyte, so its size follows the connection's stream count

bytesinteger

Bytes processed

Bytes emitted by the source, or committed to the destination

namestring

Table name

namespacestring

Table namespace

recordsRejectedinteger

Records rejected

Records rejected by the destination. Only set on destination tables

rowsinteger

Rows processed

Unitattempt

Number of attempts made during the Airbyte sync (emitted when wait is enabled)

Unitattempt

Number of failed attempts during the Airbyte job (emitted when wait is enabled)

Unitbyte

Number of bytes emitted during processing (emitted when wait is enabled)

Time between the start of the job and the end of its last attempt (emitted when wait is enabled)

Unitrecord

Number of records successfully committed (emitted when wait is enabled)

Unitrecord

Number of records emitted during processing (emitted when wait is enabled)

Unitrecord

Number of records rejected by the destination, tagged by stream (emitted when wait is enabled)

Unitmessage

Number of state messages emitted (emitted when wait is enabled)

More Plugins in this Category

Firecrawl
Firecrawl

Tasks that scrape and crawl websites into clean, LLM-ready content with Firecrawl.

Data

2 tasks

Liquibase
Liquibase

Tasks that run Liquibase change management commands.

Data

2 tasks

8 blueprints