Source Acquisition Capability

The Source Acquisition capability abstracts data ingestion across all source types and acquisition modes.

Contract

public interface SourceAcquisitionContract extends BaseProviderContract {
    ConnectionResult connect();
    SchemaDescriptor discoverSchema();
    DataStream acquire(String mode, Map<String, Object> parameters);
    CheckpointState checkpoint();
    DataStream resume(CheckpointState checkpoint);
    void disconnect();
}

Acquisition Modes

Mode	Description	Use Case
snapshot	Full extract of all data	Initial loads, small tables
incremental	Delta based on watermark/change tracking	Regular updates
time_window	Bounded time range	Backfills, reprocessing
streaming	Continuous consumption	Real-time pipelines
event_notification	Trigger-based acquisition	Event-driven architectures
change_feed	CDC log consumption	Database replication

Pipeline Configuration

- name: acquire-data
  type: source
  capability: source-acquisition
  timeout: PT10M
  config:
    mode: incremental
    watermark: updated_at
    batchSize: 10000

Checkpointing

The capability supports resumable acquisition:

1. checkpoint() — Persists current position (offsets, watermarks, cursors)
2. resume(checkpoint) — Continues from last saved position

For replayable sources (S3, databases, Kafka with retention): resume guarantees no data loss or duplication.

For non-replayable sources (transient streams, push APIs): the Intake service buffers data into a replayable intermediate store.

Connection Timeout

Default: 30 seconds. Maximum: 300 seconds.

If connection fails within the timeout, the Intake service reports a connection failure with source identifier, provider type, and failure reason.

Example Providers

Provider	Sources
JDBC Provider	PostgreSQL, MySQL, Oracle, SQL Server
Kafka Provider	Kafka topics (streaming, change_feed)
S3 Provider	S3 objects (snapshot, incremental)
REST API Provider	HTTP endpoints (event_notification)
SFTP Provider	File-based sources (snapshot, time_window)