Features

Filters

All sink connectors supports filtering events using regular expressions, JsonPath expressions, or prefixes. The expression is first checked as a JsonPath, then as a regex, and if neither, it's used as a prefix for filtering.

KurrentDB Record

When a connector consumes events from KurrentDB, you have access to the following objects that represent KurrentDB records:

{
  "recordId": "string",
  "position": {
    "streamId": "string",
    "partitionId": "number"
  },
  "sequenceId": "number",
  "isRedacted": "boolean",
  "schemaInfo": {
    "subject": "string",
    "type": "string"
  },
  "headers": {},
  "value": {},
  "streamId": "string",
  "partitionId": "number"
}

{
  "recordId": "46ed7c22-38c9-4a62-bbaf-e40f3d5b84c9",
  "position": {
    "streamId": "VehicleRegistration-5b27b19d80814adeb13471c7a6a1e285",
    "partitionId": -1
  },
  "sequenceId": 1,
  "isRedacted": false,
  "schemaInfo": {
    "subject": "VehicleRegistered",
    "type": "json"
  },
  "headers": {},
  "value": {
    "registrationId": "5b27b19d-8081-4ade-b134-71c7a6a1e285",
    "registrationNumber": 4474795452,
    "vehicle": {
      "make": "Audi",
      "model": "A6",
      "year": 2018,
      "engineType": "petrol"
    }
  },
  "streamId": "VehicleRegistration-5b27b19d80814adeb13471c7a6a1e285",
  "partitionId": -1
}

You can use this schema as a reference when creating your filters. The value object contains the actual event data, while the schemaInfo object contains the event type and subject. The streamId and partitionId properties are used to identify the stream and partition from which the event originated.

Stream ID Filter

The stream ID filter allows you to filter events based on the stream ID. This filter is applied at the stream scope, meaning it filters events by their stream ID. An example of a stream ID filter is shown below:

{
  "subscription:filter:scope": "stream",
  "subscription:filter:filterType": "streamId",
  "subscription:filter:expression": "some-stream"
}

In this case, the filter will only match events from the some-stream stream.

Regex Filters

The simplest and fastest way to filter events is by using regular expressions. This can be done at the stream scope, which applies to the stream ID, or at the record scope, which applies to the event type.

An example of a Regex expression is shown below:

{
  "subscription:filter:scope": "record",
  "subscription:filter:filterType": "regex",
  "subscription:filter:expression": "^eventType.*"
}

This filter will only match records where the event type starts with eventType. This filter type can be used for both stream and record scopes.

Prefix Filters

You can also filter records by prefix using the following configuration:

{
  "subscription:filter:scope": "stream",
  "subscription:filter:filterType": "prefix",
  "subscription:filter:expression": "prefix1,prefix2"
}

In this case, the filter will only match events where the stream ID starts with prefix1 or prefix2. This filter type can be used for both stream and record scopes.

JsonPath Filters

JSONPath provides a standardized string syntax for selecting and extracting JSON values from KurrentDB records. Following the RFC 9535 standard, JSONPath allows for efficient querying of JSON data within your KurrentDB connectors. The filtering process is managed at the connector level and is applied only at the record scope. JSONPath filters apply exclusively to events with the application/json content type.

An example of a JsonPath filter is shown below:

{
  "subscription:filter:scope": "record",
  "subscription:filter:filterType": "jsonPath",
  "subscription:filter:expression": "$[?($.value.vehicle.year==2018)]"
}

In this case, the filter will only match events where the year field is equal to 2018.

Refer to the Subscription Configuration section in settings page for more details on the available scopes and additional configuration options.

Transformations

Connectors support transformations using JavaScript, allowing you to modify the records received from the KurrentDB stream. This feature enables you to tailor the data to meet your specific requirements before it is processed further. Transformations can be applied to any part of the record, providing flexibility in how the data is handled and utilized within your application. It takes a base64 encoded string as input for the transformations.

All records that are transformed will receive a property IsTransformed that will be set to true.

Below is an example of a valid JavaScript function, which will then be encoded into a base64 string:

function transform(record) {
  let { make, model } = record.value.vehicle;
  record.schemaInfo.subject = 'Vehicle';
  record.value.vehicle.makemodel = `${make} ${model}`;
}

