Elasticsearch Sink V2 Connector for Confluent Cloud

The fully-managed Elasticsearch Sink V2 connector for Confluent Cloud moves data from Apache Kafka® to Elasticsearch. The connector supports Avro, JSON Schema, Protobuf, or JSON (schemaless) data output from Apache Kafka® topics. It writes data from a topic in Kafka to an Elasticsearch index. Elasticsearch is often used for text queries, analytics, and as a key-value store.

The connector supports both the analytics and key-value store use cases.

  • Analytics use case: Each message in Kafka is treated as an event. When key.ignore=true, the connector uses topic+partition+offset as a unique identifier for events, which are then converted to unique documents in Elasticsearch.

  • Key-value store use case: When key.ignore=false, the connector uses keys from Kafka messages as document IDs in Elasticsearch, with external versioning from Kafka offsets to ensure updates to a key are written in order. Elasticsearch’s idempotent write semantics guarantees exactly-once delivery.

This Quick Start is for version 2 of fully-managed Confluent Cloud connector. For the earlier version of this connector, see Elasticsearch Service Sink Connector for Confluent Cloud. If migrating from V1 to V2, see Legacy to V2 connector migration.

If you require private networking for fully-managed connectors, make sure to set up the proper networking beforehand. For more information, see Manage Networking for Confluent Cloud Connectors.

Features

The Elasticsearch Sink V2 connector inserts Kafka records into an Elasticsearch index. The connector provides the following features:

  • Authentication: Supports multiple authentication methods - BASIC (username and password, default), API_KEY (Elasticsearch API key-based authentication), or NONE (no authentication).

  • Elasticsearch server version support: Supports Elasticsearch V8 (8.x clusters, default) and V9 (9.x clusters with compatibility headers).

  • Deployment models: The connector supports writing to Elasticsearch running in Elastic Cloud (managed clusters) and Elastic Cloud Serverless projects.

  • Supports SSL/TLS security: The connector supports SSL/TLS encryption, which is automatically used when the connection URL starts with <https://>. For PKI authentication, set elastic.ssl.enabled=true to enable client certificate authentication with keystore and truststore configuration.

  • Input data formats: The connector supports Avro, JSON Schema, Protobuf, or JSON (schemaless) input data formats. Schema Registry must be enabled to use a Schema Registry-based format (for example, Avro, JSON_SR (JSON Schema), or Protobuf). See Schema Registry Enabled Environments for additional information.

  • Resource Types: The connector supports writing to different Elasticsearch resource types like, INDEX (regular indices, default), DATASTREAM (data streams for time-series data), ALIAS_INDEX (pre-created index aliases), or ALIAS_DATASTREAM (pre-created data stream aliases).

  • Alias Support: The connector supports writing to aliases for both indices and data streams. These aliases must be pre-created in Elasticsearch, and the connector does not create or manage aliases on behalf of the user.

  • Supports Automatic resource creation: The connector supports Automatic resource creation. Its behavior for resolving and creating Elasticsearch resources is controlled by the auto.create configuration:

    • When auto.create = true (default): The connector automatically creates Elasticsearch indices or data streams based on the Kafka topic name. No explicit topic-to-resource mapping is required. For data streams, the resource name is constructed as {data.stream.type}-{data.stream.dataset}-{data.stream.namespace}.

    • When auto.create=false: The connector writes only to pre-existing Elasticsearch resources and does not attempt to create indices, data streams, or aliases. In this mode, you must provide a topic.to.resource.mapping configuration to explicitly map each Kafka topic to its target Elasticsearch resource (for example, orders:elasticsearch-orders,users:elasticsearch-users). All referenced resources must exist before the connector starts. Each Kafka topic must map to exactly one Elasticsearch resource; many-to-one or one-to-many mappings are not supported.

  • Supports write methods: The connector supports two write methods. INSERT (default) uses the index operation with external versioning for exactly-once delivery. UPSERT uses the update operation with doc_as_upsert and retry_on_conflict for partial document updates.

  • Data stream support: For time-series data, the connector supports Elasticsearch data streams with automatic @timestamp injection. Configure data.stream.type (LOGS, METRICS, or custom), data.stream.dataset, data.stream.namespace (defaults to ${topic}), and data.stream.timestamp.field (priority list of fields to map to @timestamp).

  • Topic Mutating SMTs support: Topic Mutating SMTs are supported when auto.create property is set to true. When auto.create is set to false with topic.to.resource.mapping, topic mutating SMTs causes connector failures. The connector supports the following SMTs:

    • org.apache.kafka.connect.transforms.TimestampRouter

    • io.confluent.connect.transforms.MessageTimestampRouter

    • io.confluent.connect.transforms.ExtractTopic

    • io.confluent.connect.cloud.transforms.TopicRegexRouter

    For details, see the Single Message Transformations (SMT) documentation.

For more information and examples to use with the Confluent Cloud API for Connect, see the Confluent Cloud API for Connect Usage Examples section.

Limitations

Review the following information:

Quick Start

Use this quick start to get up and running with the Confluent Cloud Elasticsearch Sink V2 connector. The quick start provides the basics of selecting the connector and configuring it to stream events to an Elasticsearch deployment.

Note

The connector only works with the Elasticsearch Service from Elastic Cloud.

Prerequisites

  • Authorized access to a Confluent Cloud cluster on Amazon Web Services (AWS), Microsoft Azure (Azure), or Google Cloud.

  • The Confluent CLI installed and configured for the cluster. See Install the Confluent CLI.

  • Schema Registry must be enabled to use a Schema Registry-based format (for example, Avro, JSON_SR (JSON Schema), or Protobuf). See Schema Registry Enabled Environments for additional information.

  • For networking considerations, see Networking and DNS. To use a set of public egress IP addresses, see Public Egress IP Addresses for Confluent Cloud Connectors.

  • The Elasticsearch Service deployment must be in the same region as your Confluent Cloud deployment.

  • You must configure a valid Elasticsearch authentication method in the connector configuration. Supported authentication types include BASIC authentication (username and password) and API key-based authentication, depending on your Elasticsearch deployment. Authentication credentials are obtained when you create your Elastic deployment.

  • The Elasticsearch Sink V2 connector (user) must have the following privileges:

    • Cluster privileges: monitor, manage, and all.

    • Index privileges: create_index, read, write, and view_index_metadata.

  • Kafka cluster credentials. The following lists the different ways you can provide credentials.

    • Enter an existing service account resource ID.

    • Create a Confluent Cloud service account for the connector. Make sure to review the ACL entries required in the service account documentation. Some connectors have specific ACL requirements.

    • Create a Confluent Cloud API key and secret. To create a key and secret, you can use confluent api-key create or you can autogenerate the API key and secret directly in the Cloud Console when setting up the connector.

Using the Confluent Cloud Console

Step 1: Launch your Confluent Cloud cluster

To create and launch a Kafka cluster in Confluent Cloud, see Create a kafka cluster in Confluent Cloud.

Step 2: Add a connector

In the left navigation menu, click Connectors. If you already have connectors in your cluster, click + Add connector.

