Create sink consumer

curl -X POST "https://api.sequinstream.com/api/sinks" \
  -H "Authorization: Bearer YOUR_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "kafka-ids",
    "status": "active",
    "source": {
      "include_schemas": null,
      "exclude_schemas": null,
      "include_tables": null,
      "exclude_tables": null
    },
    "actions": ["insert", "update", "delete"],
    "destination": {
      "type": "kafka",
      "hosts": "localhost:9092",
      "tls": false,
      "topic": "records"
    },
    "database": "dune",
    "filter": "",
    "transform": "id-transform",
    "routing": "",
    "message_grouping": true,
    "batch_size": 100,
    "max_retry_count": null,
    "load_shedding_policy": "pause_on_full",
    "timestamp_format": "iso8601",
    "annotations": {}
  }'

{
  "id": "4ed2a8e5-47a7-4b51-9270-d2f4fdcb94fb",
  "name": "my-kafka-sink",
  "status": "active",
  "database": "my-database",
  "source": {
    "include_schemas": null,
    "exclude_schemas": null,
    "include_tables": null,
    "exclude_tables": null
  },
  "tables": [
    {
      "name": "public.products",
      "group_column_names": ["category"]
    }
  ],
  "actions": ["insert", "update", "delete"],
  "destination": {
    "type": "kafka",
    "hosts": "localhost:9092",
    "tls": false,
    "topic": "records",
    "username": "kafka_user",
    "password": "kafka_password",
    "sasl_mechanism": "PLAIN"
  },
  "filter": "none",
  "transform": "id-transform",
  "routing": "none",
  "message_grouping": true,
  "max_retry_count": null,
  "annotations": {},
  "active_backfills": [],
  "batch_size": 50,
  "load_shedding_policy": "pause_on_full",
  "timestamp_format": "iso8601",
  "health": {
    "name": "Consumer health",
    "status": "healthy",
    "checks": [
      {
        "name": "Sink configuration",
        "status": "healthy"
      },
      ...
    ]
  }
}

POST

sinks

curl -X POST "https://api.sequinstream.com/api/sinks" \
  -H "Authorization: Bearer YOUR_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "kafka-ids",
    "status": "active",
    "source": {
      "include_schemas": null,
      "exclude_schemas": null,
      "include_tables": null,
      "exclude_tables": null
    },
    "actions": ["insert", "update", "delete"],
    "destination": {
      "type": "kafka",
      "hosts": "localhost:9092",
      "tls": false,
      "topic": "records"
    },
    "database": "dune",
    "filter": "",
    "transform": "id-transform",
    "routing": "",
    "message_grouping": true,
    "batch_size": 100,
    "max_retry_count": null,
    "load_shedding_policy": "pause_on_full",
    "timestamp_format": "iso8601",
    "annotations": {}
  }'

{
  "id": "4ed2a8e5-47a7-4b51-9270-d2f4fdcb94fb",
  "name": "my-kafka-sink",
  "status": "active",
  "database": "my-database",
  "source": {
    "include_schemas": null,
    "exclude_schemas": null,
    "include_tables": null,
    "exclude_tables": null
  },
  "tables": [
    {
      "name": "public.products",
      "group_column_names": ["category"]
    }
  ],
  "actions": ["insert", "update", "delete"],
  "destination": {
    "type": "kafka",
    "hosts": "localhost:9092",
    "tls": false,
    "topic": "records",
    "username": "kafka_user",
    "password": "kafka_password",
    "sasl_mechanism": "PLAIN"
  },
  "filter": "none",
  "transform": "id-transform",
  "routing": "none",
  "message_grouping": true,
  "max_retry_count": null,
  "annotations": {},
  "active_backfills": [],
  "batch_size": 50,
  "load_shedding_policy": "pause_on_full",
  "timestamp_format": "iso8601",
  "health": {
    "name": "Consumer health",
    "status": "healthy",
    "checks": [
      {
        "name": "Sink configuration",
        "status": "healthy"
      },
      ...
    ]
  }
}

