| RECORD\_METADATA | ID | FIRST\_NAME | LAST\_NAME | \_STREAMKAP\_SOURCE\_TS\_MS | \_STREAMKAP\_TS\_MS | \_\_DELETED | \_STREAMKAP\_OFFSET |
|---|---|---|---|---|---|---|---|
|
|
1302 | Paul | Smith | 1727770177000 | 1727770177618 | FALSE | 309707 |
Replace `{SCHEMA}` with the name of the schema that contains the table you want to put into capture mode. Likewise, replace `{TABLE}` with the name of the table to put into capture mode. | | Disable change data capture mode on a table | `CALL ASNCDC.REMOVETABLE('{SCHEMA}', '{TABLE}');` | | Reinitialize the ASN capture service | `VALUES ASNCDC.ASNCDCSERVICES('reinit','asncdc');`
Do this after you put a table into capture mode or after you remove a table from capture mode. | # Deployment Options Source: https://docs.streamkap.com/deployment-options Choose between Streamkap Cloud and Bring Your Own Cloud (BYOC) deployment models Streamkap supports the following deployment options: | Option | Description | | :-------------------------------------------------------- | :--------------------------------------------------------------------------- | | [Streamkap Cloud](/streamkap-cloud) | The fastest and easiest way to get started. | | [Bring Your Own Cloud (BYOC)](/bring-your-own-cloud-byoc) | Best suited for companies who can't have data leave their cloud environment. |
Streamkap handles network failures and outages well. If a monitored database stops, the connector resumes from the last recorded log sequence number (LSN) once communication is restored. It retrieves this offset and queries PostgreSQL for a matching LSN in the replication slot.
A replication slot is required for change capture, but PostgreSQL removes slots during upgrades and doesn’t restore them. When the connector restarts, it requests the last known offset, but PostgreSQL cannot return it.
Creating a new replication slot isn't enough to prevent data loss. New slots only track changes from their creation point and lack earlier offsets. The connector fetches its last known offset from Kafka but can't retrieve corresponding data from the new slot. It skips older change events and resumes from the latest log position, causing silent data loss with no warnings.
### Procedure Follow these steps to minimize data loss. Note that a few steps may require support from Streamkap. We recommend notifying us about your database upgrade ahead of time to ensure you have the necessary support. * Using your database's upgrade procedure, ensure writes to it have stopped. * Allow the connector to capture all change events before starting the upgrade procedure. Ask Streamkap to confirm this for you.
## Procedure Follow these steps to minimize data loss. Note that a few steps may require support from Streamkap. We recommend notifying us about your database upgrade ahead of time to ensure you have the necessary support. * Using your database's upgrade procedure, ensure writes to it have stopped. * Allow the connector to capture all change events before starting the upgrade procedure. Ask Streamkap to confirm this for you.
Streamkap handles network failures and outages well. If a monitored database stops, the connector resumes from the last recorded log sequence number (LSN) once communication is restored. It retrieves this offset and queries PostgreSQL for a matching LSN in the replication slot.
A replication slot is required for change capture, but PostgreSQL removes slots during upgrades and doesn’t restore them. When the connector restarts, it requests the last known offset, but PostgreSQL cannot return it.
Creating a new replication slot isn't enough to prevent data loss. New slots only track changes from their creation point and lack earlier offsets. The connector fetches its last known offset from Kafka but can't retrieve corresponding data from the new slot. It skips older change events and resumes from the latest log position, causing silent data loss with no warnings.
### Procedure Follow these steps to minimize data loss. Note that a few steps may require support from Streamkap. We recommend notifying us about your database upgrade ahead of time to ensure you have the necessary support. * Using your database's upgrade procedure, ensure writes to it have stopped. * Allow the connector to capture all change events before starting the upgrade procedure. Ask Streamkap to confirm this for you.
By default, only the primary key and Streamkap metadata column values are retained for deleted records. All other columns will be empty. This leaves you with an incomplete record.
If you require - for auditing and historical tracking purposes - **all** column values for deleted records, or if your deletion strategy for your destination is 'soft deletes' (retain the deleted record with a deletion flag), you should set the `REPLICA IDENTITY` to `FULL` for all capture tables.
By default, changes to partitions are published from the partition itself. The connector expects changes to come from the root table.
To ensure compatibility and consistent replication, enable this setting:
`ALTER PUBLICATION streamkap_pub SET (publish_via_partition_root = true);`
**For `TOPIC` resources:** * `ALL` - All operations * `WRITE` - Write/produce messages * `READ` - Read/consume messages * `ALTER` - Modify resource configurations * `ALTER_CONFIGS` - Modify resource configurations * `CREATE` - Create new resources * `DELETE` - Delete resources * `DESCRIBE` - View resource metadata * `DESCRIBE_CONFIGS` - View resource configurations **For `GROUP` resources (consumers only):** * `READ` - Join and consume from consumer group * `DELETE` - Delete consumer group * `DESCRIBE` - View consumer group metadata * **Pattern Type**: How the resource name is matched * `LITERAL` - Exact match of the resource name * `PREFIXED` - Match resources with the specified prefix * **Name**: The specific resource name or prefix to apply the ACL to ### Connection Details Once a user is created, your endpoints are shown under "Proxy Endpoints". These endpoints follow the naming pattern: `
Ensure certificates are properly configured: ```bash theme={null} pip install --upgrade certifi ``` ```python theme={null} import certifi 'ssl.ca.location': certifi.where() ``` 2. **For CLI tools**
Try different certificate paths: ```bash theme={null} -X ssl.ca.location=/etc/ssl/cert.pem # or -X ssl.ca.location=/etc/ssl/certs/ca-certificates.crt ``` 3. **Disable hostname verification** (temporary): ```python theme={null} 'ssl.endpoint.identification.algorithm': 'none' ``` 4. **Contact support** if issues persist - may require infrastructure team resolution **Note**: Metadata operations (listing topics) may work while data operations fail
* Verify the hostname, port, and network connectivity * Check firewall rules allow traffic on the Redis port * Ensure Redis is configured to accept connections from Streamkap's IP addresses * For Redis Cluster, verify all cluster nodes are accessible * Check TLS/SSL settings match your Redis configuration
* Verify the username and password are correct (usernames are case-sensitive) * For Redis 6.0+, ensure the user has appropriate ACL permissions * Check if `requirepass` is set correctly in redis.conf for Redis versions \< 6.0 * Ensure the user has write permissions to the target database
* Increase the batch size to write more records per batch * Enable Multi-Exec for better transaction handling * Monitor Redis metrics (CPU, memory, network I/O) * Consider using Redis Cluster for horizontal scaling * Check network latency between Streamkap and Redis
* Review your key pattern to ensure uniqueness * Consider including additional fields in the key pattern (e.g., timestamp) * For append operations, use LPUSH/RPUSH instead of SET * Enable TTL to automatically expire old keys * Review your data model and key naming strategy
* Ensure the Redis command matches your data structure * Verify the Kafka record format is compatible with the selected command * Check that nested structures are properly handled for HSET commands * Review the character encoding settings for string data * Validate that sorted set scores are numeric for ZADD commands
```bash theme={null} redis-cli CONFIG GET notify-keyspace-events ``` The response should include `KEA` or the specific event types you need. If not, enable them as described in [Enable Keyspace Notifications](/redis-source-generic#2-enable-keyspace-notifications).
* Verify the username and password are correct * For Redis 6.0+, ensure the user has appropriate ACL permissions * Check if `requirepass` is set in redis.conf for Redis versions \< 6.0 * Verify network connectivity and firewall rules
* Reduce the batch size to lower per-batch load * Increase the poll interval to reduce polling frequency * Monitor Redis metrics (CPU, memory, network I/O) * Consider using a dedicated Redis replica for CDC * Limit keyspace notifications to specific event types instead of using `KEA`
* Verify keyspace notifications include all necessary event types * Check that the connector is running and connected * Review connector logs for errors or warnings * Ensure the Redis user has permissions to read all keys * Verify the database number is correct in the configuration
`new Date("2025-01-01T00:00:00Z")` | `{"$date": "2025-01-01T00:00:00Z"}` | | `NumberLong("9000000000")` | `{"$numberLong": "9000000000"}` | | `NumberDecimal("19.99")` | `{"$numberDecimal": "19.99"}` | | `UUID("3b241101-e2bb-4255-8caf-4136c566a962")` | `{"$binary": {"base64": "OyQRAeK7QlWMr0E2xWapYg==", "subType": "04"}}` | | `/^A.*/i` (regex literal) | `{"$regularExpression": {"pattern": "^A.*", "options": "i"}}` | So a `mongosh` query like: ```javascript theme={null} db.orders.find({ _id: ObjectId("65a1f2b3c4d5e6f7a8b9c0d1") }) ``` becomes the following Streamkap filter: ```json theme={null} {"_id": {"$oid": "65a1f2b3c4d5e6f7a8b9c0d1"}} ``` #### Best Practices for Filtered Snapshots When using Filtered snapshots, we **strongly recommend**: * **Use closed range filters** when applying comparative operators on timestamp or date fields. Closed ranges (lower bound *inclusive*, upper bound *exclusive*) ensure you capture all intended data without gaps or overlaps. * **SQL Source:** `created_at >= '2025-01-01' AND created_at < '2025-02-01'` * **MongoDB-family Source:** `{"created_at": {"$gte": {"$date": "2025-01-01T00:00:00Z"}, "$lt": {"$date": "2025-02-01T00:00:00Z"}}}` * **Filter on indexed or primary key fields** for optimal performance. Filtering on columns that are part of your table's indices or primary key allows the database to efficiently locate matching rows, significantly reducing the load on your source database. * **Make sure your index covers both the filter column AND the ordering column.** When a filtered snapshot starts, it needs to find the highest ordering-column value that matches your filter (MongoDB uses `_id`; SQL uses the primary key). An index on just the filter column — or just the ordering column — isn't enough, and the database may end up scanning a large chunk of the table. Check this with your query planner before triggering on a large table. * **MongoDB-family Source:** if you're filtering on `lastupdated`, the index you need is `{ lastupdated: 1, _id: -1 }`. Verify with: `db.
**For `TOPIC` resources:** * `ALL` - All operations * `WRITE` - Write/produce messages * `READ` - Read/consume messages * `ALTER` - Modify resource configurations * `ALTER_CONFIGS` - Modify resource configurations * `CREATE` - Create new resources * `DELETE` - Delete resources * `DESCRIBE` - View resource metadata * `DESCRIBE_CONFIGS` - View resource configurations **For `GROUP` resources (consumers only):** * `READ` - Join and consume from consumer group * `DELETE` - Delete consumer group * `DESCRIBE` - View consumer group metadata * **Pattern Type**: How the resource name is matched * `LITERAL` - Exact match of the resource name * `PREFIXED` - Match resources with the specified prefix * **Name**: The specific resource name or prefix to apply the ACL to ### Connection Details Once a user is created, your endpoints are shown under "Proxy Endpoints". These endpoints follow the naming pattern: `
Ensure certificates are properly configured: ```bash theme={null} pip install --upgrade certifi ``` ```python theme={null} import certifi 'ssl.ca.location': certifi.where() ``` 2. **For CLI tools**
Try different certificate paths: ```bash theme={null} -X ssl.ca.location=/etc/ssl/cert.pem # or -X ssl.ca.location=/etc/ssl/certs/ca-certificates.crt ``` 3. **Disable hostname verification** (temporary): ```python theme={null} 'ssl.endpoint.identification.algorithm': 'none' ``` 4. **Contact support** if issues persist - may require infrastructure team resolution **Note**: Metadata operations (listing topics) may work while data operations fail