Step 3: Select your connector

Click the Elasticsearch Sink V2 connector card.

Elasticsearch Sink V2 Connector Card

Step 4: Enter the connector details

Note

  • Ensure you have all your prerequisites completed.

  • An asterisk ( * ) designates a required entry.

At the Add Elasticsearch Sink V2 Connector screen, complete the following:

If you’ve already populated your Kafka topics, select the topics you want to connect from the Topics list.

To create a new topic, click +Add new topic.

  1. Select the way you want to provide Kafka Cluster credentials. You can choose one of the following options:

    • My account: This setting allows your connector to globally access everything that you have access to. With a user account, the connector uses an API key and secret to access the Kafka cluster. This option is not recommended for production.

    • Service account: This setting limits the access for your connector by using a service account. This option is recommended for production.

    • Use an existing API key: This setting allows you to specify an API key and a secret pair. You can use an existing pair or create a new one. This method is not recommended for production environments.

    Note

    Freight clusters support only service accounts for Kafka authentication.

  2. Click Continue.

  1. Configure the authentication properties:

    • Connection URI: Elasticsearch Service connection URI. For example, https://123123.us-east-1.aws.found.io:9243.

    • Elasticsearch Server Version: The version of your Elasticsearch server. Select V8 for Elasticsearch 8.x clusters or V9 for Elasticsearch 9.x clusters. Default is V8.

    • Authentication Type: The authentication method for connecting to Elasticsearch. Set to BASIC to use username/password authentication, API_KEY to use Elasticsearch API key-based authentication, or NONE for no authentication. Default is BASIC.

    • Connection user: The username used to authenticate with Elasticsearch Service.

    • Connection password: The password used to authenticate with Elasticsearch Service.

    • API Key Value: The Elasticsearch API key value used for authentication. Only required when auth.type is set to API_KEY. You can generate API keys from the Elasticsearch Management interface or using the Elasticsearch API.

    • SSL Enabled: Set to true to enable PKI authentication with SSL support using client certificates. When set to false, all SSL configuration properties below are ignored. Note that the connector will still use SSL/TLS encryption if your connection URL uses https, but without client certificate authentication.

    • SSL Keystore file: The key store file. This is optional for client and can be used for two-way authentication for client.

    • Keystore Key Password: The password of the private key in the key store file. This is required for clients only if two-way authentication is configured.

    • Keystore password: The store password for the key store file. This is optional for client and only needed if ‘ssl.keystore.location’ is configured. Key store password is not supported for PEM format.

    • Keystore file type: The file format of the key store file. This is optional for client.

    • SSL Truststore file: The Truststore file with the certificates of the trusted CAs.

    • Truststore password: The password for the trust store file. If a password is not set, trust store file configured will still be used, but integrity checking is disabled. Trust store password is not supported for PEM format.

    • Truststore type: The file format of the trust store file.

    • Keymanager algorithm: The algorithm used by key manager factory for SSL connections.

    • Trustmanager algorithm: The algorithm used by trust manager factory for SSL connections.

    • SSL Endpoint identification algorithm: The endpoint identification algorithm to validate server hostname using server certificate.

  2. Click Continue.

Note

