Elasticsearch Sink

Overview

The Elasticsearch sink pulls messages from a KurrentDB stream and stores them in an Elasticsearch index. The records will be serialized into JSON documents, compatible with Elasticsearch's document structure.

Quickstart

You can create the Elasticsearch Sink connector as follows:

$JSON = @"
{
  "settings": {
    "instanceTypeName": "elasticsearch-sink",
    "url": "http://localhost:9200",
    "indexName": "sample-index",
    "subscription:filter:scope": "stream",
    "subscription:filter:filterType": "streamId",
    "subscription:filter:expression": "example-stream"
  }
}
"@ `

curl.exe -X POST `
  -H "Content-Type: application/json" `
  -d $JSON `
  http://localhost:2113/connectors/elasticsearch-sink-connector

JSON='{
  "settings": {
    "instanceTypeName": "elasticsearch-sink",
    "url": "http://localhost:9200",
    "indexName": "sample-index",
    "subscription:filter:scope": "stream",
    "subscription:filter:filterType": "streamId",
    "subscription:filter:expression": "example-stream"
  }
}'

curl -X POST \
  -H "Content-Type: application/json" \
  -d "$JSON" \
  http://localhost:2113/connectors/elasticsearch-sink-connector

After creating and starting the Elasticsearch sink connector, every time an event is appended to the example-stream, the Elasticsearch sink connector will send the record to the specified index in Elasticsearch. You can find a list of available management API endpoints in the API Reference.

Settings

Adjust these settings to specify the behavior and interaction of your Elasticsearch sink connector with KurrentDB, ensuring it operates according to your requirements and preferences.

The Elasticsearch sink can be configured with the following options:

Authentication

The Elasticsearch sink connector supports multiple authentication methods.

Basic Authentication

To use Basic Authentication, set the method to basic and provide the username and password:

{
  "authentication:method": "basic",
  "authentication:username": "elastic",
  "authentication:password": "your_secure_password"
}

API Key Authentication

To use API Key authentication, set the method to apiKey and provide the API key:

{
  "authentication:method": "apiKey",
  "authentication:apiKey": "your_api_key"
}

Token Authentication

To use Token authentication, set the method to token and provide the Base64 encoded API key:

{
  "authentication:method": "token",
  "authentication:base64ApiKey": "your_base64_encoded_api_key"
}

Certificate Authentication

To configure Mutual TLS authentication, provide the client and root certificates:

{
  "authentication:clientCertificate:rawData": "base64_encoded_client_certificate",
  "authentication:clientCertificate:password": "certificate_password",
  "authentication:rootCertificate:rawData": "base64_encoded_root_certificate",
  "authentication:rootCertificate:password": "certificate_password"
}

Document ID

The ID of the document can be generated automatically based on the source specified and expression if needed.

By default, the Elasticsearch sink uses the recordId as the document ID. This is the unique identifier generated for every record in KurrentDB.

If you opt not to use the default recordId, the following are alternative configuration options to generate document IDs based on different sources:

Set Document ID using Stream ID

You can extract part of the stream name using a regular expression (regex) to define the document id. The expression is optional and can be customized based on your naming convention. In this example, the expression captures the stream name up to _data.

{
  "documentId:source": "stream",
  "documentId:expression": "^(.*)_data$"
}

Alternatively, if you only need the last segment of the stream name (after a hyphen), you can use the streamSuffix source. This doesn't require an expression since it automatically extracts the suffix.

{
  "documentId:source": "streamSuffix"
}

The streamSuffix source is useful when stream names follow a structured format, and you want to use only the trailing part as the document ID. For example, if the stream is named user-123, the document ID would be 123.

Set Document ID from Headers

You can generate the document ID by concatenating values from specific event headers. In this case, two header values (key1 and key2) are combined to form the ID.

{
  "documentId:source": "headers",
  "documentId:expression": "key1,key2"
}

The Headers source allows you to pull values from the event's metadata. The documentId:expression field lists the header keys (in this case, key1 and key2), and their values are concatenated to generate the document ID. This is useful when headers hold important metadata that should define the document's unique identifier, such as region, user ID, or other identifiers.

{
  "key1": "value1",
  "key2": "value2"
}

// outputs "value1-value2"

Set Document ID from Partition Key

If your event has a partition key, you can use it as the document ID. The partitionKey source directly uses this key without requiring an expression.

{
  "documentId:source": "partitionKey"
}

This uses the record's partition key as a unique document ID.

Resilience Configuration

The Elasticsearch sink provides extensive resilience options to handle connection issues and retry failed operations:

{
  "resilience:enabled": true,
  "resilience:connectionLimit": 80,
  "resilience:maxRetries": 5,
  "resilience:requestTimeout": 30000,
  "resilience:retryOnErrorTypes": "timeout,server_error"
}

This configuration enables resilience, limits connections to 80, sets a maximum of 5 retries, sets a request timeout of 30 seconds, and specifies that the connector should retry on timeout and server error types.