The transformation function must be a JavaScript function named transform.

Additionally, it must adhere to the KurrentDB Record structure. Otherwise, it will not start. For example, the following will NOT work:

{
  function transform(record) {
    return {
      name: 'invalid',
    };
  }
}

It also doesn't support advanced Javascript syntax such as generators and tail calls.

Transformation can be configured as follows:

{
  "transformer:enabled": "true",
  "transformer:function": "ZnVuY3Rpb24gdHJhbnNmb3JtKHJlY29yZCkgewogIGxldCB7IG1ha2UsIG1vZGVsIH0gPSByZWNvcmQudmFsdWUudmVoaWNsZTsKICByZWNvcmQuc2NoZW1hSW5mby5zdWJqZWN0ID0gJ1ZlaGljbGUnOwogIHJlY29yZC52YWx1ZS52ZWhpY2xlLm1ha2Vtb2RlbCA9IGAke21ha2V9ICR7bW9kZWx9YDsKfQ=="
}

Checkpointing

Connectors periodically store the position of the last event that they have successfully processed. Then, if the connector host is restarted, the connectors can continue from close to where they got up to. The checkpoint information is stored in the $connectors/{connector-id}/checkpoints system stream in KurrentDB. Each connector has its own dedicated stream for storing checkpoints.

By default, when the connector is started and there are no checkpoints, it will begin from the latest position in the stream. You can configure this behavior using the initialPosition property in the settings. If you need to start from a specific position, you can use the Start API with the position parameter.

Resilience

Currently, connectors only support at least once delivery guarantees, meaning that events may be delivered more than once. Events are delivered in order, ensuring that event x is not delivered until all preceding events have been delivered.

Most connectors have a built-in resilience mechanism to ensure the reliable delivery of data and messages, preserving system integrity and minimizing downtime during unexpected disruptions. To see if a connector supports resilience, refer to the connector's individual page.

The default resilience strategy is as follows:

Automatic Retries and Resilience Strategy

To help systems recover smoothly from disruptions, an automatic retry strategy is in place. This approach uses exponential backoff to manage retry timing after a failure, gradually increasing the time between attempts to balance quick recovery with system stability.

The retry process unfolds in three phases:

1. Phase 1 – Rapid Retries: Right after a failure, the system tries again every 5 seconds for up to 1 minute. This phase addresses minor issues that may quickly resolve on their own.

2. Phase 2 – Moderate Delays: If the issue continues, retries slow to every 10 minutes for the next 1 hour. This phase eases system strain by spacing out attempts, giving more time for resolution.

3. Phase 3 – Extended Delays: For issues that still aren’t resolved, the final phase sets retries to occur every 1 hour indefinitely. This ensures ample recovery time while reducing unnecessary attempts, though manual intervention may eventually be required.

These phases allow the system to progressively adjust retry frequency, balancing resilience and performance.

For more details on customizing these settings, refer to the Resilience Configuration section.

Data Protection

KurrentDB Connectors protect sensitive configuration fields using envelope encryption to secure confidential information such as passwords and access tokens.

When you configure a connector, the system encrypts sensitive data using Data Encryption Keys (DEKs), which are then wrapped (encrypted) using a master key based on your provided token. The encrypted data is stored securely in KurrentDB rather than in plaintext.

Refer to each connector's individual documentation to see which fields are considered sensitive and protected.

Configuring Data Protection

To enable data protection, configure the data protection settings in your configuration file. You can provide an encryption token using either a token file or by specifying the token directly in the configuration.

Using a Token File

Connectors:
  Enabled: true
  DataProtection:
    TokenFile: "/path/to/token/file"

Using a Token Directly

Connectors:
  Enabled: true
  DataProtection:
    Token: "<your-secret-token>"

Key Vault Configuration

KurrentDB Connectors currently only supports the Surge key vault for storing encryption keys:

The Surge key vault is KurrentDB's native key storage mechanism that stores encryption keys directly within KurrentDB itself. Rather than requiring a separate external key management system, Surge stores the encrypted keys in internal system streams. More key vault options will be available in the future.

Connectors:
  Enabled: true
  DataProtection:
    Token: "<your-secret-token>"
    
    KeyVaults:
      Surge:
        KeyMaxHistory: 10