Configuration properties that are not shown in the Cloud Console use the default values. See Configuration Properties for all property values and definitions.

  • Input Kafka record value format: Sets the input Kafka record value format. Valid entries are AVRO, JSON_SR, PROTOBUF, or JSON. Note that you need to have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, and PROTOBUF.

  • Auto Create Resources: Whether to automatically create Elasticsearch indices or data streams based on topic names. When set to true, the connector automatically creates the necessary Elasticsearch resources (indices or data streams) before writing data. When set to false, the resources must already exist in Elasticsearch or be specified via topic.to.resource.mapping.

  • Resource Type: The type of Elasticsearch resource to write to. Set to INDEX for regular indices, DATASTREAM for data streams, ALIAS_INDEX for index aliases, or ALIAS_DATASTREAM for data stream aliases.

  • Topic to Resource Mapping: A list of topic-to-resource mappings in the format topic:resource (e.g., orders:elasticsearch-orders,users:elasticsearch-users). When specified, the connector uses the provided resource name instead of the topic name for writing to Elasticsearch. These resources must already exist in Elasticsearch. The type of resource is determined by the resource.type configuration.

  • Data Stream Type: Describes the generic type of data to be written to a data stream. The default value is none, indicating that the connector will write to regular indices. If set, this configuration will be used alongside data.stream.dataset to construct the data stream name in the form of {data.stream.type}-{data.stream.dataset}-{data.stream.namespace}. Possible values are logs, metrics, none, and custom index templates defined in the destination cluster are also supported.

  • Data Stream Dataset: Describes the data ingested and its structure to be written to a data stream. This can be any arbitrary string, provided it is no longer than 100 characters, in all lowercase, and does not contain spaces or any special characters /\*"<>|,#:-. If no value is set, the connector writes to regular indices. If set, this configuration will be used alongside data.stream.type to construct the data stream name in the form of {data.stream.type}-{data.stream.dataset}-{data.stream.namespace}.

  • Data Stream Namespace: Generic name describing a user-configurable arbitrary grouping for writing to a data stream. It can be any string up to 100 characters, in lowercase, without spaces or special characters (/*”<>|,#:-). If unset, the connector writes to regular indices. When set, it is used with data.stream.type and data.stream.dataset to form the data stream name in the format {data.stream.type}-{data.stream.dataset}-{data.stream.namespace}. Default is ${topic}, which means the topic name.

  • Data Stream Timestamp Field: All documents sent to a data stream need an @timestamp field with values of type date or date_nanos. Otherwise, the document will not be sent. If multiple fields are provided, the first field listed that also appears in the record will be used. If this configuration is left empty, all of the documents will use the Kafka record timestamp as the @timestamp field value. Note that @timestamp still needs to be explicitly listed if records already contain this field.

Show advanced configurations

  • Schema context: Select a schema context to use for this connector, if using a schema-based data format. This property defaults to the Default context, which configures the connector to use the default schema set up for Schema Registry in your Confluent Cloud environment. A schema context allows you to use separate schemas (like schema sub-registries) tied to topics in different Kafka clusters that share the same Schema Registry environment. For example, if you select a non-default context, a Source connector uses only that schema context to register a schema and a Sink connector uses only that schema context to read from. For more information about setting up a schema context, see What are schema contexts and when should you use them?.

  • Key ignore: Whether to ignore the record key for the purpose of forming the Elasticsearch document ID. When this is set to true, document IDs will be generated as topic+partition+offset taken from the record. When this is set to false, the record key will be used as the Elasticsearch document ID.

  • Topics for ‘Ignore Key’ mode: List of topics for which key.ignore should be true.

  • Schema ignore: Whether to ignore schemas during indexing. When this is set to true, the record schema will be ignored for the purpose of registering an Elasticsearch mapping. Elasticsearch will infer the mapping from the data (dynamic mapping needs to be enabled by the user).

  • Topics for ‘Ignore Schema’ mode: List of topics for which schema.ignore should be true.

  • Compact Map Entries: Defines how map entries with string keys within record values should be written to JSON. When set to true, these entries are written compactly as entryKey: entryValue. When set to false, map entries with string keys are written as a nested document {key: entryKey, value: entryValue}.

  • Write Method: Method used for writing data to Elasticsearch, and one of INSERT or UPSERT. The default method is INSERT, in which the connector constructs a document from the record value and inserts that document into Elasticsearch, completely replacing any existing document with the same ID; this matches previous behavior. The UPSERT method will create a new document if one with the specified ID does not yet exist, or will update an existing document with the same ID by adding/replacing only those fields present in the record value. The UPSERT method may require additional time and resources of Elasticsearch, so consider increasing the read.timeout.ms and decreasing the batch.size configuration properties.

  • Behavior for null-valued records: How to handle records with a non-null key and a null value (i.e., Kafka tombstone records). Valid options are IGNORE, DELETE, and FAIL. IGNORE skips the record. DELETE removes the corresponding document from Elasticsearch. FAIL causes the connector to fail.

  • Behavior on malformed documents: How to handle records that Elasticsearch rejects due to some malformation of the document itself, such as an index mapping conflict, a field name containing illegal characters, or a record with a missing id. ‘ignore’ will skip the bad records and ‘fail’ will fail the connector.

  • Batch Size: The number of records to process as a batch when writing to Elasticsearch using the bulk API.

  • Connection compression: Whether to use GZip compression on HTTP connection to ElasticSearch. To make this setting to work the http.compression setting also needs to be enabled at the Elasticsearch nodes before using it.

  • Read Timeout: How long to wait in milliseconds for the Elasticsearch server to send a response. The task fails if any read operation times out.

  • External Version Header Name: Header name to pull value for external versioning, defaults to using the kafka record offset. Must have a numeric value.

  • Elasticsearch Generated IDs: Specifies whether to use auto-generated Elasticsearch document IDs for insertion requests. Note that this setting removes exactly once guarantees, and message delivery will be at least once. This only applies if the write method is set to INSERT. When set to true, the Ignore Key mode option will also be ignored when sending data to Elasticsearch.

Additional Configs

  • Value Converter Replace Null With Default: Whether to replace fields that have a default value and that are null to the default value. When set to true, the default value is used, otherwise null is used. Applicable for JSON Converter.

  • Value Converter Schema ID Deserializer: The class name of the schema ID deserializer for values. This is used to deserialize schema IDs from the message headers.

  • Value Converter Reference Subject Name Strategy: Set the subject reference name strategy for value. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being DefaultReferenceSubjectNameStrategy.

  • Schema ID For Value Converter: The schema ID to use for deserialization when using ConfigSchemaIdDeserializer. This is used to specify a fixed schema ID to be used for deserializing message values. Only applicable when value.converter.value.schema.id.deserializer is set to ConfigSchemaIdDeserializer.

  • Value Converter Schemas Enable: Include schemas within each of the serialized values. Input messages must contain schema and payload fields and may not contain additional fields. For plain JSON data, set this to false. Applicable for JSON Converter.

  • Errors Tolerance: Use this property if you would like to configure the connector’s error handling behavior. WARNING: This property should be used with CAUTION for SOURCE CONNECTORS as it may lead to dataloss. If you set this property to ‘all’, the connector will not fail on errant records, but will instead log them (and send to DLQ for Sink Connectors) and continue processing. If you set this property to ‘none’, the connector task will fail on errant records.

  • Value Converter Ignore Default For Nullables: When set to true, this property ensures that the corresponding record in Kafka is NULL, instead of showing the default column value. Applicable for AVRO,PROTOBUF and JSON_SR Converters.

  • Key Converter Schema ID Deserializer: The class name of the schema ID deserializer for keys. This is used to deserialize schema IDs from the message headers.

  • Value Converter Decimal Format: Specify the JSON/JSON_SR serialization format for Connect DECIMAL logical type values with two allowed literals: BASE64 to serialize DECIMAL logical types as base64 encoded binary data and NUMERIC to serialize Connect DECIMAL logical type values in JSON/JSON_SR as a number representing the decimal value.

  • Schema GUID For Key Converter: The schema GUID to use for deserialization when using ConfigSchemaIdDeserializer. This is used to specify a fixed schema GUID to be used for deserializing message keys. Only applicable when key.converter.key.schema.id.deserializer is set to ConfigSchemaIdDeserializer.

  • Schema GUID For Value Converter: The schema GUID to use for deserialization when using ConfigSchemaIdDeserializer. This is used to specify a fixed schema GUID to be used for deserializing message values. Only applicable when value.converter.value.schema.id.deserializer is set to ConfigSchemaIdDeserializer.

  • Value Converter Connect Meta Data: Allow the Connect converter to add its metadata to the output schema. Applicable for Avro Converters.

  • Value Converter Value Subject Name Strategy: Determines how to construct the subject name under which the value schema is registered with Schema Registry.

  • Key Converter Key Subject Name Strategy: How to construct the subject name for key schema registration.

  • Schema ID For Key Converter: The schema ID to use for deserialization when using ConfigSchemaIdDeserializer. This is used to specify a fixed schema ID to be used for deserializing message keys. Only applicable when key.converter.key.schema.id.deserializer is set to ConfigSchemaIdDeserializer.

Auto-restart policy

  • Enable Connector Auto-restart: Control the auto-restart behavior of the connector and its task in the event of user-actionable errors. Defaults to true, enabling the connector to automatically restart in case of user-actionable errors. Set this property to false to disable auto-restart for failed connectors. In such cases, you would need to manually restart the connector.

Consumer configuration

  • Max poll interval(ms): Set the maximum delay between subsequent consume requests to Kafka. Use this property to improve connector performance in cases when the connector cannot send records to the sink system. The default is 300,000 milliseconds (5 minutes).

  • Max poll records: Set the maximum number of records to consume from Kafka in a single request. Use this property to improve connector performance in cases when the connector cannot send records to the sink system. The default is 500 records.

Transforms

Processing position

  • Set offsets: Click Set offsets to define a specific offset for this connector to begin procession data from. For more information on managing offsets, see Manage offsets.

See Configuration Properties for all property values and definitions.

  • Click Continue.

Based on the number of topic partitions you select, you will be provided with a recommended number of tasks.

  1. To change the number of recommended tasks, enter the number of tasks for the connector to use in the Tasks field.

  2. Click Continue.

  1. Verify the connection details.

  2. Click Launch.

    The status for the connector should go from Provisioning to Running.

Step 5: Check the results in Elasticsearch

Verify that new records are being added to your Elasticsearch deployment.

For more information and examples to use with the Confluent Cloud API for Connect, see the Confluent Cloud API for Connect Usage Examples section.

Tip

When you launch a connector, a Dead Letter Queue topic is automatically created. See View Connector Dead Letter Queue Errors in Confluent Cloud for details.

Using the Confluent CLI

Complete the following steps to set up and run the connector using the Confluent CLI.

Note

Make sure you have all your prerequisites completed.

Step 1: List the available connectors

Enter the following command to list available connectors:

confluent connect plugin list

Step 2: List the connector configuration properties

Enter the following command to show the connector configuration properties:

confluent connect plugin describe <connector-plugin-name>

The command output shows the required and optional configuration properties.

Step 3: Create the connector configuration file

Create a JSON file that contains the connector configuration properties. The following example shows the required connector properties.

{
  "connector.class": "ElasticsearchSinkV2",
  "name": "elasticsearch-v2-connector",
  "kafka.auth.mode": "KAFKA_API_KEY",
  "kafka.api.key": "<my-kafka-api-key>",
  "kafka.api.secret": "<my-kafka-api-secret>",
  "topics": "<topic1>, <topic2>",
  "input.data.format": "JSON",
  "connection.url": "<elasticsearch-URI>",
  "auth.type": "BASIC",
  "connection.username": "<elasticsearch-username>",
  "connection.password": "<elasticsearch-password>",
  "resource.type": "INDEX",
  "auto.create": "true",
  "key.ignore": "true",
  "schema.ignore": "true",
  "tasks.max": "1"
}

Note the following property definitions:

  • "connector.class": Identifies the connector plugin name.

  • "name": Sets a name for your new connector.

  • "kafka.auth.mode": Identifies the connector authentication mode you want to use. There are two options: SERVICE_ACCOUNT or KAFKA_API_KEY (the default). To use an API key and secret, specify the configuration properties kafka.api.key and kafka.api.secret, as shown in the example configuration (above). To use a service account, specify the Resource ID in the property kafka.service.account.id=<service-account-resource-ID>. To list the available service account resource IDs, use the following command:

    confluent iam service-account list

    For example:

    confluent iam service-account list

   Id     | Resource ID |       Name        |    Description
+---------+-------------+-------------------+-------------------
   123456 | sa-l1r23m   | sa-1              | Service account 1
   789101 | sa-l4d56p   | sa-2              | Service account 2

  • "input.data.format": Sets the input Kafka record value format (data coming from the Kafka topic). Valid entries are AVRO, JSON_SR, PROTOBUF, or JSON. You must have Confluent Cloud Schema Registry configured if using a schema-based message format (for example, Avro, JSON_SR (JSON Schema), or Protobuf).

  • "connection.url": Enter the connection URI. This is the Elasticsearch endpoint you can copy from your Elasticsearch deployment console. The URI you enter should look like this: https://ec5bfac80bc14c26a77eefb6585f196c.us-west-2.aws.found.io:9243.

  • "connection.username" and "connection.password" Enter the Elasticsearch deployment username and password. An example showing where these are on the Elastic deployment console is shown in the prerequisites.

The following are optional properties you can include in the configuration:

  • key.ignore: Whether to ignore the record key for the purpose of forming the Elasticsearch document ID. When this is set to true, document IDs are created from the topic name, partition, and offset (i.e., topic+partition+offset). Defaults to false if not used.

  • topic.key.ignore: List of topics for which key.ignore should be true. Unlike global key.ignore=true, this preserves external versioning while using topic+partition+offset as document IDs.

  • schema.ignore: Whether to ignore schemas during indexing. When this property is set to true, the record schema is ignored and Elasticsearch infers the mapping from the data. For this to work, Elasticsearch dynamic mapping must be enabled. Defaults to false.

  • topic.schema.ignore: List of topics for which schema.ignore should be true.

  • compact.map.entries: Defines how map entries with string keys in record values should be written to JSON. When this property is set to true, the entries are written compactly as "entryKey": "entryValue". Otherwise, map entries with string keys are written as a nested document ({"key": "entryKey", "value": "entryValue"}). Defaults to true if not used.

  • behavior.on.null.values: How to handle records with a non-null key and a null value (i.e., Kafka tombstone records). Valid options are IGNORE, DELETE, and FAIL. Defaults to IGNORE if not used.

  • behavior.on.malformed.documents: How to handle records that Elasticsearch rejects due to some malformation of the document itself, such as an index mapping conflict or a field name containing illegal characters. Valid options are ignore (skip bad records, send to DLQ) and fail (fail the connector). Defaults to fail.

  • batch.size: The number of records to process as a batch when writing to Elasticsearch. This value defaults to 50 if not used.

  • connection.compression: Whether to use Gzip compression on the HTTP connection to Elasticsearch. To make this setting work the http.compression configuration must be set to true on the Elasticsearch nodes. For more information about the Elasticsearch HTTP properties, see Elasticsearch HTTP Settings. Defaults to false.

  • read.timeout.ms: How long to wait in milliseconds for the Elasticsearch server to send a response. The task fails if any read operation times out. This value defaults to 15000 ms (15 seconds).

  • use.autogenerated.ids: Specifies whether to use auto-generated Elasticsearch document IDs for insertion requests. Note that this configuration removes exactly once guarantees, and message delivery will be at least once. This only applies if the write method is set to INSERT. Defaults to false.

  • external.version.header: Header name to pull value for external versioning, defaults to using the Kafka record offset. Must have a numeric value.

Single Message Transforms: See the Single Message Transforms (SMT) documentation for details about adding SMTs using the CLI. See Unsupported transformations for a list of SMTs that are not supported with this connector.

See Configuration Properties for all property values and definitions.

Step 4: Load the configuration file and create the connector

Enter the following command to load the configuration and start the connector:

confluent connect cluster create --config-file <file-name>.json

For example:

confluent connect cluster create --config-file elasticsearch-sink-v2-config.json

Example output:

Created connector elasticsearch-v2-connector lcc-ix4dl

Step 5: Check the connector status

Enter the following command to check the connector status:

confluent connect cluster list

Example output:

ID          |       Name                 | Status  | Type
+-----------+----------------------------+---------+------+
lcc-ix4dl   | elasticsearch-v2-connector | RUNNING | sink

Step 6: Check the results in Elasticsearch.

Verify that new records are being added to the Elasticsearch deployment.

For more information and examples to use with the Confluent Cloud API for Connect, see the Confluent Cloud API for Connect Usage Examples section.

Tip

When you launch a connector, a Dead Letter Queue topic is automatically created. See View Connector Dead Letter Queue Errors in Confluent Cloud for details.

Legacy to V2 connector migration

The Elasticsearch Service Sink legacy (V1) connector is being replaced by the Elasticsearch Sink V2 connector. The Elasticsearch Sink V2 connector is built on a modern HTTP framework, offering improved reliability, better error handling, and a cleaner configuration model compared to the legacy connector.

Due to fundamental differences in the underlying architecture and configuration properties, a direct in-place upgrade is not possible. The following sections provide a safe migration strategy to ensure a transition with no data loss.

Prerequisites

  • Confluent Cloud: An active account with access to the legacy connector.

  • Elasticsearch: Access to an Elasticsearch cluster (v8 or v9 supported).

  • Permissions: Necessary credentials (Basic Authentication or an API key) for the target Elasticsearch instance.

Migration steps

Follow these steps to migrate your pipelines from the legacy connector to the V2 connector.

  1. Understand the key changes: Review the differences in behavior, particularly regarding document ID logic, error handling, and resource creation. See Changed connector behavior for a detailed comparison.

  2. Set up a parallel pipeline: Never attempt to upgrade a production connector in place.

    • Create a new Elasticsearch Sink V2 connector in a test environment, pointing it to a new, separate Elasticsearch index or data stream for testing.

    • Use the same Kafka topic as your production connector to compare data flow directly.

  3. Validate data integrity: Run the legacy and V2 connectors in parallel, writing to separate indices. Compare the data in both resources to ensure consistency.

    • Verify that DELETE operations and tombstone records are handled as expected.

    • Ensure that SSL/TLS and authentication are correctly configured.

  4. Cutover to V2: After validating the V2 pipeline, perform the following:

    • Stop the legacy Elasticsearch Service Sink connector.

    • Start or resume the new V2 connector from the last committed offsets of the V1 connector to prevent data gaps or duplicates.

    • After confirming the V2 connector is running stably, decommission the legacy connector and update any monitoring and alerting systems to track V2-specific metrics.

Key differences & breaking changes

The following sections detail the configuration and behavioral changes between the Elasticsearch legacy (V1) and V2 connectors.

New V2 properties

These properties provide functionality or structure that was not available in the legacy version.

V2 Property

Default Value

Description

elastic.server.version

V8

Elasticsearch server version. Valid values: V8, V9.

auth.type

BASIC

Authentication method. Valid values: NONE, BASIC, API_KEY.

api.key.value

(None)

Elasticsearch API key (used when auth.type=API_KEY).

elastic.ssl.enabled

false

Replaces elastic.security.protocol. Boolean to enable SSL.

auto.create

true

Explicitly enables/disables auto-creation of resources.

resource.type

INDEX

Defines the target resource: INDEX, DATASTREAM, ALIAS_INDEX, or ALIAS_DATASTREAM.

Discontinued properties

The following legacy configuration properties have been removed in V2.

Discontinued V1 Property

V2 Behavior & Equivalent

drop.invalid.message

Not supported. V2 fails the task on preprocessing errors due to framework limitations.

linger.ms

Removed. V2 uses framework-level batching.

flush.timeout.ms

Removed. V2 uses framework-level timeout handling.

external.resource.usage

Separated into auto.create and resource.type.

topic.to.external.resource.mapping

Renamed to topic.to.resource.mapping for consistency.

Properties with changed behavior

The following configuration properties exist in both versions but have different implementations in V2.

Property

V1 (Legacy) Behavior

V2 Behavior

connection.url

Supported multiple URLs in a list.

Supports only a single URL string.

key.ignore

Type: STRING

Type: BOOLEAN

data.stream.type

Default: NONE (signaled index creation).

Default: LOGS. NONE is no longer used.

topic.key.ignore

Topics in list disabled external versioning.

Topics in list enable external versioning.

Changed connector behavior

Resource and mapping creation

In V1, resource and mapping creation were independent flows. V2 creates resources and mappings together only when auto.create=true.

  • If the resource already exists, V2 assumes mappings are present and skips creation to avoid conflicts.

  • If auto.create=false, the connector expects users to pre-create the resource and mappings.

Document ID and DELETE Logic

There are significant changes to how DELETE records (tombstones) are handled:

  • V1 Behavior: When behavior.on.null.records=DELETE, the key.ignore setting was not consistently honored if auto.generated.ids=true. The connector often implicitly used topic+partition+offset as the document _id.

  • V2 Behavior: DELETE handling strictly depends on the key.ignore configuration.

    • If key.ignore=false: The record key must be present or the task will fail.

    • If key.ignore=true: The connector derives the _id using topic+partition+offset for DELETE operations.

  • Note: Unlike V1, auto.generated.ids in V2 applies only to INSERT operations.

Error routing

  • V1: Connector-level and record-level errors were routed to a Dead Letter Queue (DLQ).

  • V2: All errors are routed to the error topic, aligning with the Confluent Cloud standardized error-handling model.

Batch size defaults

The default batch.size has changed from 2000 (V1) to 50 (V2).

External versioning

  • V1: Disabled if the document _id was not derived from the record key.

  • V2: Behavior is determined by key.ignore. If key.ignore=false, external versioning is enabled for all topics. Notably, for topics in topic.key.ignore, external versioning remains enabled in V2 even when using t+p+o for the _id.

Success topic behavior

V2 automatically provisions a success topic for framework consistency, but it remains empty as the connector does not emit success events.

Migration FAQs

The frequently asked questions (FAQs) address common questions and issues encountered while migrating from the legacy to Elasticsearch Sink V2 connector for Confluent Cloud.

Why should I migrate to the V2 connector?

The V2 connector provides significant technical and operational benefits:

  • Elasticsearch 9.x & 8.x Support: Provides native compatibility with the latest Elasticsearch server versions.

  • Serverless Support: Architected specifically to support modern serverless Elasticsearch environments.

  • Secure Authentication: Introduces support for API_KEY and NONE authentication types in addition to standard BASIC authentication.

  • Modern Resource Mapping: Uses a consistent topic.to.resource.mapping model to map Kafka topics directly to indices, aliases, or data streams.

What happens if I don’t migrate before the legacy connector End of life (EOL)?

Elasticsearch 7.x has been officially deprecated by Elastic and is scheduled to go out of service this year. In alignment with this, the Elasticsearch Sink V1 connector is scheduled for EOL in 2027.

After the EOL date:

  • No New Connectors: You can no longer create new legacy Elasticsearch V1 connectors.

  • Lack of Support: Existing connectors may continue to function, but they will not receive updates, security patches, or bug fixes. To ensure long-term stability, migrate to the Elasticsearch Sink V2 connector as soon as possible.

Are these two connectors just different versions of the same software?

No. They are two completely different connectors from an architectural standpoint. While the legacy V1 connector was built using the Elasticsearch SDK, the V2 connector was written from the ground up to leverage a modern, more efficient API. This fundamental difference in technology is why a direct, in-place upgrade is not possible and requires a parallel migration strategy.

Is there a price difference between V1 and V2?

No. Confluent prices both the legacy and V2 connectors the same on Confluent Cloud.

Do I need to stop my legacy connector before starting V2?

No. During the validation phase, you should run them in parallel. However, once you cut over to the V2 connector for production traffic, you must stop the legacy connector to prevent duplicate data writes.

Will my old offsets work with the V2 connector?

Yes. Offsets from the legacy connector are compatible and can be reused by the V2 connector to ensure continuity and avoid data gaps.

What happened to the drop.invalid.message setting?

Due to framework level constraints in the HTTP V2 pipeline, the connector cannot selectively drop records once formatting has started. Preprocessing failures will now result in a task failure to ensure data integrity.

Why were the key.ignore and schema.ignore types changed to Boolean?

V1 allowed any string input, which often led to configuration errors. V2 uses Booleans to provide a better UI experience (dropdowns) and to enforce strict validation.

Will V2 be available for self-managed Confluent Platform?

At this time, there are no plans to release a self-managed V2 connector. If you require this capability, submit a Feature Request through your Confluent account team to help us track and prioritize interest.

Configuration Properties

Use the following configuration properties with the fully-managed connector.

Which topics do you want to get data from?

topics.regex

A regular expression that matches the names of the topics to consume from. This is useful when you want to consume from multiple topics that match a certain pattern without having to list them all individually.

  • Type: string

  • Importance: low

topics

Identifies the topic name or a comma-separated list of topic names.

  • Type: list

  • Importance: high

errors.deadletterqueue.topic.name

The name of the topic to be used as the dead letter queue (DLQ) for messages that result in an error when processed by this sink connector, or its transformations or converters. Defaults to ‘dlq-${connector}’ if not set. The DLQ topic will be created automatically if it does not exist. You can provide ${connector} in the value to use it as a placeholder for the logical cluster ID.

  • Type: string

  • Default: dlq-${connector}

  • Importance: low

reporter.result.topic.name

The name of the topic to produce records to after successfully processing a sink record. Defaults to ‘success-${connector}’ if not set. You can provide ${connector} in the value to use it as a placeholder for the logical cluster ID.

  • Type: string

  • Default: success-${connector}

  • Importance: low

reporter.error.topic.name

The name of the topic to produce records to after each unsuccessful record sink attempt. Defaults to ‘error-${connector}’ if not set. You can provide ${connector} in the value to use it as a placeholder for the logical cluster ID.

  • Type: string

  • Default: error-${connector}

  • Importance: low

Schema Config

schema.context.name

Add a schema context name. A schema context represents an independent scope in Schema Registry. It is a separate sub-schema tied to topics in different Kafka clusters that share the same Schema Registry instance. If not used, the connector uses the default schema configured for Schema Registry in your Confluent Cloud environment.

  • Type: string

  • Default: default

  • Importance: medium

Input messages

input.data.format

Sets the input Kafka record value format. Valid entries are AVRO, JSON_SR, PROTOBUF, or JSON. Note that you need to have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, and PROTOBUF.

  • Type: string

  • Default: JSON

  • Importance: high

How should we connect to your data?

name

Sets a name for your connector.

  • Type: string

  • Valid Values: A string at most 64 characters long

  • Importance: high

Kafka Cluster credentials

kafka.auth.mode

Kafka Authentication mode. It can be one of KAFKA_API_KEY or SERVICE_ACCOUNT. It defaults to KAFKA_API_KEY mode, whenever possible.

  • Type: string

  • Valid Values: SERVICE_ACCOUNT, KAFKA_API_KEY

  • Importance: high

kafka.api.key

Kafka API Key. Required when kafka.auth.mode==KAFKA_API_KEY.

  • Type: password

  • Importance: high

kafka.service.account.id

The Service Account that will be used to generate the API keys to communicate with Kafka Cluster.

  • Type: string

  • Importance: high

kafka.api.secret

Secret associated with Kafka API key. Required when kafka.auth.mode==KAFKA_API_KEY.

  • Type: password

  • Importance: high

Consumer configuration

max.poll.interval.ms

The maximum delay between subsequent consume requests to Kafka. This configuration property may be used to improve the performance of the connector, if the connector cannot send records to the sink system. Defaults to 300000 milliseconds (5 minutes).

  • Type: long

  • Default: 300000 (5 minutes)

  • Valid Values: [60000,…,1800000] for non-dedicated clusters and [60000,…] for dedicated clusters

  • Importance: low

max.poll.records

The maximum number of records to consume from Kafka in a single request. This configuration property may be used to improve the performance of the connector, if the connector cannot send records to the sink system. Defaults to 500 records.

  • Type: long

  • Default: 500

  • Valid Values: [1,…,500] for non-dedicated clusters and [1,…] for dedicated clusters

  • Importance: low

Number of tasks for this connector

tasks.max

Maximum number of tasks for the connector.

  • Type: int

  • Valid Values: [1,…]

  • Importance: high

How should we connect to your Elasticsearch Service?

connection.url

Elasticsearch Service connection URI (e.g. https://123123.us-east-1.aws.found.io:9243).

  • Type: string

  • Importance: high

elastic.server.version

The version of your Elasticsearch server. Select V8 for Elasticsearch 8.x clusters or V9 for Elasticsearch 9.x clusters. Default is V8.

  • Type: string

  • Default: V8

  • Importance: high

auth.type

The authentication method for connecting to Elasticsearch. Set to BASIC to use username/password authentication, API_KEY to use Elasticsearch API key-based authentication, or NONE for no authentication. Default is BASIC.

  • Type: string

  • Default: BASIC

  • Importance: high

connection.username

The username used to authenticate with Elasticsearch Service.

  • Type: string

  • Importance: medium

connection.password

The password used to authenticate with Elasticsearch Service.

  • Type: password

  • Importance: medium

api.key.value

The Elasticsearch API key value used for authentication. Only required when auth.type is set to API_KEY. You can generate API keys from the Elasticsearch Management interface or using the Elasticsearch API.

  • Type: password

  • Importance: medium

Security

elastic.ssl.enabled

Set to true to enable PKI authentication with SSL support using client certificates. When set to false, all SSL configuration properties below are ignored. Note that the connector will still use SSL/TLS encryption if your connection URL uses https, but without client certificate authentication.

  • Type: boolean

  • Default: false

  • Importance: medium

elastic.https.ssl.keystore.file

The key store file. This is optional for client and can be used for two-way authentication for client.

  • Type: password

  • Importance: medium

elastic.https.ssl.key.password

The password of the private key in the key store file. This is required for clients only if two-way authentication is configured.

  • Type: password

  • Importance: medium

elastic.https.ssl.keystore.password

The store password for the key store file. This is optional for client and only needed if ssl.keystore.location is configured. Key store password is not supported for PEM format.

  • Type: password

  • Importance: medium

elastic.https.ssl.keystore.type

The file format of the key store file. This is optional for client.

  • Type: string

  • Default: JKS

  • Importance: medium

elastic.https.ssl.truststore.file

The Truststore file with the certificates of the trusted CAs.

  • Type: password

  • Importance: medium

elastic.https.ssl.truststore.password

The password for the trust store file. If a password is not set, trust store file configured will still be used, but integrity checking is disabled. Trust store password is not supported for .PEM format.

  • Type: password

  • Importance: medium

elastic.https.ssl.truststore.type

The file format of the trust store file.

  • Type: string

  • Default: JKS

  • Importance: medium

elastic.https.ssl.keymanager.algorithm

The algorithm used by key manager factory for SSL connections.

  • Type: string

  • Default: SunX509

  • Importance: low

elastic.https.ssl.trustmanager.algorithm

The algorithm used by trust manager factory for SSL connections.

  • Type: string

  • Default: PKIX

  • Importance: low

elastic.https.ssl.endpoint.identification.algorithm

The endpoint identification algorithm to validate server hostname using server certificate.

  • Type: string

  • Default: https

  • Importance: low

Destination Configuration

auto.create

Whether to automatically create Elasticsearch indices or data streams based on topic names. When set to true, the connector automatically creates the necessary Elasticsearch resources (indices or data streams) before writing data. When set to false, the resources must already exist in Elasticsearch or be specified via topic.to.resource.mapping.

  • Type: boolean

  • Default: true

  • Importance: high

resource.type

The type of Elasticsearch resource to write to. Set to INDEX for regular indices, DATASTREAM for data streams, ALIAS_INDEX for index aliases, or ALIAS_DATASTREAM for data stream aliases.

  • Type: string

  • Default: INDEX

  • Importance: high

topic.to.resource.mapping

A list of topic-to-resource mappings in the format topic:resource (e.g., orders:elasticsearch-orders,users:elasticsearch-users). When specified, the connector uses the provided resource name instead of the topic name for writing to Elasticsearch. These resources must already exist in Elasticsearch. The type of resource is determined by the resource.type configuration.

  • Type: list

  • Default: “”

  • Importance: high

Data Streams

data.stream.type

Describes the generic type of data to be written to a data stream. Valid options are LOGS for log data or METRICS for metric/time-series data. This value is used alongside data.stream.dataset and data.stream.namespace to construct the data stream name in the format {data.stream.type}-{data.stream.dataset}-{data.stream.namespace}. Custom index templates defined in the destination cluster are also supported.

  • Type: string

  • Default: LOGS

  • Importance: low

data.stream.dataset

Describes the data ingested and its structure to be written to a data stream. This can be any arbitrary string, provided it is no longer than 100 characters, in all lowercase, and does not contain spaces or special characters /\*\"<>|,#:-. When set, this value is used alongside data.stream.type and data.stream.namespace to construct the data stream name in the format {data.stream.type}-{data.stream.dataset}-{data.stream.namespace}.

  • Type: string

  • Default: “”

  • Importance: low

data.stream.namespace

Describes a user-configurable arbitrary grouping for organizing data within a data stream. It can be any string up to 100 characters, in lowercase, without spaces or special characters /\*\"<>|,#:-. When set, it is used with data.stream.type and data.stream.dataset to form the data stream name in the format {data.stream.type}-{data.stream.dataset}-{data.stream.namespace}. Defaults to ${topic}, which resolves to the Kafka topic name.

  • Type: string

  • Default: ${topic}

  • Importance: low

data.stream.timestamp.field

All documents sent to a data stream need an @timestamp field with values of type date or date_nanos. List the field names to use for the @timestamp mapping. If multiple fields are provided, the first field listed that also appears in the record is used. If this configuration is left empty, all documents will use the Kafka record timestamp as the @timestamp field value. Note that @timestamp still needs to be explicitly listed if records already contain this field.

  • Type: list

  • Default: “”

  • Importance: low

Data Conversion

write.method

Method used for writing data to Elasticsearch. The default is INSERT, in which the connector constructs a document from the record value and inserts that document into Elasticsearch, completely replacing any existing document with the same ID. The UPSERT method creates a new document if one with the specified ID does not yet exist, or updates an existing document with the same ID by adding or replacing only those fields present in the record value. The UPSERT method may require additional time and resources of Elasticsearch, so consider increasing read.timeout.ms and decreasing batch.size if you experience timeouts.

  • Type: string

  • Default: INSERT

  • Importance: medium

key.ignore

Whether to ignore the record key for the purpose of forming the Elasticsearch document ID. When this is set to true, document IDs are generated as topic+partition+offset taken from the record. When set to false, the record key is used as the Elasticsearch document ID.

  • Type: boolean

  • Default: false

  • Importance: high

topic.key.ignore

List of topics for which key.ignore should be true.

  • Type: list

  • Default: “”

  • Importance: low

use.autogenerated.ids

Specifies whether to use auto-generated Elasticsearch document IDs for insertion requests. Note that this setting removes exactly once guarantees, and message delivery will be at least once. This only applies if the write method is set to INSERT. When set to true, the Ignore Key mode option is also ignored when sending data to Elasticsearch.

  • Type: boolean

  • Default: false

  • Importance: medium

schema.ignore

Whether to ignore schemas during indexing. When set to true, the record schema is ignored for the purpose of registering an Elasticsearch mapping. Elasticsearch infers the mapping from the data (dynamic mapping needs to be enabled by the user).

  • Type: boolean

  • Default: false

  • Importance: low

topic.schema.ignore

List of topics for which schema.ignore should be true.

  • Type: list

  • Default: “”

  • Importance: low

compact.map.entries

Defines how map entries with string keys within record values should be written to JSON. When set to true, these entries are written compactly as entryKey: entryValue. When set to false, map entries with string keys are written as a nested document {key: entryKey, value: entryValue}.

  • Type: boolean

  • Default: true

  • Importance: low

external.version.header

Header name to pull value for external versioning, defaults to using the kafka record offset. Must have a numeric value.

  • Type: string

  • Default: “”

  • Importance: low

Error Handling

behavior.on.null.values

How to handle records with a non-null key and a null value (i.e., Kafka tombstone records). Valid options are IGNORE, DELETE, and FAIL. IGNORE skips the record. DELETE removes the corresponding document from Elasticsearch. FAIL causes the connector to fail.

  • Type: string

  • Default: IGNORE

  • Importance: low

behavior.on.malformed.documents

How to handle records that Elasticsearch rejects due to some malformation of the document itself, such as an index mapping conflict, a field name containing illegal characters, or a record with a missing id. IGNORE: skip the bad records and continue processing. FAIL: fail the connector task.

  • Type: string

  • Default: FAIL

  • Importance: low

Connection Details

batch.size

The number of records to process as a batch when writing to Elasticsearch using the bulk API.

  • Type: int

  • Default: 50

  • Valid Values: [1,…]

  • Importance: medium

read.timeout.ms

How long to wait in milliseconds for the Elasticsearch server to send a response. The connector task fails if any read operation times out.

  • Type: int

  • Default: 15000 (15 seconds)

  • Valid Values: [1000,…,60000]

  • Importance: low

connection.compression

Whether to use GZip compression on the HTTP connection to Elasticsearch. To enable this, the http.compression setting must also be enabled on your Elasticsearch cluster.

  • Type: boolean

  • Default: false

  • Importance: low

Additional Configs

consumer.override.auto.offset.reset

Defines the behavior of the consumer when there is no committed position (which occurs when the group is first initialized) or when an offset is out of range. You can choose either to reset the position to the “earliest” offset (the default) or the “latest” offset. You can also select “none” if you would rather set the initial offset yourself and you are willing to handle out of range errors manually. More details: https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#auto-offset-reset

  • Type: string

  • Importance: low

consumer.override.isolation.level

Controls how to read messages written transactionally. If set to read_committed, consumer.poll() will only return transactional messages which have been committed. If set to read_uncommitted (the default), consumer.poll() will return all messages, even transactional messages which have been aborted. Non-transactional messages will be returned unconditionally in either mode. More details: https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#isolation-level

  • Type: string

  • Importance: low

header.converter

The converter class for the headers. This is used to serialize and deserialize the headers of the messages.

  • Type: string

  • Importance: low

key.converter.use.schema.guid

The schema GUID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema GUID to be used for deserializing message keys. Only applicable when key.converter.key.schema.id.deserializer is set to ConfigSchemaIdDeserializer.

  • Type: string

  • Importance: low

key.converter.use.schema.id

The schema ID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema ID to be used for deserializing message keys. Only applicable when key.converter.key.schema.id.deserializer is set to ConfigSchemaIdDeserializer.

  • Type: int

  • Importance: low

value.converter.allow.optional.map.keys

Allow optional string map key when converting from Connect Schema to Avro Schema. Applicable for Avro Converters.

  • Type: boolean

  • Importance: low

value.converter.auto.register.schemas

Specify if the Serializer should attempt to register the Schema.

  • Type: boolean

  • Importance: low

value.converter.connect.meta.data

Allow the Connect converter to add its metadata to the output schema. Applicable for Avro Converters.

  • Type: boolean

  • Importance: low

value.converter.enhanced.avro.schema.support

Enable enhanced schema support to preserve package information and Enums. Applicable for Avro Converters.

  • Type: boolean

  • Importance: low

value.converter.enhanced.protobuf.schema.support

Enable enhanced schema support to preserve package information. Applicable for Protobuf Converters.

  • Type: boolean

  • Importance: low

value.converter.flatten.unions

Whether to flatten unions (oneofs). Applicable for Protobuf Converters.

  • Type: boolean

  • Importance: low

value.converter.generate.index.for.unions

Whether to generate an index suffix for unions. Applicable for Protobuf Converters.

  • Type: boolean

  • Importance: low

value.converter.generate.struct.for.nulls

Whether to generate a struct variable for null values. Applicable for Protobuf Converters.

  • Type: boolean

  • Importance: low

value.converter.int.for.enums

Whether to represent enums as integers. Applicable for Protobuf Converters.

  • Type: boolean

  • Importance: low

value.converter.latest.compatibility.strict

Verify latest subject version is backward compatible when use.latest.version is true.

  • Type: boolean

  • Importance: low

value.converter.object.additional.properties

Whether to allow additional properties for object schemas. Applicable for JSON_SR Converters.

  • Type: boolean

  • Importance: low

value.converter.optional.for.nullables

Whether nullable fields should be specified with an optional label. Applicable for Protobuf Converters.

  • Type: boolean

  • Importance: low

value.converter.optional.for.proto2

Whether proto2 optionals are supported. Applicable for Protobuf Converters.

  • Type: boolean

  • Importance: low

value.converter.scrub.invalid.names

Whether to scrub invalid names by replacing invalid characters with valid characters. Applicable for Avro and Protobuf Converters.

  • Type: boolean

  • Importance: low

value.converter.use.latest.version

Use latest version of schema in subject for serialization when auto.register.schemas is false.

  • Type: boolean

  • Importance: low

value.converter.use.optional.for.nonrequired

Whether to set non-required properties to be optional. Applicable for JSON_SR Converters.

  • Type: boolean

  • Importance: low

value.converter.use.schema.guid

The schema GUID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema GUID to be used for deserializing message values. Only applicable when value.converter.value.schema.id.deserializer is set to ConfigSchemaIdDeserializer.

  • Type: string

  • Importance: low

value.converter.use.schema.id

The schema ID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema ID to be used for deserializing message values. Only applicable when value.converter.value.schema.id.deserializer is set to ConfigSchemaIdDeserializer.

  • Type: int

  • Importance: low

value.converter.wrapper.for.nullables

Whether nullable fields should use primitive wrapper messages. Applicable for Protobuf Converters.

  • Type: boolean

  • Importance: low

value.converter.wrapper.for.raw.primitives

Whether a wrapper message should be interpreted as a raw primitive at root level. Applicable for Protobuf Converters.

  • Type: boolean

  • Importance: low

errors.tolerance

Use this property if you would like to configure the connector’s error handling behavior. WARNING: This property should be used with CAUTION for SOURCE CONNECTORS as it may lead to dataloss. If you set this property to ‘all’, the connector will not fail on errant records, but will instead log them (and send to DLQ for Sink Connectors) and continue processing. If you set this property to ‘none’, the connector task will fail on errant records.

  • Type: string

  • Default: all

  • Importance: low

key.converter.key.schema.id.deserializer

The class name of the schema ID deserializer for keys. This is used to deserialize schema IDs from the message headers.

  • Type: string

  • Default: io.confluent.kafka.serializers.schema.id.DualSchemaIdDeserializer

  • Importance: low

key.converter.key.subject.name.strategy

How to construct the subject name for key schema registration.

  • Type: string

  • Default: TopicNameStrategy

  • Importance: low

value.converter.decimal.format

Specify the JSON/JSON_SR serialization format for Connect DECIMAL logical type values with two allowed literals:

BASE64 to serialize DECIMAL logical types as base64 encoded binary data and

NUMERIC to serialize Connect DECIMAL logical type values in JSON/JSON_SR as a number representing the decimal value.

  • Type: string

  • Default: BASE64

  • Importance: low

value.converter.flatten.singleton.unions

Whether to flatten singleton unions. Applicable for Avro and JSON_SR Converters.

  • Type: boolean

  • Default: false

  • Importance: low

value.converter.ignore.default.for.nullables

When set to true, this property ensures that the corresponding record in Kafka is NULL, instead of showing the default column value. Applicable for AVRO,PROTOBUF and JSON_SR Converters.

  • Type: boolean

  • Default: false

  • Importance: low

value.converter.reference.subject.name.strategy

Set the subject reference name strategy for value. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being DefaultReferenceSubjectNameStrategy.

  • Type: string

  • Default: DefaultReferenceSubjectNameStrategy

  • Importance: low

value.converter.replace.null.with.default

Whether to replace fields that have a default value and that are null to the default value. When set to true, the default value is used, otherwise null is used. Applicable for JSON Converter.

  • Type: boolean

  • Default: true

  • Importance: low

value.converter.schemas.enable

Include schemas within each of the serialized values. Input messages must contain schema and payload fields and may not contain additional fields. For plain JSON data, set this to false. Applicable for JSON Converter.

  • Type: boolean

  • Default: false

  • Importance: low

value.converter.value.schema.id.deserializer

The class name of the schema ID deserializer for values. This is used to deserialize schema IDs from the message headers.

  • Type: string

  • Default: io.confluent.kafka.serializers.schema.id.DualSchemaIdDeserializer

  • Importance: low

value.converter.value.subject.name.strategy

Determines how to construct the subject name under which the value schema is registered with Schema Registry.

  • Type: string

  • Default: TopicNameStrategy

  • Importance: low

Auto-restart policy

auto.restart.on.user.error

Enable connector to automatically restart on user-actionable errors.

  • Type: boolean

  • Default: true

  • Importance: medium

Next Steps

For an example that shows fully-managed Confluent Cloud connectors in action with Confluent Cloud for Apache Flink, see the Cloud ETL Demo. This example also shows how to use Confluent CLI to manage your resources in Confluent Cloud.

../../_images/topology.png