You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by ni...@apache.org on 2022/11/30 10:48:06 UTC
[pulsar] branch master updated: [improve][io] ElasticSearch sink: conditional id hashing (#18668)
This is an automated email from the ASF dual-hosted git repository.
nicoloboschi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 90c5534cd75 [improve][io] ElasticSearch sink: conditional id hashing (#18668)
90c5534cd75 is described below
commit 90c5534cd75275613a014640ea6283afdf1721d3
Author: Nicolò Boschi <bo...@gmail.com>
AuthorDate: Wed Nov 30 11:47:56 2022 +0100
[improve][io] ElasticSearch sink: conditional id hashing (#18668)
---
.../io/elasticsearch/ElasticSearchConfig.java | 8 +
.../pulsar/io/elasticsearch/ElasticSearchSink.java | 34 ++--
.../io/elasticsearch/ElasticSearchSinkTests.java | 174 ++++++++++++++-------
site2/docs/io-elasticsearch-sink.md | 75 ++++-----
4 files changed, 182 insertions(+), 109 deletions(-)
diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java
index 4f9a91a81ba..c70be56415f 100644
--- a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java
+++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java
@@ -294,6 +294,14 @@ public class ElasticSearchConfig implements Serializable {
)
private IdHashingAlgorithm idHashingAlgorithm = IdHashingAlgorithm.NONE;
+ @FieldDoc(
+ defaultValue = "false",
+ help = "This option only works if idHashingAlgorithm is set."
+ + "If enabled, the hashing is performed only if the id is greater than 512 bytes otherwise "
+ + "the hashing is performed on each document in any case."
+ )
+ private boolean conditionalIdHashing = false;
+
@FieldDoc(
defaultValue = "false",
help = "When the message key schema is AVRO or JSON, copy the message key fields into the Elasticsearch _source."
diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java
index 1e7df555271..05ca909f0c6 100644
--- a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java
+++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java
@@ -240,20 +240,28 @@ public class ElasticSearchSink implements Sink<GenericObject> {
if (id != null
&& idHashingAlgorithm != null
&& idHashingAlgorithm != ElasticSearchConfig.IdHashingAlgorithm.NONE) {
- Hasher hasher;
- switch (idHashingAlgorithm) {
- case SHA256:
- hasher = Hashing.sha256().newHasher();
- break;
- case SHA512:
- hasher = Hashing.sha512().newHasher();
- break;
- default:
- throw new UnsupportedOperationException("Unsupported IdHashingAlgorithm: "
- + idHashingAlgorithm);
+ final byte[] idBytes = id.getBytes(StandardCharsets.UTF_8);
+
+ boolean performHashing = true;
+ if (elasticSearchConfig.isConditionalIdHashing() && idBytes.length <= 512) {
+ performHashing = false;
+ }
+ if (performHashing) {
+ Hasher hasher;
+ switch (idHashingAlgorithm) {
+ case SHA256:
+ hasher = Hashing.sha256().newHasher();
+ break;
+ case SHA512:
+ hasher = Hashing.sha512().newHasher();
+ break;
+ default:
+ throw new UnsupportedOperationException("Unsupported IdHashingAlgorithm: "
+ + idHashingAlgorithm);
+ }
+ hasher.putBytes(idBytes);
+ id = base64Encoder.encodeToString(hasher.hash().asBytes());
}
- hasher.putString(id, StandardCharsets.UTF_8);
- id = base64Encoder.encodeToString(hasher.hash().asBytes());
}
if (log.isDebugEnabled()) {
diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java
index 7ab2342e101..9fad03c3579 100644
--- a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java
+++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSinkTests.java
@@ -91,10 +91,11 @@ public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase {
protected Map<String, Object> map;
protected ElasticSearchSink sink;
- static Schema kvSchema;
- static Schema<UserProfile> valueSchema;
- static GenericSchema<GenericRecord> genericSchema;
- static GenericRecord userProfile;
+ Schema kvSchema;
+ Schema<UserProfile> valueSchema;
+ GenericSchema<GenericRecord> genericSchema;
+ GenericRecord userProfile;
+ String recordKey;
@BeforeMethod(alwaysRun = true)
public final void initBeforeClass() {
@@ -103,16 +104,6 @@ public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase {
}
container = createElasticsearchContainer();
container.start();
-
- valueSchema = Schema.JSON(UserProfile.class);
- genericSchema = Schema.generic(valueSchema.getSchemaInfo());
- userProfile = genericSchema.newRecordBuilder()
- .set("name", "bob")
- .set("userName", "boby")
- .set("email", "bob@bob.com")
- .build();
- kvSchema = Schema.KeyValue(Schema.STRING, genericSchema, KeyValueEncodingType.SEPARATED);
-
}
@AfterClass(alwaysRun = true)
@@ -124,8 +115,19 @@ public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase {
@SuppressWarnings("unchecked")
@BeforeMethod
public final void setUp() throws Exception {
- map = new HashMap<String, Object> ();
- map.put("elasticSearchUrl", "http://"+container.getHttpHostAddress());
+
+ valueSchema = Schema.JSON(UserProfile.class);
+ genericSchema = Schema.generic(valueSchema.getSchemaInfo());
+ userProfile = genericSchema.newRecordBuilder()
+ .set("name", "bob")
+ .set("userName", "boby")
+ .set("email", "bob@bob.com")
+ .build();
+ recordKey = "bob";
+ kvSchema = Schema.KeyValue(Schema.STRING, genericSchema, KeyValueEncodingType.SEPARATED);
+
+ map = new HashMap<>();
+ map.put("elasticSearchUrl", "http://" + container.getHttpHostAddress());
map.put("schemaEnable", "true");
map.put("createIndexIfNeeded", "true");
sink = new ElasticSearchSink();
@@ -133,32 +135,19 @@ public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase {
mockRecord = mock(Record.class);
mockSinkContext = mock(SinkContext.class);
- when(mockRecord.getKey()).thenAnswer(new Answer<Optional<String>>() {
- long sequenceCounter = 0;
- public Optional<String> answer(InvocationOnMock invocation) throws Throwable {
- return Optional.of( "key-" + sequenceCounter++);
- }});
-
-
- when(mockRecord.getValue()).thenAnswer(new Answer<GenericObject>() {
- public GenericObject answer(InvocationOnMock invocation) throws Throwable {
- return new GenericObject() {
- @Override
- public SchemaType getSchemaType() {
- return SchemaType.KEY_VALUE;
- }
+ when(mockRecord.getValue()).thenAnswer((Answer<GenericObject>) invocation -> new GenericObject() {
+ @Override
+ public SchemaType getSchemaType() {
+ return SchemaType.KEY_VALUE;
+ }
- @Override
- public Object getNativeObject() {
- return new KeyValue<String, GenericObject>((String) userProfile.getField("name"), userProfile);
- }
- };
- }});
+ @Override
+ public Object getNativeObject() {
+ return new KeyValue<String, GenericObject>(recordKey, userProfile);
+ }
+ });
- when(mockRecord.getSchema()).thenAnswer(new Answer<Schema<KeyValue<String,UserProfile>>>() {
- public Schema<KeyValue<String,UserProfile>> answer(InvocationOnMock invocation) throws Throwable {
- return kvSchema;
- }});
+ when(mockRecord.getSchema()).thenAnswer((Answer<Schema<KeyValue<String, UserProfile>>>) invocation -> kvSchema);
}
@AfterMethod(alwaysRun = true)
@@ -185,7 +174,7 @@ public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase {
List<Node> nodeList = client.getClient().getLowLevelClient().getNodes();
assertEquals(nodeList.size(), 3);
}
-
+
@Test(expectedExceptions = IllegalArgumentException.class)
public final void invalidIndexNameTest() throws Exception {
map.put("indexName", "myIndex");
@@ -231,18 +220,21 @@ public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase {
when(mockRecord.getKey()).thenAnswer(new Answer<Optional<String>>() {
public Optional<String> answer(InvocationOnMock invocation) throws Throwable {
return null;
- }});
+ }
+ });
when(mockRecord.getValue()).thenAnswer(new Answer<String>() {
public String answer(InvocationOnMock invocation) throws Throwable {
return "hello";
- }});
+ }
+ });
when(mockRecord.getSchema()).thenAnswer(new Answer<Schema>() {
public Schema answer(InvocationOnMock invocation) throws Throwable {
return Schema.STRING;
- }});
+ }
+ });
map.put("indexName", "test-index");
map.put("schemaEnable", "false");
@@ -297,15 +289,15 @@ public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase {
}
}
- static class MockRecordNullValue implements Record<GenericObject> {
+ private class MockRecordNullValue implements Record<GenericObject> {
@Override
public Schema getSchema() {
- return kvSchema;
+ return kvSchema;
}
@Override
public Optional<String> getKey() {
- return Optional.of((String)userProfile.getField("name"));
+ return Optional.of((String) userProfile.getField("name"));
}
@Override
@@ -318,7 +310,7 @@ public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase {
@Override
public Object getNativeObject() {
- return new KeyValue<>((String)userProfile.getField("name"), null);
+ return new KeyValue<>((String) userProfile.getField("name"), null);
}
};
}
@@ -387,7 +379,7 @@ public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase {
@Override
public Optional<String> getKey() {
- return Optional.of((String)userProfile.getField("name"));
+ return Optional.of((String) userProfile.getField("name"));
}
@Override
@@ -400,7 +392,7 @@ public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase {
@Override
public Object getNativeObject() {
- return new KeyValue<String, GenericRecord>((String)userProfile.getField("name"), userProfile);
+ return new KeyValue<String, GenericRecord>((String) userProfile.getField("name"), userProfile);
}
};
}
@@ -455,17 +447,16 @@ public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase {
}
@DataProvider(name = "IdHashingAlgorithm")
- public Object[] schemaType() {
- return new Object[]{
- ElasticSearchConfig.IdHashingAlgorithm.SHA256,
- ElasticSearchConfig.IdHashingAlgorithm.SHA512
+ public Object[][] schemaType() {
+ return new Object[][]{
+ {ElasticSearchConfig.IdHashingAlgorithm.SHA256},
+ {ElasticSearchConfig.IdHashingAlgorithm.SHA512}
};
}
@Test(dataProvider = "IdHashingAlgorithm")
public final void testHashKey(ElasticSearchConfig.IdHashingAlgorithm algorithm) throws Exception {
- when(mockRecord.getKey()).thenAnswer((Answer<Optional<String>>) invocation -> Optional.of( "record-key"));
- final String indexName = "test-index" + UUID.randomUUID();
+ final String indexName = getNewIndexName();
map.put("indexName", indexName);
map.put("keyIgnore", "false");
map.put("idHashingAlgorithm", algorithm.toString());
@@ -480,6 +471,71 @@ public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase {
assertEquals(count, 1);
}
+
+ @DataProvider(name = "conditionalIdHashing")
+ public Object[][] conditionalIdHashing() {
+ return new Object[][]{
+ {false},
+ {true}
+ };
+ }
+
+ @Test(dataProvider = "conditionalIdHashing")
+ public final void testConditionalIdHashing(boolean conditionalIdHashing) throws Exception {
+ String longKey = "";
+ String shortKey = "";
+ String exactKey = "";
+ for (int i = 0; i < 513; i++) {
+ longKey += "a";
+ if (i < 511) {
+ shortKey += "b";
+ }
+ if (i < 512) {
+ exactKey += "c";
+ }
+ }
+ assertEquals(longKey.getBytes(StandardCharsets.UTF_8).length, 513);
+ assertEquals(shortKey.getBytes(StandardCharsets.UTF_8).length, 511);
+ assertEquals(exactKey.getBytes(StandardCharsets.UTF_8).length, 512);
+
+ final String indexName = getNewIndexName();
+ map.put("indexName", indexName);
+ map.put("keyIgnore", "false");
+ map.put("idHashingAlgorithm", "SHA256");
+ map.put("conditionalIdHashing", conditionalIdHashing + "");
+ sink.open(map, mockSinkContext);
+
+ recordKey = longKey;
+ send(1);
+ verify(mockRecord, times(1)).ack();
+ String expectedValue = "AkJcD1sNq/PSuRFfP3cjoCrYvPsVNKDSMWFP1CuBiPY";
+ assertEquals(sink.getElasticsearchClient().getRestClient()
+ .totalHits(indexName, "_id:" + expectedValue), 1);
+
+
+ recordKey = exactKey;
+ send(1);
+ verify(mockRecord, times(2)).ack();
+ expectedValue = conditionalIdHashing ? exactKey : "fiu8dRsHGN8giT4ZIIct9e+PZwO07LlTXxVWqUeWs88";
+ assertEquals(sink.getElasticsearchClient().getRestClient()
+ .totalHits(indexName, "_id:" + expectedValue), 1);
+
+ recordKey = shortKey;
+ send(1);
+ verify(mockRecord, times(3)).ack();
+ expectedValue = conditionalIdHashing ? shortKey : "/DatxoX5RmSN3ISp+GhRkdeWvPX6GYeMZldecSStEoI";
+ assertEquals(sink.getElasticsearchClient().getRestClient()
+ .totalHits(indexName, "_id:" + expectedValue), 1);
+
+ // verify there are 3 different documents
+ assertEquals(sink.getElasticsearchClient().getRestClient()
+ .totalHits(indexName, "*:*"), 3);
+ }
+
+ private String getNewIndexName() {
+ return "test-index" + UUID.randomUUID();
+ }
+
@Test
public final void testKeyValueHashAndCanonicalOutput() throws Exception {
RecordSchemaBuilder keySchemaBuilder = SchemaBuilder.record("key");
@@ -502,7 +558,7 @@ public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase {
keySchema, keyGenericRecord);
Record<GenericObject> genericObjectRecord2 = createKeyValueGenericRecordWithGenericKeySchema(
keySchema, keyGenericRecord2);
- final String indexName = "test-index" + UUID.randomUUID();
+ final String indexName = getNewIndexName();
map.put("indexName", indexName);
map.put("keyIgnore", "false");
map.put("nullValueAction", ElasticSearchConfig.NullValueAction.DELETE.toString());
@@ -548,7 +604,7 @@ public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase {
Schema<KeyValue<GenericRecord, GenericRecord>> keyValueSchema =
Schema.KeyValue(keySchema, genericSchema, KeyValueEncodingType.INLINE);
KeyValue<GenericRecord, GenericRecord> keyValue = new KeyValue<>(keyGenericRecord,
- emptyValue ? null: userProfile);
+ emptyValue ? null : userProfile);
GenericObject genericObject = new GenericObject() {
@Override
public SchemaType getSchemaType() {
@@ -567,7 +623,7 @@ public abstract class ElasticSearchSinkTests extends ElasticSearchTestBase {
}
@Override
- public Schema getSchema() {
+ public Schema getSchema() {
return keyValueSchema;
}
diff --git a/site2/docs/io-elasticsearch-sink.md b/site2/docs/io-elasticsearch-sink.md
index ba593e62ad1..edb9adf7f14 100644
--- a/site2/docs/io-elasticsearch-sink.md
+++ b/site2/docs/io-elasticsearch-sink.md
@@ -52,43 +52,44 @@ The configuration of the Elasticsearch sink connector has the following properti
### Property
-| Name | Type|Required | Default | Description
-|------|----------|----------|---------|-------------|
-| `elasticSearchUrl` | String| true |" " (empty string)| The URL of elastic search cluster to which the connector connects. |
-| `indexName` | String| false |" " (empty string)| The index name to which the connector writes messages. The default value is the topic name. It accepts date formats in the name to support event time based index with the pattern `%{+<date-format>}`. For example, suppose the event time of the record is 1645182000000L, the indexName is `logs-%{+yyyy-MM-dd}`, then the formatted index name would be `logs-2022-02-18`. |
-| `schemaEnable` | Boolean | false | false | Turn on the Schema Aware mode. |
-| `createIndexIfNeeded` | Boolean | false | false | Manage index if missing. |
-| `maxRetries` | Integer | false | 1 | The maximum number of retries for elasticsearch requests. Use -1 to disable it. |
-| `retryBackoffInMs` | Integer | false | 100 | The base time to wait when retrying an Elasticsearch request (in milliseconds). |
-| `maxRetryTimeInSec` | Integer| false | 86400 | The maximum retry time interval in seconds for retrying an elasticsearch request. |
-| `bulkEnabled` | Boolean | false | false | Enable the elasticsearch bulk processor to flush write requests based on the number or size of requests, or after a given period. |
-| `bulkActions` | Integer | false | 1000 | The maximum number of actions per elasticsearch bulk request. Use -1 to disable it. |
-| `bulkSizeInMb` | Integer | false |5 | The maximum size in megabytes of elasticsearch bulk requests. Use -1 to disable it. |
-| `bulkConcurrentRequests` | Integer | false | 0 | The maximum number of in flight elasticsearch bulk requests. The default 0 allows the execution of a single request. A value of 1 means 1 concurrent request is allowed to be executed while accumulating new bulk requests. |
-| `bulkFlushIntervalInMs` | Long | false | 1000 | The maximum period of time to wait for flushing pending writes when bulk writes are enabled. -1 or zero means the scheduled flushing is disabled. |
-| `compressionEnabled` | Boolean | false |false | Enable elasticsearch request compression. |
-| `connectTimeoutInMs` | Integer | false |5000 | The elasticsearch client connection timeout in milliseconds. |
-| `connectionRequestTimeoutInMs` | Integer | false |1000 | The time in milliseconds for getting a connection from the elasticsearch connection pool. |
-| `connectionIdleTimeoutInMs` | Integer | false |5 | Idle connection timeout to prevent a read timeout. |
-| `keyIgnore` | Boolean | false |true | Whether to ignore the record key to build the Elasticsearch document `_id`. If primaryFields is defined, the connector extract the primary fields from the payload to build the document `_id` If no primaryFields are provided, elasticsearch auto generates a random document `_id`. |
-| `primaryFields` | String | false | "id" | The comma separated ordered list of field names used to build the Elasticsearch document `_id` from the record value. If this list is a singleton, the field is converted as a string. If this list has 2 or more fields, the generated `_id` is a string representation of a JSON array of the field values. |
-| `nullValueAction` | enum (IGNORE,DELETE,FAIL) | false | IGNORE | How to handle records with null values, possible options are IGNORE, DELETE or FAIL. Default is IGNORE the message. |
-| `malformedDocAction` | enum (IGNORE,WARN,FAIL) | false | FAIL | How to handle elasticsearch rejected documents due to some malformation. Possible options are IGNORE, DELETE or FAIL. Default is FAIL the Elasticsearch document. |
-| `stripNulls` | Boolean | false |true | If stripNulls is false, elasticsearch _source includes 'null' for empty fields (for example {"foo": null}), otherwise null fields are stripped. |
-| `socketTimeoutInMs` | Integer | false |60000 | The socket timeout in milliseconds waiting to read the elasticsearch response. |
-| `typeName` | String | false | "_doc" | The type name to which the connector writes messages to. <br /><br /> The value should be set explicitly to a valid type name other than "_doc" for Elasticsearch version before 6.2, and left to default otherwise. |
-| `indexNumberOfShards` | int| false |1| The number of shards of the index. |
-| `indexNumberOfReplicas` | int| false |1 | The number of replicas of the index. |
-| `username` | String| false |" " (empty string)| The username used by the connector to connect to the elastic search cluster. <br /><br />If `username` is set, then `password` should also be provided. |
-| `password` | String| false | " " (empty string)|The password used by the connector to connect to the elastic search cluster. <br /><br />If `username` is set, then `password` should also be provided. |
-| `ssl` | ElasticSearchSslConfig | false | | Configuration for TLS encrypted communication |
-| `compatibilityMode` | enum (AUTO,ELASTICSEARCH,ELASTICSEARCH_7,OPENSEARCH) | false | AUTO | Specify compatibility mode with the ElasticSearch cluster. `AUTO` value will try to auto detect the correct compatibility mode to use. Use `ELASTICSEARCH_7` if the target cluster is running ElasticSearch 7 or prior. Use `ELASTICSEARCH` if the target cluster is running ElasticSearch 8 or higher. Use `OPENSEARCH` if the target cluster is running OpenSearch. |
-| `token` | String| false | " " (empty string)|The token used by the connector to connect to the ElasticSearch cluster. Only one between basic/token/apiKey authentication mode must be configured. |
-| `apiKey` | String| false | " " (empty string)|The apiKey used by the connector to connect to the ElasticSearch cluster. Only one between basic/token/apiKey authentication mode must be configured. |
-| `canonicalKeyFields` | Boolean | false | false | Whether to sort the key fields for JSON and Avro or not. If it is set to `true` and the record key schema is `JSON` or `AVRO`, the serialized object does not consider the order of properties. |
-| `stripNonPrintableCharacters` | Boolean| false | true| Whether to remove all non-printable characters from the document or not. If it is set to true, all non-printable characters are removed from the document. |
-| `idHashingAlgorithm` | enum(NONE,SHA256,SHA512)| false | NONE|Hashing algorithm to use for the document id. This is useful in order to be compliant with the ElasticSearch _id hard limit of 512 bytes. |
-| `copyKeyFields` | Boolean | false | false |If the message key schema is AVRO or JSON, the message key fields are copied into the ElasticSearch document. |
+| Name | Type | Required | Default | Description |
+|--------------------------------|------------------------------------------------------|----------|--------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| `elasticSearchUrl` | String | true | " " (empty string) | The URL of elastic search cluster to which the connector connects. |
+| `indexName` | String | false | " " (empty string) | The index name to which the connector writes messages. The default value is the topic name. It accepts date formats in the name to support event time based index with the pattern `%{+<date-format>}`. For example, suppose the event time of the record is 1645182000000L, the indexName is `logs-%{+yyyy-MM-dd}`, then the formatted index name would be `logs-2022-02-18`. |
+| `schemaEnable` | Boolean | false | false | Turn on the Schema Aware mode. |
+| `createIndexIfNeeded` | Boolean | false | false | Manage index if missing. |
+| `maxRetries` | Integer | false | 1 | The maximum number of retries for elasticsearch requests. Use -1 to disable it. |
+| `retryBackoffInMs` | Integer | false | 100 | The base time to wait when retrying an Elasticsearch request (in milliseconds). |
+| `maxRetryTimeInSec` | Integer | false | 86400 | The maximum retry time interval in seconds for retrying an elasticsearch request. |
+| `bulkEnabled` | Boolean | false | false | Enable the elasticsearch bulk processor to flush write requests based on the number or size of requests, or after a given period. |
+| `bulkActions` | Integer | false | 1000 | The maximum number of actions per elasticsearch bulk request. Use -1 to disable it. |
+| `bulkSizeInMb` | Integer | false | 5 | The maximum size in megabytes of elasticsearch bulk requests. Use -1 to disable it. |
+| `bulkConcurrentRequests` | Integer | false | 0 | The maximum number of in flight elasticsearch bulk requests. The default 0 allows the execution of a single request. A value of 1 means 1 concurrent request is allowed to be executed while accumulating new bulk requests. |
+| `bulkFlushIntervalInMs` | Long | false | 1000 | The maximum period of time to wait for flushing pending writes when bulk writes are enabled. -1 or zero means the scheduled flushing is disabled. |
+| `compressionEnabled` | Boolean | false | false | Enable elasticsearch request compression. |
+| `connectTimeoutInMs` | Integer | false | 5000 | The elasticsearch client connection timeout in milliseconds. |
+| `connectionRequestTimeoutInMs` | Integer | false | 1000 | The time in milliseconds for getting a connection from the elasticsearch connection pool. |
+| `connectionIdleTimeoutInMs` | Integer | false | 5 | Idle connection timeout to prevent a read timeout. |
+| `keyIgnore` | Boolean | false | true | Whether to ignore the record key to build the Elasticsearch document `_id`. If primaryFields is defined, the connector extract the primary fields from the payload to build the document `_id` If no primaryFields are provided, elasticsearch auto generates a random document `_id`. |
+| `primaryFields` | String | false | "id" | The comma separated ordered list of field names used to build the Elasticsearch document `_id` from the record value. If this list is a singleton, the field is converted as a string. If this list has 2 or more fields, the generated `_id` is a string representation of a JSON array of the field values. |
+| `nullValueAction` | enum (IGNORE,DELETE,FAIL) | false | IGNORE | How to handle records with null values, possible options are IGNORE, DELETE or FAIL. Default is IGNORE the message. |
+| `malformedDocAction` | enum (IGNORE,WARN,FAIL) | false | FAIL | How to handle elasticsearch rejected documents due to some malformation. Possible options are IGNORE, DELETE or FAIL. Default is FAIL the Elasticsearch document. |
+| `stripNulls` | Boolean | false | true | If stripNulls is false, elasticsearch _source includes 'null' for empty fields (for example {"foo": null}), otherwise null fields are stripped. |
+| `socketTimeoutInMs` | Integer | false | 60000 | The socket timeout in milliseconds waiting to read the elasticsearch response. |
+| `typeName` | String | false | "_doc" | The type name to which the connector writes messages to. <br /><br /> The value should be set explicitly to a valid type name other than "_doc" for Elasticsearch version before 6.2, and left to default otherwise. |
+| `indexNumberOfShards` | int | false | 1 | The number of shards of the index. |
+| `indexNumberOfReplicas` | int | false | 1 | The number of replicas of the index. |
+| `username` | String | false | " " (empty string) | The username used by the connector to connect to the elastic search cluster. <br /><br />If `username` is set, then `password` should also be provided. |
+| `password` | String | false | " " (empty string) | The password used by the connector to connect to the elastic search cluster. <br /><br />If `username` is set, then `password` should also be provided. |
+| `ssl` | ElasticSearchSslConfig | false | | Configuration for TLS encrypted communication |
+| `compatibilityMode` | enum (AUTO,ELASTICSEARCH,ELASTICSEARCH_7,OPENSEARCH) | false | AUTO | Specify compatibility mode with the ElasticSearch cluster. `AUTO` value will try to auto detect the correct compatibility mode to use. Use `ELASTICSEARCH_7` if the target cluster is running ElasticSearch 7 or prior. Use `ELASTICSEARCH` if the target cluster is running ElasticSearch 8 or higher. Use `OPENSEARCH` if the target cluster is running OpenSearch. |
+| `token` | String | false | " " (empty string) | The token used by the connector to connect to the ElasticSearch cluster. Only one between basic/token/apiKey authentication mode must be configured. |
+| `apiKey` | String | false | " " (empty string) | The apiKey used by the connector to connect to the ElasticSearch cluster. Only one between basic/token/apiKey authentication mode must be configured. |
+| `canonicalKeyFields` | Boolean | false | false | Whether to sort the key fields for JSON and Avro or not. If it is set to `true` and the record key schema is `JSON` or `AVRO`, the serialized object does not consider the order of properties. |
+| `stripNonPrintableCharacters` | Boolean | false | true | Whether to remove all non-printable characters from the document or not. If it is set to true, all non-printable characters are removed from the document. |
+| `idHashingAlgorithm` | enum(NONE,SHA256,SHA512) | false | NONE | Hashing algorithm to use for the document id. This is useful in order to be compliant with the ElasticSearch _id hard limit of 512 bytes. |
+| `conditionalIdHashing` | Boolean | false | false | This option only works if idHashingAlgorithm is set. If enabled, the hashing is performed only if the id is greater than 512 bytes otherwise the hashing is performed on each document in any case. |
+| `copyKeyFields` | Boolean | false | false | If the message key schema is AVRO or JSON, the message key fields are copied into the ElasticSearch document. |
### Definition of ElasticSearchSslConfig structure: