Amazon RDS PostgreSQL Serverless

Configure PostgreSQL CDC on Amazon RDS Serverless with Streamkap, covering parameter groups, replication user setup, and heartbeat options.

Prerequisites

• PostgreSQL version ≥ 10
• A database user with sufficient privileges to configure the database, including enabling logical replication and creating users

PostgreSQL Setup

1. Grant Database Access

• Configure one of the Connection Options to ensure Streamkap can reach your database.

2. Enable Logical Replication

Logical replication is a method of replicating data objects and their changes, based upon their replication identity (usually a primary key). The Connector relies on PostgreSQLs implementation of this.

• Open the Amazon RDS console at https://console.aws.amazon.com/rds/.
• In the navigation pane, choose Parameter groups.

• Select the parameter group to edit.
• Choose Edit from Actions.
• Set rds.logical_replication to 1.
• Set wal_sender_timeout to 0. A nonzero value may cause disconnects in low/intermittent traffic databases. Enable Heartbeats or set an appropriate value if needed.
• Choose Save changes.

If you created a new parameter group, associate it with your Aurora DB cluster:

• In the navigation pane, choose Databases and select the target DB cluster.
• Choose Modify.
• Change the DB cluster parameter group setting.
• Choose Continue and review modifications.
• The change is applied immediately, regardless of the Scheduling of modifications setting.
• On the confirmation page, choose Modify cluster.

A reboot is required to apply the changes.

3. Create Database User

It’s recommended to create a separate user and role for the Connector to access your PostgreSQL database. Below is an example script that does that.

-- Replace { ... } placeholders as required
CREATE USER streamkap_user PASSWORD '{password}';

-- Create a role for Streamkap
CREATE ROLE streamkap_role;
GRANT streamkap_role TO streamkap_user;
GRANT rds_replication TO streamkap_role;

-- Grant Streamkap permissions on the database, schema and all tables to capture
GRANT CONNECT ON DATABASE "{database}" TO streamkap_role; 
GRANT CREATE, USAGE ON SCHEMA "{schema}" TO streamkap_role;
GRANT SELECT ON ALL TABLES IN SCHEMA "{schema}" TO streamkap_role;
ALTER DEFAULT PRIVILEGES IN SCHEMA "{schema}" GRANT SELECT ON TABLES TO streamkap_role;

4. Enable Snapshots

To backfill your data, the Connector needs to be able to perform snapshots. See Snapshots & Backfilling for more information.

To enable this feature, there are 2 methods available:

Method 1: Enable read only connection

This method is recommended if you cannot create a table in the source database and grant the Connector read/write privileges to that.

• Set Read only to Yes during Streamkap Setup. No other configuration should be necessary.

Method 2: Create a table in the source database

You will need to create the table and give necessary permissions to the streamkap_user. The Connector will use this collection for managing snapshots. Below is an example script that does that.

-- Create the schema
CREATE SCHEMA streamkap;

-- Switch to the newly created schema
SET search_path TO streamkap;

-- Create the table
CREATE TABLE streamkap_signal (
  id VARCHAR(255) PRIMARY KEY, 
  type VARCHAR(32) NOT NULL, 
  data VARCHAR(2000) NULL
);

-- Grant necessary privileges on the table to the role
GRANT CREATE, USAGE ON SCHEMA streamkap TO streamkap_role;
GRANT SELECT ON ALL TABLES IN SCHEMA streamkap TO streamkap_role;
GRANT SELECT, UPDATE, INSERT, DELETE ON TABLE streamkap_signal TO streamkap_role;

5. Heartbeats

Connectors use “offsets”—like bookmarks—to track their position in the database’s log or change stream. When no changes occur for long periods, these offsets may become outdated, and the Connector might lose its place or stop capturing changes.

Heartbeats ensure the Connector stays active and continues capturing changes.

There are two layers of heartbeat protection:

Layer 1: Connector heartbeats (enabled by default)

The Connector periodically emits heartbeat messages to an internal topic, even when no actual data changes are detected. This keeps offsets fresh and prevents staleness.

No configuration is necessary for this layer; it is automatically enabled. We recommend keeping this layer enabled for all deployments.