Creates a new sink consumer.

Request fields

name

string

required

The name of the sink consumer

status

string

required

The initial status of the sink consumer (active, disabled, paused)

source

object

required

The source configuration for the sink consumer

Show source properties

include_schemas

array

List of schemas to include, or null to include all

exclude_schemas

array

List of schemas to exclude, or null to exclude none

include_tables

array

List of tables to include, or null to include allA table name can be specified with schema (e.g. my-schema.my-table) or without (e.g. my-table) which defaults to schema public.

exclude_tables

array

List of tables to exclude, or null to exclude noneA table name can be specified with schema (e.g. my-schema.my-table) or without (e.g. my-table) which defaults to schema public.

tables

array

Additional configuration for individual tables.To configure which tables are consumed, see the source field.

Show tables properties

name

string

required

The name of the table.A table name can be specified with schema (e.g. my-schema.my-table) or without (e.g. my-table) which defaults to schema public.

group_column_names

array

The column names to group by or null to disable grouping

actions

array

required

The database actions to include in the sink (insert, update, delete)

destination

object

required

The destination configuration for the sink consumer. The shape varies by destination type.

Show Kafka

type

string

required

Must be “kafka”

hosts

string

required

Comma-separated list of Kafka hosts

tls

boolean

Whether to use TLS

topic

string

required

Kafka topic name

username

string

Optional username for authentication

password

string

Optional password for authentication

sasl_mechanism

string

Optional SASL mechanism (PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, AWS_MSK_IAM)

aws_region

string

AWS region (required for AWS_MSK_IAM)

aws_access_key_id

string

AWS access key ID (required for AWS_MSK_IAM)

aws_secret_access_key

string

AWS secret access key (required for AWS_MSK_IAM)

Show SQS

type

string

required

Must be “sqs”

queue_url

string

required

SQS queue URL

region

string

required

AWS region

access_key_id

string

required

AWS access key ID

secret_access_key

string

required

AWS secret access key

is_fifo

boolean

Whether the queue is a FIFO queue

Show Kinesis

type

string

required

Must be “kinesis”

stream_arn

string

required

Kinesis stream ARN

access_key_id

string

required

AWS access key ID

secret_access_key

string

required

AWS secret access key

Show RabbitMQ

type

string

required

Must be “rabbitmq”

host

string

required

RabbitMQ host

port

integer

required

RabbitMQ port

username

string

required

Username for authentication

password

string

required

Password for authentication

virtual_host

string

required

Virtual host name (default: ”/”)

tls

boolean

Whether to use TLS

exchange

string

required

Exchange name

Show Redis Stream

type

string

required

Must be “redis_stream”

host

string

required

Redis host

port

integer

required

Redis port

stream_key

string

required

Redis stream key

database

integer

Redis database number (default: 0)

tls

boolean

Whether to use TLS

username

string

Optional username for authentication

password

string

Optional password for authentication

Show Redis String

type

string

required

Must be “redis_string”

host

string

required

Redis host

port

integer

required

Redis port

database

integer

Redis database number (default: 0)

tls

boolean

Whether to use TLS

username

string

Optional username for authentication

password

string

Optional password for authentication

expire_ms

integer

Optional TTL in milliseconds for the keys

mode

string

Mode of operation (static or dynamic, default: static)

Show Azure Event Hub

type

string

required

Must be “azure_event_hub”

namespace

string

required

Event Hub namespace

event_hub_name

string

required

Event Hub name

shared_access_key_name

string

required

Shared access key name

shared_access_key

string

required

Shared access key

Show NATS

type

string

required

Must be “nats”

host

string

required

NATS host

port

integer

required

NATS port

username

string

Optional username for authentication

password

string

Optional password for authentication

jwt

string

Optional JWT for authentication

nkey_seed

string

Optional NKey seed for authentication

tls

boolean

Whether to use TLS

Show GCP PubSub

type

string

required

Must be “gcp_pubsub”

project_id

string

required

GCP project ID

topic_id

string

required

PubSub topic ID

credentials

string

required

Base64-encoded credentials

use_emulator

boolean

Whether to use the emulator

emulator_base_url

string

Emulator base URL if using emulator

Show Elasticsearch

type

string

required

Must be “elasticsearch”

endpoint_url

string

required

Elasticsearch endpoint URL

index_name

string

required

Elasticsearch index name

auth_type

string

required

Authentication type (api_key, basic, bearer)

auth_value

string

required

Authentication value, either an API key or a base64-encoded string of a username and password

batch_size

integer

Number of documents to batch together (default: 100)

Show Meilisearch

type

string

required

Must be “meilisearch”

endpoint_url

string

required

Meilisearch endpoint URL

index_name

string

required

Meilisearch index name

primary_key

string

Primary key field name (default: “id”)

api_key

string

required

Meilisearch API key

batch_size

integer

Number of documents to batch together (default: 100)

timeout_seconds

integer

Request timeout in seconds (default: 5)

Show S2

type

string

required

Must be “s2”

basin

string

required

The name of the S2 basin

stream

string

required

The name of the S2 stream

access_token

string

required

The access token for the S2 account associated with the basin and stream

Show Sequin Stream

type

string

required

Must be “sequin_stream”

Show SNS

type

string

required

Must be “sns”

topic_arn

string

required

SNS topic ARN

region

string

required

AWS region

access_key_id

string

required

AWS access key ID

secret_access_key

string

required

AWS secret access key

is_fifo

boolean

Whether the topic is a FIFO topic

Show Typesense

type

string

required

Must be “typesense”

endpoint_url

string

required

Typesense endpoint URL

collection_name

string

required

Typesense collection name

api_key

string

required

Typesense API key

batch_size

integer

Number of documents to batch together (default: 100)

timeout_seconds

integer

Request timeout in seconds (default: 5)

Show Webhook

type

string

required

Must be “webhook”

http_endpoint

string

required

HTTP endpoint name

http_endpoint_path

string

required

HTTP endpoint path

batch

boolean

Whether the webhook body should be wrapped in a list to support batching. Defaults to true.

database

string

required

The source database for the sink consumer

enrichment

string

The enrichment function for the sink consumer

filter

string

The filter function for the sink consumer

transform

string

The transform function for the sink consumer

routing

string

The routing function for the sink consumer

message_grouping

boolean

Whether message grouping is enabled for ordering purposes. When true (default), messages are grouped by primary key. When false, grouping is disabled for higher throughput batching.

batch_size

integer

Number of records to batch together (1-1000)

max_retry_count

integer

The maximum number of times a message will be retried if delivery fails. Once this limit is reached, the message will be discarded. Defaults to null, meaning messages are retried indefinitely.

load_shedding_policy

string

Determines how Sequin handles overload when sink consumers can’t keep up with incoming messages. Options are:

pause_on_full (default) — pauses replication until the buffer has room again
discard_on_full — drops messages for overloaded consumers to avoid pausing replication

For more details, see load shedding policy.

timestamp_format

string

The format of the timestamp in the source data. Possible values include iso8601 and unix_microsecond.

annotations

object

Additional metadata you can attach to this sink consumer. Annotations can be any JSON object.

Response fields

string

The unique identifier of the sink consumer

name

string

The name of the sink consumer

status

string

The current status of the sink consumer (active, disabled, paused)

database

string

The source database for the sink consumer

source

object

The source configuration for the sink consumer. This determines which tables are consumed.

Show source properties

include_schemas

array

List of schemas to include, or null to include all

exclude_schemas

array

List of schemas to exclude, or null to exclude none

include_tables

array

List of tables to include, or null to include allA table name can be specified with schema (e.g. my-schema.my-table) or without (e.g. my-table) which defaults to schema public.

exclude_tables

array

List of tables to exclude, or null to exclude noneA table name can be specified with schema (e.g. my-schema.my-table) or without (e.g. my-table) which defaults to schema public.

tables

array

