Elasticsearch sink connector Features Connector name elasticsearch Compatibility Elasticsearch 7.x Elasticsearch 6.x Delivery guarantee At least once Supported task sizes S, M, L Multiplex capability A single instance of this connector can write a single Decodable stream to one or more Elasticsearch indices. Supported stream types Append stream Change stream Configuration properties Property Description Required Default hosts A semicolon-delimited list of Elasticsearch hosts to connect to. For example: http://es-node-1:9092;http://es-node-2:9093 While the Elasticsearch sink does support endpoints using TLS/SSL encryption, it doesn’t currently support client certificate authentication or specifying additional TLS/SSL options. Yes index The name of the Elasticsearch index to send data to. If the index doesn’t exist, Decodable will create it for you. Decodable supports static and dynamic index creation. See Types of Index. Yes username The username to connect to your Elasticsearch cluster. — password The authentication token associated with your Elasticsearch password. This must be provided as a secret resource. — document-type The Elasticsearch document mapping type. Elasticsearch v6 only Modifying the document-type of a connection that has already ingested records to an index will result in runtime errors. If you change the document-type, make sure to change or delete the index as well. — version The version of Elasticsearch to which you are connecting. INFO: Make sure to specify the correct version of Elasticsearch you are using when configuring your sink connection to avoid unexpected compatibility errors. — 7 Advanced sink.bulk-flush.max-actions Maximum number of buffered records per request. — 1000 sink.bulk-flush.max-size Maximum size of buffered records per request (in MB) — 2 sink.bulk-flush.interval The interval to flush buffered records (in seconds). — 1 Types of Index Static Index A single target index that all records will be continuously written into represented by a plain string index name. For example, users. Dynamic Index A dynamic index includes {field_name} in the index name to reference a field value and dynamically generate a target index for each record. The dynamic index name can also be of the form {field_name|date_format_string}, where date_format_string is a format pattern compatible with Java’s DateTimeFormatter. For example: users-at-{ts|yyyy-MM-dd}, where ts is a timestamp field in the connector schema and yyyy-MM-dd is an optional date format pattern. index-for-user-{user} When using a dynamic index generated by the current system time with a change stream, there is no guarantee that the records with the same primary key can generate the same index name. Therefore, if you are using a dynamic index based on the system time you should only use an append stream. Connector starting state and offsets A new sink connection will start reading from the Latest point in the source Decodable stream. This means that only data that’s written to the stream when the connection has started will be sent to the external system. You can override this when you start the connection to Earliest if you want to send all the existing data on the source stream to the target system, along with all new data that arrives on the stream. When you restart a sink connection it will continue to read data from the point it most recently stored in the checkpoint before the connection stopped. You can also opt to discard the connection’s state and restart it afresh from Earliest or Latest as described above. Learn more about starting state here.