Layer 2: Source database heartbeats (recommended)

You can configure regular updates to a dedicated heartbeat table in the source database. This simulates activity, ensuring change events are generated consistently, maintaining log progress and providing additional resilience.

How this layer is configured depends on the connection type (if supported by the Source):

• Read-write connections (when Read only is No during Streamkap Setup): The Connector updates the heartbeat table directly.
• Read-only connections (when Read only is Yes during Streamkap Setup): A scheduled job on the primary database updates the heartbeat table, and these changes replicate to the read replica for the Connector to consume.

This layer requires you to set up a heartbeat table—and for read-only connections, a scheduled job (e.g., pg_cron for PostgreSQL, event_scheduler for MySQL)—on your source database.

-- Create the streamkap schema
CREATE SCHEMA IF NOT EXISTS streamkap;

-- Switch to the streamkap schema
SET search_path TO streamkap;

-- Create the heartbeat table with id, text, and last_update fields
CREATE TABLE streamkap_heartbeat (
    id SERIAL PRIMARY KEY,
    text TEXT,
    last_update TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Grant permission to the Streamkap user
GRANT USAGE ON SCHEMA streamkap TO streamkap_user;
GRANT SELECT, UPDATE, INSERT, DELETE ON TABLE streamkap_heartbeat TO streamkap_user;

-- Insert the first row into the heartbeat table
INSERT INTO streamkap_heartbeat (text) VALUES ('test_heartbeat');

CREATE EXTENSION IF NOT EXISTS pg_cron;

-- Create the streamkap schema
CREATE SCHEMA IF NOT EXISTS streamkap;

-- Create the heartbeat table
CREATE TABLE streamkap.streamkap_heartbeat (
    id SERIAL PRIMARY KEY,
    text TEXT,
    last_update TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Insert the initial row
INSERT INTO streamkap.streamkap_heartbeat (text) VALUES ('test_heartbeat');

-- Schedule a job to update the heartbeat table every minute
SELECT cron.schedule(
    'streamkap_heartbeat_job',
    '*/1 * * * *',
    $$UPDATE streamkap.streamkap_heartbeat SET text = 'updated_heartbeat', last_update = CURRENT_TIMESTAMP WHERE id = 1;$$
);

GRANT USAGE ON SCHEMA streamkap TO {cron user};
GRANT SELECT, UPDATE, INSERT, DELETE ON streamkap.streamkap_heartbeat TO {cron user};

-- Grant permissions to the Streamkap user for monitoring and diagnostics
GRANT USAGE ON SCHEMA streamkap TO streamkap_user;
GRANT SELECT, UPDATE, INSERT, DELETE ON streamkap.streamkap_heartbeat TO streamkap_user;

SELECT cron.schedule(
    'streamkap_cron_cleanup',
    '0 0 * * *',
    $$DELETE FROM cron.job_run_details WHERE end_time < now() - interval '7 days';$$
);

-- View all scheduled jobs
SELECT * FROM cron.job;

-- View recent job execution history
SELECT * FROM cron.job_run_details ORDER BY start_time DESC LIMIT 10;

-- Unschedule a job (replace {jobid} with the actual job ID)
SELECT cron.unschedule({jobid});

SELECT slot_name, confirmed_flush_lsn, restart_lsn
FROM pg_replication_slots
WHERE slot_name = 'streamkap_pgoutput_slot';

6. Create Publication & Slot

Publications contain a set of change events for the tables you want the Connector to capture.

• Create a publication for your tables. You can create a publication for all tables or selected tables.

-- Create a publication for all tables to capture
CREATE PUBLICATION streamkap_pub FOR ALL TABLES;

-- Create a publication for specific tables to capture
CREATE PUBLICATION streamkap_pub FOR TABLE table1, table2, table3, ...;

-- Verify the tables to capture were added to the publication
SELECT * FROM pg_publication_tables where pubname = 'streamkap_pub';

A replication slot represents a stream of change events the Connector reads from.

• Create a replication slot.

-- Create a logical replication slot
SELECT pg_create_logical_replication_slot('streamkap_pgoutput_slot', 'pgoutput');

-- Verify the replication slot is working (this may take a few moments to return the count)
SELECT count(*) FROM pg_logical_slot_peek_binary_changes('streamkap_pgoutput_slot', null, null, 'proto_version', '1', 'publication_names', 'streamkap_pub');

Excluding columns from replication

By default a publication streams every column of each captured table. On PostgreSQL 15+ you can attach a column list to a table in the publication so that only the listed columns are replicated — any column you leave out is excluded at the source and never enters the replication stream.

-- Only the listed columns are published; every other column is excluded.
-- Each table's list MUST include its replica identity (primary key),
-- otherwise UPDATE/DELETE for that table cannot be replicated.
CREATE PUBLICATION streamkap_pub FOR TABLE
  public.users    (id, email, created_at, updated_at),
  public.orders   (id, user_id, status, total),
  public.products (id, sku, name, price);

-- Add or change a table's column list on an existing publication
ALTER PUBLICATION streamkap_pub SET TABLE public.users (id, email, updated_at);

-- Verify the published columns per table
SELECT pubname, schemaname, tablename, attnames
FROM pg_publication_tables WHERE pubname = 'streamkap_pub';

Streamkap Setup

Follow these steps to configure your new connector:

1. Create the Source

• Navigate to Add Connectors.
• Choose PostgreSQL.

2. Connection Settings

• Name: Enter a name for your connector.

• Hostname: Specify the hostname.

• Port: Default is 5432.

• Connect via SSH Tunnel: The Connector will connect to an SSH server in your network which has access to your database. This is necessary if the Connector cannot connect directly to your database.

  • See SSH Tunnel for setup instructions.

• Username: Username to access the database. By default, Streamkap scripts use streamkap_user.

• Password: Password to access the database.

• Database: Specify the database to stream data from.

• Read only: Whether or not to use a read-only connection. Requires PostgreSQL version 13 or higher.

• Heartbeats: Enabled by default. Choose one of three modes—see Heartbeats for setup instructions:

  • Read-write connections: configure a heartbeat table in the source database and set Heartbeat Table Schema.

  • Read-only connections: configure a scheduled heartbeat job on the primary database using pg_cron, and include the heartbeat table in Schema and Table Capture.

  • Logical-message (PostgreSQL 14+, works under read-only): set Use Logical-Message Heartbeat to Yes and leave Heartbeat Table Schema blank. No table or write privileges required on the source.

3. Snapshot Settings

• Signal Table: Full path to the signal table including schema and table name (e.g., streamkap.streamkap_signal). This table is used for incremental snapshotting. See Enable Snapshots for setup instructions.

4. Replication Settings

• Replication Slot Name: The name of the replication slot for the connector to use. Default is streamkap_pgoutput_slot.
• Publication Name: The name of the publication for the connector to use. Default is streamkap_pub.

5. Advanced Parameters

• SSL mode: Whether to use an encrypted connection to the PostgreSQL server. By default, it’s required.
• Prefix with Database Name?: Changes the format of topics to DatabaseName_TopicName
• Represent binary data as: Specifies how the data for binary columns e.g. blob, binary, varbinary should be interpreted. Your destination for this data can impact which option you choose. Default is bytes.
• Re-select Post Processor (Advanced): When a column’s value is stored out-of-line in TOAST, it is not written to the WAL, so the connector re-fetches it from the source database at event time. Re-selection was always on and not configurable before, so the default is true for backwards compatibility rather than as a recommendation; turning it off means those values arrive as unavailable-value placeholders instead. Re-selection costs one extra query against the source per affected row — setting REPLICA IDENTITY FULL on the affected tables avoids it entirely. Default is true. See PostgreSQL Source FAQ for details.
• Re-select Error Handling (Advanced): What happens when a re-select fails. Fail (the default) stops the connector; Warn logs a warning and continues with the value missing. Only shown when the Re-select Post Processor is enabled.

Click Next.

6. Schema and Table Capture

• Add Schemas/Tables: Specify the schema(s) and table(s) for capture
  • You can bulk upload here. The format is a simple list of schemas and tables, with each entry on a new row. Save as a .csv file without a header.

Click Save.

Troubleshooting

ALTER TABLE {table} REPLICA IDENTITY FULL;