# How to connect Elasticsearch correctly in a Kafka connector? **URL:** \ **Category:** Elasticsearch\ **Tags:** docker\ **Created:** [December 4, 2021, 10:26am UTC](https://discuss.elastic.co/t/how-to-connect-elasticsearch-correctly-in-a-kafka-connector/290949 "2021-12-04T10:26:32Z")\ **Posts on this page:** 4\ **Page:** 1
**Author:** ![Hongbo-Miao](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/hongbo-miao/32/92664_2.png) [@Hongbo-Miao](https://discuss.elastic.co/u/Hongbo-Miao)\ **Post date:** [December 4, 2021, 10:26am UTC](https://discuss.elastic.co/t/how-to-connect-elasticsearch-correctly-in-a-kafka-connector/290949/1 "2021-12-04T10:26:32Z")
# Step 1 - Deploy Elasticsearch in Kubernetes (succeed) I deployed Elasticsearch in Kubernetes by following the tutorial > **[Deploy an Elasticsearch cluster | Elastic Docs](https://www.elastic.co/docs/deploy-manage/deploy/cloud-on-k8s/elasticsearch-deployment-quickstart)** > > To deploy a simple Elasticsearch cluster specification, with one Elasticsearch node: The operator automatically creates and manages Kubernetes resources... ```auto apiVersion: elasticsearch.k8s.elastic.co/v1 kind: Elasticsearch metadata: name: hm-elasticsearch namespace: elastic spec: version: 7.14.0 nodeSets: - name: default count: 1 config: node.store.allow_mmap: false ``` Based on the tutorial, I can succeed calling by providing username `elastic` and password `passw0rd` in the header by ```auto curl -u "elastic:passw0rd" -k " https://hm-elasticsearch-es-http.elastic:9200" ``` which returns ```auto { "name": "hm-elasticsearch-es-default-0", "cluster_name": "hm-elasticsearch", "cluster_uuid": "TWgIk0YGR_GVr7IJZcW62g", "version": { "number": "7.14.0", "build_flavor": "default", "build_type": "docker", "build_hash": "dd5a0a2acaa2045ff9624f3729fc8a6f40835aa1", "build_date": "2021-07-29T20:49:32.864135063Z", "build_snapshot": false, "lucene_version": "8.9.0", "minimum_wire_compatibility_version": "6.8.0", "minimum_index_compatibility_version": "6.0.0-beta1" }, "tagline": "You Know, for Search" } ``` # Step 2 - Add ElasticsearchSinkConnector (failed) Now I am trying to adding **ElasticsearchSinkConnector** , however, I am having issue to set it up. Based on [this ElasticsearchSinkConnector config](https://github.com/debezium/debezium-examples/blob/main/unwrap-smt/es-sink.json), I wrote my config, and then try to verify by ```auto curl --location --request PUT 'http://hm-connect-cluster-connect-api.kafka:8083/connector-plugins/io.confluent.connect.elasticsearch.ElasticsearchSinkConnector/config/validate' \ --header 'Content-Type: application/json' \ --data-raw '{ "name": "elasticsearch-sink", "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector", "tasks.max": "1", "topics": "roles", "connection.url": "https://hm-elasticsearch-es-http.elastic:9200", "connection.username": "elastic", "connection.password": "tD23h49YN34ouji74E4kt2ow", "transforms": "unwrap,key", "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState", "transforms.unwrap.drop.tombstones": "false", "transforms.key.type": "org.apache.kafka.connect.transforms.ExtractField$Key", "transforms.key.field": "id", "key.ignore": "false", "type.name": "role", "behavior.on.null.values": "delete" }' ``` It returns the error ```auto { "name": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector", "error_count": 3, "groups": [ "Common", "Transforms", "Predicates", "Error Handling", "Transforms: unwrap", "Transforms: key", "Connector", "Data Conversion", "Proxy", "Security", "Kerberos", "Data Stream" ], "configs": [ // ... { "definition": { "name": "connection.url", "type": "LIST", "required": true, "default_value": null, "importance": "HIGH", "documentation": "The comma-separated list of one or more Elasticsearch URLs, such as ``http://eshost1:9200,http://eshost2:9200`` or ``https://eshost3:9200``. HTTPS is used for all connections if any of the URLs starts with ``https:``. A URL without a protocol is treated as ``http``.", "group": "Connector", "width": "LONG", "display_name": "Connection URLs", "dependents": [], "order": 1 }, "value": { "name": "connection.url", "value": "https://hm-elasticsearch-es-http.elastic:9200", "recommended_values": [], "errors": [ "Could not connect to Elasticsearch. Error message: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target" ], "visible": true } }, { "definition": { "name": "connection.username", "type": "STRING", "required": false, "default_value": null, "importance": "MEDIUM", "documentation": "The username used to authenticate with Elasticsearch. The default is the null, and authentication will only be performed if both the username and password are non-null.", "group": "Connector", "width": "SHORT", "display_name": "Connection Username", "dependents": [], "order": 2 }, "value": { "name": "connection.username", "value": "elastic", "recommended_values": [], "errors": [ "Could not authenticate the user. Check the 'connection.username' and 'connection.password'. Error message: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target" ], "visible": true } }, { "definition": { "name": "connection.password", "type": "PASSWORD", "required": false, "default_value": null, "importance": "MEDIUM", "documentation": "The password used to authenticate with Elasticsearch. The default is the null, and authentication will only be performed if both the username and password are non-null.", "group": "Connector", "width": "SHORT", "display_name": "Connection Password", "dependents": [], "order": 3 }, "value": { "name": "connection.password", "value": "[hidden]", "recommended_values": [], "errors": [ "Could not authenticate the user. Check the 'connection.username' and 'connection.password'. Error message: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target" ], "visible": true } }, // ... ] } ``` The full verification response with all fields can be found at [hm-connect-cluster-connect-api verify error ยท GitHub](https://gist.github.com/Hongbo-Miao/34e51bfe91dfd169259ab81ba3fd942a) Then I found this document [Elasticsearch Connector with Security | Confluent Documentation](https://docs.confluent.io/kafka-connect-elasticsearch/current/security.html#step-4-configure-the-connector) which has more config regarding SSL ```auto "elastic.security.protocol": "SSL" "elastic.https.ssl.keystore.location": "/home/directory/elasticsearch-6.6.0/config/certs/keystore.jks" "elastic.https.ssl.keystore.password": "asdfasdf" "elastic.https.ssl.key.password": "asdfasdf" "elastic.https.ssl.keystore.type": "JKS" "elastic.https.ssl.truststore.location": "/home/directory/elasticsearch-6.6.0/config/certs/truststore.jks" "elastic.https.ssl.truststore.password": "asdfasdf" "elastic.https.ssl.truststore.type": "JKS" "elastic.https.ssl.protocol": "TLS" ``` In my case, I only need provider `username` and `password` to succeed. But in above config, there is no place to provide `username`. So I am not sure how to really write it correctly. Based on the deployment method in step 1, is it using **keystore** or **truststore**? ---
**Author:** ![Hongbo-Miao](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/hongbo-miao/32/92664_2.png) [@Hongbo-Miao](https://discuss.elastic.co/u/Hongbo-Miao)\ **Post date:** [December 9, 2021, 10:45pm UTC](https://discuss.elastic.co/t/how-to-connect-elasticsearch-correctly-in-a-kafka-connector/290949/3 "2021-12-09T22:45:47Z")
I cannot edit original question any more. I posted the same question with more info at Stack Overflow [kubernetes - How to connect Elasticsearch deployed by Elastic Operator correctly in a Kafka connector? - Stack Overflow](https://stackoverflow.com/questions/70267486/how-to-connect-elasticsearch-deployed-by-elastic-operator-correctly-in-a-kafka-c) ---
**Author:** ![Hongbo-Miao](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/hongbo-miao/32/92664_2.png) [@Hongbo-Miao](https://discuss.elastic.co/u/Hongbo-Miao)\ **Post date:** [December 11, 2021, 5:29am UTC](https://discuss.elastic.co/t/how-to-connect-elasticsearch-correctly-in-a-kafka-connector/290949/4 "2021-12-11T05:29:12Z")
I finally made it! Posted the solution at [kubernetes - How to connect Elasticsearch deployed by Elastic Operator correctly in a Kafka connector? - Stack Overflow](https://stackoverflow.com/questions/70267486/how-to-connect-elasticsearch-deployed-by-elastic-operator-correctly-in-a-kafka-c) Hope it helps future people! ๐Ÿ˜ƒ ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [January 8, 2022, 5:29am UTC](https://discuss.elastic.co/t/how-to-connect-elasticsearch-correctly-in-a-kafka-connector/290949/5 "2022-01-08T05:29:19Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.