Additional configuration for individual tables.To configure which tables are consumed, see the source field.

Show tables properties

name

string

The name of the table.A table name can be specified with schema (e.g. my-schema.my-table) or without (e.g. my-table) which defaults to schema public.

group_column_names

array

The column names to group by or null to disable grouping

actions

array

The database actions to include in the sink (insert, update, delete)

destination

object

The destination configuration for the sink consumer. The shape varies by destination type.

Show destination properties

Show Azure Event Hub properties

type

string

Must be “azure_event_hub”

namespace

string

Event Hub namespace

event_hub_name

string

Event Hub name

shared_access_key_name

string

Shared access key name

shared_access_key

string

Shared access key

Show Elasticsearch properties

type

string

Must be “elasticsearch”

endpoint_url

string

Elasticsearch endpoint URL

index_name

string

Elasticsearch index name

auth_type

string

Authentication type (api_key, basic, bearer)

auth_value

string

Authentication value

batch_size

integer

Number of documents to batch together (default: 100)

Show GCP PubSub properties

type

string

Must be “gcp_pubsub”

project_id

string

GCP project ID

topic_id

string

PubSub topic ID

credentials

string

Base64-encoded credentials

use_emulator

boolean

Whether to use the emulator

emulator_base_url

string

Emulator base URL if using emulator

Show Webhook properties

type

string

Must be “webhook”

http_endpoint

string

HTTP endpoint name

http_endpoint_path

string

HTTP endpoint path

batch

boolean

Whether the webhook body should be wrapped in a list to support batching.

Show Kafka properties

type

string

Must be “kafka”

hosts

string

Comma-separated list of Kafka hosts

tls

boolean

Whether to use TLS

topic

string

Kafka topic name

username

string

Optional username for authentication

password

string

Optional password for authentication

sasl_mechanism

string

Optional SASL mechanism (plain, scram_sha_256, scram_sha_512, aws_msk_iam)

aws_region

string

AWS region (required for aws_msk_iam)

aws_access_key_id

string

AWS access key ID (required for aws_msk_iam)

aws_secret_access_key

string

AWS secret access key (required for aws_msk_iam)

Show Kinesis properties

type

string

Must be “kinesis”

stream_arn

string

Kinesis stream ARN

access_key_id

string

AWS access key ID

secret_access_key

string

AWS secret access key

Show Meilisearch properties

type

string

Must be “meilisearch”

endpoint_url

string

Meilisearch endpoint URL

index_name

string

Meilisearch index name

primary_key

string

Primary key field name (default: “id”)

api_key

string

Meilisearch API key

batch_size

integer

Number of documents to batch together (default: 100)

timeout_seconds

integer

Request timeout in seconds (default: 5)

Show NATS properties

type

string

Must be “nats”

host

string

NATS host

port

integer

NATS port

username

string

Optional username for authentication

password

string

Optional password for authentication

jwt

string

Optional JWT for authentication

nkey_seed

string

Optional NKey seed for authentication

tls

boolean

Whether to use TLS

Show RabbitMQ properties

type

string

Must be “rabbitmq”

host

string

RabbitMQ host

port

integer

RabbitMQ port

username

string

Username for authentication

password

string

Password for authentication

virtual_host

string

Virtual host name (default: ”/”)

tls

boolean

Whether to use TLS

exchange

string

Exchange name

Show Redis Stream properties

type

string

Must be “redis_stream”

host

string

Redis host

port

integer

Redis port

stream_key

string

Redis stream key

database

integer

Redis database number (default: 0)

tls

boolean

Whether to use TLS

username

string

Optional username for authentication

password

string

Optional password for authentication

Show Redis String properties

type

string

Must be “redis_string”

host

string

Redis host

port

integer

Redis port

database

integer

Redis database number (default: 0)

tls

boolean

Whether to use TLS

username

string

Optional username for authentication

password

string

Optional password for authentication

expire_ms

integer

Optional TTL in milliseconds for the keys

mode

string

Mode of operation (static or dynamic, default: static)

