Create sink consumer

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",
    "table": "public.records",
    "filters": [],
    "destination": {
      "type": "kafka",
      "hosts": "localhost:9092",
      "tls": false,
      "topic": "records"
    },
    "database": "dune",
    "transform": "id-transform",
    "batch_size": 100,
    "actions": ["insert", "update", "delete"],
    "group_column_names": ["id"]
  }'

{
  "id": "4ed2a8e5-47a7-4b51-9270-d2f4fdcb94fb",
  "name": "my-kafka-sink",
  "status": "active",
  "filters": [],
  "destination": {
    "type": "kafka",
    "hosts": "localhost:9092",
    "tls": false,
    "topic": "records",
    "username": "kafka_user",
    "password": "kafka_password",
    "sasl_mechanism": "PLAIN"
  },
  "database": "dune",
  "transform": "id-transform",
  "schema": "public",
  "max_retry_count": null,
  "annotations": {},
  "active_backfills": [],
  "batch_size": 50,
  "load_shedding_policy": "pause_on_full",
  "timestamp_format": "iso8601",
  "actions": ["insert", "update", "delete"],
  "group_column_names": ["id"]
}

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)

table

string

The source table for the sink consumer in the format schema.table or table (schema defaults to public)

Must specify either this parameter or schema.

schema

string

The source schema for the sink consumer. If this is specified, the sink consumer will stream changes from all tables in this schema.

Must specify either this parameter or table.

filters

array

Array of filters to apply to the source data

destination

object

required

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

database

string

required

The source database for the sink consumer

transform

string

The transform to apply to messages

batch_size

integer

Number of records to batch together (1-1000)

actions

array

required

The actions to perform on the records (insert, update, delete)

group_column_names

array

Column names used for grouping records

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.

ack_wait_ms

integer

Acknowledgement timeout in milliseconds (minimum 500ms)

annotations

object

Additional metadata you can attach to this sink consumer. Annotations can be any JSON object (not a list or a scalar). You can use annotations in your transform, to attach additional information about the sink to each message.

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)

table

string

The source table for the sink consumer.

The table name is of the format schema.table. Schema is optional and defaults to public.

Mutually exclusive with schema.

schema

string

The source schema for the sink consumer.

The sink consumer will stream changes from all tables in this schema.

Mutually exclusive with table.

filters

array

Array of filters to apply to the source data

destination

object

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

database

string

The source database for the sink consumer.

When you create a database in Sequin, you specify the name of the database resource. That value is this database field.

transform

string

The transform to apply to messages.

This is the name of the transform that you created in the Sequin console.

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.

annotations

object

User-defined annotations for the sink consumer.

This is an object containing key-value pairs of metadata.

active_backfills

array

Array of active backfill IDs.

Backfills that were completed or cancelled will not appear here.

batch_size

integer

Number of records to batch together (1-1000)

This is the number of records that will be sent in a single message to the destination.

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.

actions

array

The database actions to include in the sink.

Possible values include insert, update, delete.

group_column_names

array

Column names used for grouping records

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",
    "table": "public.records",
    "filters": [],
    "destination": {
      "type": "kafka",
      "hosts": "localhost:9092",
      "tls": false,
      "topic": "records"
    },
    "database": "dune",
    "transform": "id-transform",
    "batch_size": 100,
    "actions": ["insert", "update", "delete"],
    "group_column_names": ["id"]
  }'

{
  "id": "4ed2a8e5-47a7-4b51-9270-d2f4fdcb94fb",
  "name": "my-kafka-sink",
  "status": "active",
  "filters": [],
  "destination": {
    "type": "kafka",
    "hosts": "localhost:9092",
    "tls": false,
    "topic": "records",
    "username": "kafka_user",
    "password": "kafka_password",
    "sasl_mechanism": "PLAIN"
  },
  "database": "dune",
  "transform": "id-transform",
  "schema": "public",
  "max_retry_count": null,
  "annotations": {},
  "active_backfills": [],
  "batch_size": 50,
  "load_shedding_policy": "pause_on_full",
  "timestamp_format": "iso8601",
  "actions": ["insert", "update", "delete"],
  "group_column_names": ["id"]
}

Get sink consumer Update 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",
    "table": "public.records",
    "filters": [],
    "destination": {
      "type": "kafka",
      "hosts": "localhost:9092",
      "tls": false,
      "topic": "records"
    },
    "database": "dune",
    "transform": "id-transform",
    "batch_size": 100,
    "actions": ["insert", "update", "delete"],
    "group_column_names": ["id"]
  }'

{
  "id": "4ed2a8e5-47a7-4b51-9270-d2f4fdcb94fb",
  "name": "my-kafka-sink",
  "status": "active",
  "filters": [],
  "destination": {
    "type": "kafka",
    "hosts": "localhost:9092",
    "tls": false,
    "topic": "records",
    "username": "kafka_user",
    "password": "kafka_password",
    "sasl_mechanism": "PLAIN"
  },
  "database": "dune",
  "transform": "id-transform",
  "schema": "public",
  "max_retry_count": null,
  "annotations": {},
  "active_backfills": [],
  "batch_size": 50,
  "load_shedding_policy": "pause_on_full",
  "timestamp_format": "iso8601",
  "actions": ["insert", "update", "delete"],
  "group_column_names": ["id"]
}

Management API

​Request fields

​Response fields

Request fields

Response fields