> ## Documentation Index
> Fetch the complete documentation index at: https://docs.streamnative.io/llms.txt
> Use this file to discover all available pages before exploring further.

> The official ElasticSearch Kafka Connect Sink connector.

# Kafka connect elasticsearch sink

The ElasticSearch Kafka Connect Sink connector is a Kafka Connect connector that writes data from Kafka topics to ElasticSearch.

<Note title="✅ Available on StreamNative Cloud">
  This connector is available as a built-in connector on StreamNative Cloud.
</Note>

### Prerequisites

* A running ElasticSearch cluster

### Quick Start

1. Setup the kcctl client: [doc](https://docs.streamnative.io/docs/kafka-connect-setup)
2. Set up a ElasticSearch cluster
3. Create a JSON file like the following:

   ```json theme={null}
   {
       "name": "elasticsearch-sink",
       "config": {
           "connector.class": "io.aiven.connect.elasticsearch.ElasticsearchSinkConnector",
           "tasks.max": "1",
           "topics": "kafka-elastic-input",
           "connection.url": "http://elastic:9200",
           "type.name": "kafka-connect",
           "key.ignore": "true",
           "schema.ignore": "true",
           "value.converter": "org.apache.kafka.connect.json.JsonConverter",
           "value.converter.schemas.enable": "false"
       }
   }
   ```
4. Run the following command to create the connector:

   ```bash theme={null}
   kcctl apply -f <filename>.json
   ```

### Configuration

The ElasticSearch Kafka Connect Sink connector is configured using the following properties:

| Parameter | Required | Description | Default | |
| - | - | - | - | - |
| connection.url | true | List of Elasticsearch HTTP connection URLs e.g. `http://eshost1:9200, http://eshost2:9200` | | |
| type.name | true | The Elasticsearch type name to use when indexing. | | |
| connection.username | false | The username used to authenticate with Elasticsearch. | | |
| connection.password | false | The password used to authenticate with Elasticsearch. | | |
| batch.size | false | The number of records to process as a batch when writing to Elasticsearch. | 2000 | |
| max.in.flight.requests | false | The maximum number of indexing requests that can be in-flight to Elasticsearch before blocking further requests. | 5 | |
| max.buffered.records | false | The maximum number of records each task will buffer before blocking acceptance of more records. | 20000 | |
| linger.ms | false | Linger time in milliseconds for batching. | 1 | |
| flush.timeout.ms | false | The timeout in milliseconds to use for periodic flushing. | 10000 | |
| max.retries | false | The maximum number of retries that are allowed for failed indexing requests. | 5 | |
| retry.backoff.ms | false | How long to wait in milliseconds before attempting the first retry of a failed indexing. | 100 | |
| key.ignore | false | Whether to ignore the record key for the purpose of forming the Elasticsearch document ID. | false | |
| topic.key.ignore | false | List of topics for which `key.ignore` should be true. | | |
| schema.ignore | false | Whether to ignore schemas during indexing. | false | |
| topic.schema.ignore | false | List of topics for which `schema.ignore` should be true. | | |
| drop.invalid.message | false | Whether to drop kafka message when it cannot be converted to output message. | false | |
| compact.map.entries | false | Defines how map entries with string keys within record values should be written to JSON. | true | |
| connection.timeout.ms | false | How long to wait in milliseconds when establishing a connection to the Elasticsearch server. | 1000 | |
| read.timeout.ms | false | How long to wait in milliseconds for the Elasticsearch server to send a response. | 3000 | |
| behavior.on.null.values | false | How to handle records with a non-null key and a null value, Valid options are 'ignore', 'delete', and 'fail'. | ignore | |
| behavior.on.malformed.documents | false | How to handle records that Elasticsearch rejects due to some malformation of the document itself, Valid options are 'ignore', 'warn', and 'fail'. | fail | |

For more information about the configuration properties, see the [Official ElasticSearch Kafka Connect Sink Connector documentation](https://github.com/Aiven-Open/elasticsearch-connector-for-apache-kafka/blob/v7.0.0/README.md).


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.