Show S2 properties

type

string

Must be “s2”

basin

string

The name of the S2 basin

stream

string

The name of the S2 stream

access_token

string

The access token for the S2 account associated with the basin and stream

Show Sequin Stream properties

type

string

Must be “sequin_stream”

Show SNS properties

type

string

Must be “sns”

topic_arn

string

SNS topic ARN

region

string

AWS region

access_key_id

string

AWS access key ID

secret_access_key

string

AWS secret access key

is_fifo

boolean

Whether the topic is a FIFO topic

Show SQS properties

type

string

Must be “sqs”

queue_url

string

SQS queue URL

region

string

AWS region

access_key_id

string

AWS access key ID

secret_access_key

string

AWS secret access key

is_fifo

boolean

Whether the queue is a FIFO queue

Show Typesense properties

type

string

Must be “typesense”

endpoint_url

string

Typesense endpoint URL

collection_name

string

Typesense collection name

api_key

string

Typesense API key

batch_size

integer

Number of documents to batch together (default: 100)

timeout_seconds

integer

Request timeout in seconds (default: 5)

filter

string

The filter function for the sink consumer

transform

string

The transform function for the sink consumer

routing

string

The routing function for the sink consumer

enrichment

string

The enrichment function for the sink consumer

message_grouping

boolean

Whether message grouping is enabled for delivery orderingSee message grouping and ordering for more details.

max_retry_count

integer

The maximum number of times a message will be retried if delivery fails

annotations

object

User-defined annotations for the sink consumer

active_backfills

array

Array of active backfill IDs

batch_size

integer

Number of records to batch together (1-1000)

load_shedding_policy

string

Determines how Sequin handles overload when sink consumers can’t keep up with incoming messages

timestamp_format

string

The format of the timestamp in the source data

health

object

Health status information for the sink consumer

Show health properties

name

string

Name of the health check

status

string

Overall health status

checks

array

Array of individual health checks

curl -X POST "https://api.sequinstream.com/api/sinks" \
  -H "Authorization: Bearer YOUR_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "kafka-ids",
    "status": "active",
    "source": {
      "include_schemas": null,
      "exclude_schemas": null,
      "include_tables": null,
      "exclude_tables": null
    },
    "actions": ["insert", "update", "delete"],
    "destination": {
      "type": "kafka",
      "hosts": "localhost:9092",
      "tls": false,
      "topic": "records"
    },
    "database": "dune",
    "filter": "",
    "transform": "id-transform",
    "routing": "",
    "message_grouping": true,
    "batch_size": 100,
    "max_retry_count": null,
    "load_shedding_policy": "pause_on_full",
    "timestamp_format": "iso8601",
    "annotations": {}
  }'

{
  "id": "4ed2a8e5-47a7-4b51-9270-d2f4fdcb94fb",
  "name": "my-kafka-sink",
  "status": "active",
  "database": "my-database",
  "source": {
    "include_schemas": null,
    "exclude_schemas": null,
    "include_tables": null,
    "exclude_tables": null
  },
  "tables": [
    {
      "name": "public.products",
      "group_column_names": ["category"]
    }
  ],
  "actions": ["insert", "update", "delete"],
  "destination": {
    "type": "kafka",
    "hosts": "localhost:9092",
    "tls": false,
    "topic": "records",
    "username": "kafka_user",
    "password": "kafka_password",
    "sasl_mechanism": "PLAIN"
  },
  "filter": "none",
  "transform": "id-transform",
  "routing": "none",
  "message_grouping": true,
  "max_retry_count": null,
  "annotations": {},
  "active_backfills": [],
  "batch_size": 50,
  "load_shedding_policy": "pause_on_full",
  "timestamp_format": "iso8601",
  "health": {
    "name": "Consumer health",
    "status": "healthy",
    "checks": [
      {
        "name": "Sink configuration",
        "status": "healthy"
      },
      ...
    ]
  }
}

Get sink consumer Update sink consumer

​Request fields

​Response fields

Request fields

Response fields