> ## Documentation Index > Fetch the complete documentation index at: https://docs.automq.com/llms.txt > Use this file to discover all available pages before exploring further. # Elasticsearch Source Connector > Configure and operate the Elasticsearch Source Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting. ## Overview Elasticsearch Source Connector reads search-visible documents from Elasticsearch indices into Kafka for index data distribution, analytics ingestion, and document state synchronization. Each index maps to one Topic, whose name is formed by directly concatenating the specified prefix and the index name. The Connector polls documents through the Search API. When no read position has been saved, it reads the current documents, then reads data beyond the watermark in ascending business cursor order. This is neither a change log subscription nor CDC, and it does not provide a consistent snapshot: updates can only be read again if the cursor advances, and deletions do not generate events. Message values are document structures with a Schema, augmented with the document identifier and index name. Message keys consist of the index name and cursor values, rather than a stable document ID. ## Prerequisites * Elasticsearch indices must retain readable `_source` data. Documents must provide a non-null, top-level cursor field that supports sorting and range queries. A single-field cursor must be unique, and subsequent inserts and updates that need synchronization must strictly exceed the watermark already read. * The Elasticsearch account must be able to query the selected indices and access the index discovery endpoint `/_cat/indices`. The Connector accesses this endpoint even when a fixed index list is used. * When using custom TLS material, PKCS12 files must be readable on every Worker that may run the Connector and its Tasks. Before configuring passwords, restrict configuration initialization logging and log access to prevent password disclosure. ## License Uses Apache License 2.0. ## Quick Start Prepare a Connect Cluster, Kafka, and an Elasticsearch index containing a cursor field, and confirm network connectivity and access permissions. See [Manage Connectors](../manage-connectors) for creation and management operations. The following example is for a controlled environment without authentication. Index documents use the top-level numeric field `change_seq` as a unique, increasing cursor. ```properties theme={null} connector.class=com.github.dariobalinzo.ElasticSourceConnector es.host= es.port=9200 index.names= incrementing.field.name=change_seq topic.prefix=es- key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.json.JsonConverter ``` Replace the host and index name. Ensure that the mapping for `change_seq` supports sorting and range queries, and that documents actually contain the field. The host must not include a scheme, port, or path. The output Topic name directly combines `es-` with the index name. With no previous read position, the Connector reads currently visible documents; subsequent inserts or updates must advance the cursor and become search-visible in Elasticsearch. The example explicitly specifies JsonConverter and leaves `value.converter.schemas.enable` unset, so it uses that Converter's default of `true` and outputs `schema` and `payload`; it does not inherit this sub-setting from the Worker. The default field name conversion changes `es-id` and `es-index` to `esid` and `esindex`. ## Configuration ### Elasticsearch Connection #### `es.host` Specifies the Elasticsearch host. * **Category**: Elasticsearch Connection * **Type**: `string` * **Default**: None, required * **Importance**: High * **Valid Values / Notes**: Must not include a scheme, port, or path. Separate multiple hosts with semicolons; they share the same port and scheme. Whitespace is not automatically trimmed from entries. #### `es.port` Specifies the port used by all Elasticsearch hosts. * **Category**: Elasticsearch Connection * **Type**: `string` * **Default**: None, required * **Importance**: High * **Valid Values / Notes**: Parsed as an integer at startup. Specify a valid TCP port from `1` to `65535`; there is no explicit range validation. #### `es.scheme` Specifies the connection scheme. * **Category**: Elasticsearch Connection * **Type**: `string` * **Default**: `http` * **Importance**: Medium * **Valid Values / Notes**: Use `http` or `https`. Configuring TLS files does not automatically switch the scheme to HTTPS. Use HTTPS when authenticating with a username and password in production. #### `es.user` Specifies the HTTP Basic authentication username. * **Category**: Elasticsearch Connection * **Type**: `string` * **Default**: `null` * **Importance**: High * **Valid Values / Notes**: A non-empty value enables username/password authentication and should be accompanied by a password. An empty value disables this authentication. Restrict connection targets to trusted hosts. #### `es.password` Specifies the HTTP Basic authentication password. * **Category**: Elasticsearch Connection * **Type**: `string` * **Default**: `null` * **Importance**: High * **Valid Values / Notes**: Used only when the username is non-empty. This setting is not a password type that is automatically masked; initialization INFO logs may expose the resolved password. Before production use, restrict initialization configuration logging for `com.github.dariobalinzo.ElasticSourceConnectorConfig` and `com.github.dariobalinzo.task.ElasticSourceTaskConfig`, and check log access permissions. Config Provider variable substitution does not eliminate this risk. Do not include real credentials in configuration examples, troubleshooting materials, or shared logs. #### `es.tls.truststore.location` Specifies the path to a custom CA truststore. * **Category**: Elasticsearch Connection * **Type**: `string` * **Default**: `null` * **Importance**: Medium * **Valid Values / Notes**: Only PKCS12 is supported. The file must be readable on the relevant Workers; use it with `https`. Leaving this setting unset does not disable the client's default TLS validation. #### `es.tls.truststore.password` Specifies the truststore password. * **Category**: Elasticsearch Connection * **Type**: `string` * **Default**: Empty string * **Importance**: Medium * **Valid Values / Notes**: Must match the file password when loading the truststore and must not be `null`. Initialization configuration logs also pose a disclosure risk for this password; implement password logging protections first. #### `es.tls.keystore.location` Specifies the path to the keystore containing the client certificate and private key. * **Category**: Elasticsearch Connection * **Type**: `string` * **Default**: `null` * **Importance**: Medium * **Valid Values / Notes**: Only PKCS12 is supported. The client keystore is loaded only when a truststore path is also configured. The file must be readable on the relevant Workers. #### `es.tls.keystore.password` Specifies the client keystore password. * **Category**: Elasticsearch Connection * **Type**: `string` * **Default**: Empty string * **Importance**: Medium * **Valid Values / Notes**: The same password is used to open the keystore and read the private key; a separate private key password is not supported. Initialization configuration logs pose a disclosure risk; implement password logging protections first. #### `connection.attempts` Specifies the total number of attempts when a search encounters an I/O error. * **Category**: Elasticsearch Connection * **Type**: `string` * **Default**: `3` * **Importance**: Low * **Valid Values / Notes**: Parsed as an integer and must be greater than `0`. Applies only to search retries for I/O exceptions, not to all exceptions or index discovery requests. #### `connection.backoff.ms` Specifies the interval between search I/O retries, in milliseconds. * **Category**: Elasticsearch Connection * **Type**: `string` * **Default**: `10000` * **Importance**: Low * **Valid Values / Notes**: Parsed as a long integer and must be non-negative; a positive value is normally used. Retry waits increase data read latency. ### Index Selection and Read Position #### `index.names` Specifies a fixed index list. * **Category**: Index Selection and Read Position * **Type**: `string` * **Default**: `null` * **Importance**: Medium * **Valid Values / Notes**: A comma-separated list of explicit index names, without wildcards. Whitespace is not automatically trimmed. Whenever the submitted configuration contains this key, the fixed list takes precedence. Remove this setting when not using a fixed list; an empty string or explicit `null` is not a substitute for omitting it. #### `index.prefix` Selects indices by name prefix. * **Category**: Index Selection and Read Position * **Type**: `string` * **Default**: Empty string * **Importance**: Medium * **Valid Values / Notes**: A single literal prefix, not a list, regular expression, or glob. An empty string matches all index names. Used for selection only when `index.names` is omitted; the monitoring thread still uses this prefix. #### `incrementing.field.name` Specifies the primary cursor field. * **Category**: Index Selection and Read Position * **Type**: `string` * **Default**: Empty string * **Importance**: Medium * **Valid Values / Notes**: The default passes configuration parsing but cannot form a valid query, so a field must be specified for actual use. Prefer a non-null, top-level numeric field in `_source`. A single cursor must be unique, and subsequent data must strictly exceed the watermark. Do not assume a nested filter path, `_id`, or the injected `es-id` is a usable cursor, and do not directly use a `.keyword` path for a single cursor. #### `incrementing.secondary.field.name` Specifies the secondary sort field for documents with the same primary cursor. * **Category**: Index Selection and Read Position * **Type**: `string` * **Default**: `null` * **Importance**: Low * **Valid Values / Notes**: Any non-`null` value enables this setting; an empty string does not disable it. The field must support sorting and range queries and have a retrievable value in every document. The primary field should support exact matching. Combined values should be unique, and subsequent combinations must strictly exceed the watermark. A dual cursor cannot recover late-arriving data already behind the watermark. #### `mode` A retained mode selection setting that does not change the current read behavior. * **Category**: Index Selection and Read Position * **Type**: `string` * **Default**: Empty string * **Importance**: High * **Valid Values / Notes**: Accepts an empty string, `bulk`, `timestamp`, `incrementing`, and `timestamp+incrementing`, but has no corresponding execution branches. Setting it is not recommended; it cannot enable a separate timestamp or bulk mode. This setting is not formally marked as deprecated. ### Output Documents #### `topic.prefix` Specifies the output Topic name prefix. * **Category**: Output Documents * **Type**: `string` * **Default**: None, required * **Importance**: High * **Valid Values / Notes**: An empty string is allowed. The prefix is directly concatenated with the index name without inserting a separator. Changing the prefix does not reset the read position or automatically migrate historical data. #### `fieldname_converter` Specifies how document field and Schema names are converted. * **Category**: Output Documents * **Type**: `string` * **Default**: `avro` * **Importance**: Medium * **Valid Values / Notes**: The exact lowercase value `nop` preserves names; all other values use Avro name conversion. Conversion adds an `avro` prefix to names that do not start with a letter and removes characters other than ASCII letters and digits, potentially causing name collisions. It does not enable Avro serialization. #### `filters.whitelist` Retains only the specified document fields. * **Category**: Output Documents * **Type**: `string` * **Default**: `null` * **Importance**: Medium * **Valid Values / Notes**: Semicolon-separated paths, with support for dotted object paths. Whitespace is not automatically trimmed. Any non-`null` value enables this setting. Explicitly retain `es-id` and `es-index` when metadata is needed. Filtering affects only values, not message keys or read positions. #### `filters.blacklist` Removes document fields according to blacklist rules. * **Category**: Output Documents * **Type**: `string` * **Default**: `null` * **Importance**: Medium * **Valid Values / Notes**: Semicolon-separated paths; any non-`null` value enables this setting. Applied after the whitelist. Objects and lists are retained only when a blacklist entry starts with their path, so unmatched composite fields may also be removed. This is not the usual rule of excluding only the specified leaf fields. #### `filters.json_cast` Serializes the specified field values as JSON strings. * **Category**: Output Documents * **Type**: `string` * **Default**: `null` * **Importance**: Medium * **Valid Values / Notes**: Semicolon-separated paths; any non-`null` value enables this setting. Applied after the whitelist and blacklist. It does not parse existing JSON strings; converting a string value retains its JSON quotation marks. ### Operation and Serialization #### `poll.interval.ms` Specifies the wait after a poll returns no data at all, in milliseconds. * **Category**: Operation and Serialization * **Type**: `string` * **Default**: `5000` * **Importance**: High * **Valid Values / Notes**: Parsed as an integer and must be non-negative. Does not throttle reads when data is continuously available, and does not change the fixed 5000-millisecond index discovery interval. #### `batch.max.rows` Specifies the maximum number of documents per search for each index. * **Category**: Operation and Serialization * **Type**: `string` * **Default**: `10000` * **Importance**: Low * **Valid Values / Notes**: Parsed as an integer. Should be positive and must not exceed the Elasticsearch query window limit. This is not the total record limit for a single Task poll; results from multiple indices are added together. #### `connector.class` Specifies the Source Connector implementation class. * **Category**: Operation and Serialization * **Type**: `string` * **Default**: None, required * **Importance**: High * **Valid Values / Notes**: Use `com.github.dariobalinzo.ElasticSourceConnector`. #### `tasks.max` Specifies the maximum number of tasks. * **Category**: Operation and Serialization * **Type**: `int` * **Default**: `1` * **Importance**: High * **Valid Values / Notes**: Must be at least `1`. The actual task count is at most the smaller of the index count and this value. A single index is not split across multiple Tasks. #### `key.converter` Specifies the Converter for serializing message keys. * **Category**: Operation and Serialization * **Type**: `class` * **Default**: `null` * **Importance**: Low * **Valid Values / Notes**: Uses the Worker configuration when not specified at the Connector level. Keys are strings. Quick Start explicitly uses StringConverter; that class is not the default. #### `value.converter` Specifies the Converter for serializing Struct message values with a Schema. * **Category**: Operation and Serialization * **Type**: `class` * **Default**: `null` * **Importance**: Low * **Valid Values / Notes**: Uses the Worker configuration when not specified at the Connector level. Quick Start explicitly uses JsonConverter; that class is not the default. When this setting is explicitly specified, the Converter uses only the Connector's `value.converter.*` sub-settings. Unset sub-settings use the Converter's own defaults; Worker sub-settings are not merged. JsonConverter defaults `schemas.enable` to `true`, so Quick Start omits this default setting. Field name conversion does not automatically enable Schema Registry. ## Best Practices ### Adjust Wait Time and Batch Size During Operation **Applicable Scenario**: Ingestion is already running, and you want to reduce the wait for new data after an idle period while controlling the pressure each read places on Elasticsearch and Worker memory. **Configuration Example**: Add the following two settings to the Quick Start configuration, leaving the remaining settings unchanged. ```properties theme={null} poll.interval.ms=1000 batch.max.rows=1000 ``` **Key Notes**: Use these example values as starting points for tuning, not as universally optimal settings. A shorter idle wait may increase query frequency. Smaller pages reduce the amount of data returned per index per search, but require more search requests to read existing data. Tune based on query duration, Worker memory, and business latency requirements. When data is continuously available, the wait does not limit throughput; results from multiple indices accumulate within a poll. Smaller pages do not solve duplicate cursor problems; first ensure cursor uniqueness and advancement order. ### Expand to Continuously Created Rolling Indices **Applicable Scenario**: Data is written to new indices created by date or period, making a fixed index list increasingly difficult to maintain. You want to discover new indices under the same business prefix automatically and distribute read tasks across multiple indices. **Configuration Example**: Starting from Quick Start, remove `index.names` and add the following settings. Do not retain the fixed list alongside them. ```properties theme={null} index.prefix= tasks.max=2 ``` **Key Notes**: Replace the placeholder with the literal prefix of your business indices, without an asterisk. Prepare at least one matching index before startup; all matching indices should satisfy the same cursor constraints. When monitoring detects a change in the index list, Tasks are reconfigured. Two tasks can be used only when there are at least two indices; this does not scale reads from a single index. A new index without a saved position is read from its currently visible documents. Keep index names, Connector identity, and cursor definitions stable, and do not treat recreating an index under the same name as automatic reinitialization. Use a sufficiently specific prefix to avoid ingesting unrelated indices. ## Monitoring ### What to Monitor Monitor Kafka Connect cluster health, Connector and Task status, input/output throughput, end-to-end latency, Offset commits, errors and retries, and Worker JVM memory, GC, and thread signals. A Task in RUNNING state does not necessarily mean data continues to flow; also observe source changes and destination messages. Monitor DLQ activity only when the deployment has enabled the corresponding error handling. ### Import the Grafana Dashboard Download the shared [AutoMQ Connect Cluster Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json). Confirm that the Grafana data source can query Kafka Connect and Worker JVM metrics and matches the cluster, Connector, Task, and other labels required by the dashboard. Then use Grafana's import feature to load the JSON and select the corresponding data source. ## Limitations * Reads currently visible state through separate searches. Does not provide PIT, scroll, consistent snapshots, change history, or complete capture of intermediate update versions. * A single cursor resumes strictly beyond the watermark. Duplicate cursors spanning pages may cause omissions, and late-arriving data at or below the watermark is not recovered. There is no lookback window or built-in deduplication. * A dual cursor still requires unique combinations and subsequent combinations beyond the watermark. It cannot recover data with smaller secondary values in primary-field groups already passed. * Does not generate deletion events, tombstones, or before/after envelopes. Updates can only be read again when the cursor advances and they become search-visible. * Does not provide custom query or row filtering settings; field filters process output values only. * Recreating an index with the same name does not clear its old read position. Changing the cursor definition or Topic prefix does not automatically reset the position either. * A single index cannot be split across multiple Tasks. Global ordering across indices, Tasks, or multiple downstream partitions is not guaranteed. * The default message key changes with the cursor and is not a stable document upsert key. The added `es-id` and `es-index` overwrite source document fields with the same names. * Schema is inferred per document. Null values and empty arrays are not added as fields. Nested arrays and arrays whose first element is null are not supported by this conversion path, and conflicting types for the same field are not automatically reconciled. * Does not provide unconditional lossless or exactly-once guarantees. Recovery may reread data, so downstream processing must account for duplicates. * Search retries cover only I/O exceptions. Some query or conversion exceptions are logged within the Task before it returns, without automatically entering framework retries or a DLQ. * Custom TLS material supports only PKCS12, and the client keystore must be loaded together with a truststore. Separate private key passwords, API key settings, and OAuth settings are not supported. * Upstream declares applicability to Elasticsearch 6.x and 7.x. This does not establish compatibility with Elasticsearch 8.x or 9.x. ## FAQ ### The Task Is Running, but Inserts or Updates Produce No Messages? First confirm that documents are search-visible, then check whether the cursor is missing, duplicated, or at or below the watermark already read. Updates that do not change the cursor are not reread, and a dual cursor is not a late-arrival compensation mechanism. Check Task error logs and actual output rather than relying only on runtime status. Correct cursor generation and write ordering at the source, and design a separate backfill for missed data. Do not expect rereads merely by changing the Topic prefix. ### Why Are New Indices Not Ingested Automatically? Check whether `index.names` is still present: the fixed list takes precedence and does not expand automatically. For dynamic discovery, remove that key, confirm that `index.prefix` is a literal prefix matching actual index names, and check `/_cat/indices` permissions and error logs. An empty string or asterisk is not a substitute for selecting the correct prefix. ### How Do Reads Resume After a Restart, and Why Can Duplicates Occur? Preserve the Connector identity, index names, cursor definition, and Kafka Connect's persisted Offsets to reuse the original read position. Progress that has not been persisted may be reread after a failure. With no saved position, the Connector reads currently visible documents, rather than restoring historical deletions or intermediate versions. Downstream systems should handle duplicates using business identifiers; do not treat the default key containing the cursor as a stable document identifier. ### Why Do Field Names or Schema Differ from the Source Document? Default name conversion removes special characters; for example, the added fields become `esid` and `esindex`. Schema is also derived from actual values in each document, not from the Elasticsearch mapping, so null values, empty arrays, and field type changes affect the result. Check field filtering order and composite field rules, keep business field types consistent, and check whether downstream systems can accept per-record Schema differences. ### How Can Password Disclosure in Logs Be Prevented After Configuring Passwords? All three password settings are string types, and initialization configuration logs may print resolved values. Before deployment, verify and restrict INFO configuration output from the Connector and Task configuration classes, and restrict permissions for reading and retaining logs. Using a Config Provider alone is insufficient to protect resolved values. If disclosure is discovered, immediately revoke or rotate credentials and handle affected logs according to security procedures. Redact troubleshooting materials before submitting them, and do not copy the disclosed values again.