title: How to connect Elasticsearch deployed by Elastic Operator correctly in a Kafka connector?
body: <p>I have some CDC data in Kafka. Now I am trying to sink from Kafka to Elasticsearch. Here is what I have done so far:</p>
  <h2>Step 1 - Deploy Elasticsearch in Kubernetes (succeed)</h2>
  <p>I deployed Elasticsearch in Kubernetes by following this tutorial using Elastic Operator:</p>
  <ol>
  <li>Deploy ECK in your Kubernetes cluster: <a href="https://www.elastic.co/guide/en/cloud-on-k8s/current/k8s-deploy-eck.html" rel="nofollow noreferrer">https://www.elastic.co/guide/en/cloud-on-k8s/current/k8s-deploy-eck.html</a></li>
  <li>Deploy an Elasticsearch cluster: <a href="https://www.elastic.co/guide/en/cloud-on-k8s/current/k8s-deploy-elasticsearch.html" rel="nofollow noreferrer">https://www.elastic.co/guide/en/cloud-on-k8s/current/k8s-deploy-elasticsearch.html</a></li>
  </ol>
  <pre><code>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
  </code></pre>
  <p>Based on the tutorial, I can succeed calling by providing username <code>elastic</code> and password <code>passw0rd</code> in the header by</p>
  <pre><code>curl -u &quot;elastic:passw0rd&quot; -k &quot;https://hm-elasticsearch-es-http.elastic:9200&quot;
  </code></pre>
  <p>which returns</p>
  <pre><code>{
      &quot;name&quot;: &quot;hm-elasticsearch-es-default-0&quot;,
      &quot;cluster_name&quot;: &quot;hm-elasticsearch&quot;,
      &quot;cluster_uuid&quot;: &quot;TWgIk0YGR_GVr7IJZcW62g&quot;,
      &quot;version&quot;: {
          &quot;number&quot;: &quot;7.14.0&quot;,
          &quot;build_flavor&quot;: &quot;default&quot;,
          &quot;build_type&quot;: &quot;docker&quot;,
          &quot;build_hash&quot;: &quot;dd5a0a2acaa2045ff9624f3729fc8a6f40835aa1&quot;,
          &quot;build_date&quot;: &quot;2021-07-29T20:49:32.864135063Z&quot;,
          &quot;build_snapshot&quot;: false,
          &quot;lucene_version&quot;: &quot;8.9.0&quot;,
          &quot;minimum_wire_compatibility_version&quot;: &quot;6.8.0&quot;,
          &quot;minimum_index_compatibility_version&quot;: &quot;6.0.0-beta1&quot;
      },
      &quot;tagline&quot;: &quot;You Know, for Search&quot;
  }
  </code></pre>
  <h2>Step 2 - Add ElasticsearchSinkConnector (failed to connect Elasticsearch)</h2>
  <p>Now I am trying to adding ElasticsearchSinkConnector, however, I am having issue to set it up.</p>
  <p>I saw <a href="https://stackoverflow.com/questions/58381240/how-to-kafka-connect-elasticsearch-with-ssl">How to Kafka Connect Elasticsearch with SSL?</a>. Elastic Cloud just needs just to pass the username and password. So I thought it would be similar to mine.</p>
  <p>Also, based on <a href="https://github.com/debezium/debezium-examples/blob/main/unwrap-smt/es-sink.json" rel="nofollow noreferrer">this ElasticsearchSinkConnector config</a>, I wrote my config, and then try to verify by</p>
  <pre><code>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 '{
      &quot;name&quot;: &quot;elasticsearch-sink&quot;,
      &quot;connector.class&quot;: &quot;io.confluent.connect.elasticsearch.ElasticsearchSinkConnector&quot;,
      &quot;tasks.max&quot;: &quot;1&quot;,
      &quot;topics&quot;: &quot;opa_db_server.public.roles&quot;,
      &quot;connection.url&quot;: &quot;https://hm-elasticsearch-es-http.elastic:9200&quot;,
      &quot;connection.username&quot;: &quot;elastic&quot;,
      &quot;connection.password&quot;: &quot;passw0rd&quot;,
      &quot;transforms&quot;: &quot;unwrap,key&quot;,
      &quot;transforms.unwrap.type&quot;: &quot;io.debezium.transforms.ExtractNewRecordState&quot;,
      &quot;transforms.unwrap.drop.tombstones&quot;: &quot;false&quot;,
      &quot;transforms.key.type&quot;: &quot;org.apache.kafka.connect.transforms.ExtractField$Key&quot;,
      &quot;transforms.key.field&quot;: &quot;role_id&quot;,
      &quot;key.ignore&quot;: &quot;false&quot;,
      &quot;behavior.on.null.values&quot;: &quot;delete&quot;
  }'
  </code></pre>
  <p>It returns the error</p>
  <pre><code>{
      &quot;name&quot;: &quot;io.confluent.connect.elasticsearch.ElasticsearchSinkConnector&quot;,
      &quot;error_count&quot;: 3,
      &quot;groups&quot;: [
          &quot;Common&quot;,
          &quot;Transforms&quot;,
          &quot;Predicates&quot;,
          &quot;Error Handling&quot;,
          &quot;Transforms: unwrap&quot;,
          &quot;Transforms: key&quot;,
          &quot;Connector&quot;,
          &quot;Data Conversion&quot;,
          &quot;Proxy&quot;,
          &quot;Security&quot;,
          &quot;Kerberos&quot;,
          &quot;Data Stream&quot;
      ],
      &quot;configs&quot;: [
          // ...
          {
              &quot;definition&quot;: {
                  &quot;name&quot;: &quot;connection.url&quot;,
                  &quot;type&quot;: &quot;LIST&quot;,
                  &quot;required&quot;: true,
                  &quot;default_value&quot;: null,
                  &quot;importance&quot;: &quot;HIGH&quot;,
                  &quot;documentation&quot;: &quot;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``.&quot;,
                  &quot;group&quot;: &quot;Connector&quot;,
                  &quot;width&quot;: &quot;LONG&quot;,
                  &quot;display_name&quot;: &quot;Connection URLs&quot;,
                  &quot;dependents&quot;: [],
                  &quot;order&quot;: 1
              },
              &quot;value&quot;: {
                  &quot;name&quot;: &quot;connection.url&quot;,
                  &quot;value&quot;: &quot;https://hm-elasticsearch-es-http.elastic:9200&quot;,
                  &quot;recommended_values&quot;: [],
                  &quot;errors&quot;: [
                      &quot;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&quot;
                  ],
                  &quot;visible&quot;: true
              }
          },
          {
              &quot;definition&quot;: {
                  &quot;name&quot;: &quot;connection.username&quot;,
                  &quot;type&quot;: &quot;STRING&quot;,
                  &quot;required&quot;: false,
                  &quot;default_value&quot;: null,
                  &quot;importance&quot;: &quot;MEDIUM&quot;,
                  &quot;documentation&quot;: &quot;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.&quot;,
                  &quot;group&quot;: &quot;Connector&quot;,
                  &quot;width&quot;: &quot;SHORT&quot;,
                  &quot;display_name&quot;: &quot;Connection Username&quot;,
                  &quot;dependents&quot;: [],
                  &quot;order&quot;: 2
              },
              &quot;value&quot;: {
                  &quot;name&quot;: &quot;connection.username&quot;,
                  &quot;value&quot;: &quot;elastic&quot;,
                  &quot;recommended_values&quot;: [],
                  &quot;errors&quot;: [
                      &quot;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&quot;
                  ],
                  &quot;visible&quot;: true
              }
          },
          {
              &quot;definition&quot;: {
                  &quot;name&quot;: &quot;connection.password&quot;,
                  &quot;type&quot;: &quot;PASSWORD&quot;,
                  &quot;required&quot;: false,
                  &quot;default_value&quot;: null,
                  &quot;importance&quot;: &quot;MEDIUM&quot;,
                  &quot;documentation&quot;: &quot;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.&quot;,
                  &quot;group&quot;: &quot;Connector&quot;,
                  &quot;width&quot;: &quot;SHORT&quot;,
                  &quot;display_name&quot;: &quot;Connection Password&quot;,
                  &quot;dependents&quot;: [],
                  &quot;order&quot;: 3
              },
              &quot;value&quot;: {
                  &quot;name&quot;: &quot;connection.password&quot;,
                  &quot;value&quot;: &quot;[hidden]&quot;,
                  &quot;recommended_values&quot;: [],
                  &quot;errors&quot;: [
                      &quot;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&quot;
                  ],
                  &quot;visible&quot;: true
              }
          },
          // ...
      ]
  }
  </code></pre>
  <p>The full verification response with all fields can be found <a href="https://gist.github.com/Hongbo-Miao/34e51bfe91dfd169259ab81ba3fd942a" rel="nofollow noreferrer">here</a> which has all info for the config field can be used.</p>
  <h1>UPDATE 1 (12/8/2021):</h1>
  <p>I found <a href="https://docs.confluent.io/kafka-connect-elasticsearch/current/security.html#step-4-configure-the-connector" rel="nofollow noreferrer">this document</a> which has more config regarding SSL</p>
  <pre><code>&quot;elastic.security.protocol&quot;: &quot;SSL&quot;
  &quot;elastic.https.ssl.keystore.location&quot;: &quot;/path/to/keystore.jks&quot;
  &quot;elastic.https.ssl.keystore.password&quot;: &quot;xxx&quot;
  &quot;elastic.https.ssl.key.password&quot;: &quot;xxx&quot;
  &quot;elastic.https.ssl.keystore.type&quot;: &quot;JKS&quot;
  &quot;elastic.https.ssl.truststore.location&quot;: &quot;/path/to/truststore.jks&quot;
  &quot;elastic.https.ssl.truststore.password&quot;: &quot;xxx&quot;
  &quot;elastic.https.ssl.truststore.type&quot;: &quot;JKS&quot;
  &quot;elastic.https.ssl.protocol&quot;: &quot;TLS&quot;
  </code></pre>
  <p>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.</p>
  <h1>UPDATE 2 (12/9/2021):</h1>
  <p>I have these pods running</p>
  <pre><code>&gt; kubectl get pods --namespace=elastic
  NAME                                 READY   STATUS    RESTARTS   AGE
  hm-kibana-kb-77d4d9b456-m6th9        1/1     Running   0          6d3h
  hm-elasticsearch-es-default-0        1/1     Running   0          35h
  </code></pre>
  <p>and these secrets</p>
  <pre><code>kubectl get secrets --namespace=elastic
  NAME                                             TYPE                                  DATA   AGE
  hm-kibana-kibana-user                            Opaque                                1      6d3h
  elastic-hm-kibana-kibana-user                    Opaque                                3      6d3h
  hm-kibana-kb-http-ca-internal                    Opaque                                2      6d3h
  hm-elasticsearch-es-http-ca-internal             Opaque                                2      6d3h
  hm-elasticsearch-es-http-certs-internal          Opaque                                3      6d3h
  hm-elasticsearch-es-http-certs-public            Opaque                                2      6d3h
  hm-kibana-kb-es-ca                               Opaque                                2      6d3h
  hm-kibana-kb-http-certs-internal                 Opaque                                3      6d3h
  hm-kibana-kb-http-certs-public                   Opaque                                2      6d3h
  hm-elasticsearch-es-transport-ca-internal        Opaque                                2      6d3h
  hm-elasticsearch-es-transport-certs-public       Opaque                                1      6d3h
  hm-elasticsearch-es-remote-ca                    Opaque                                1      6d3h
  hm-elasticsearch-es-elastic-user                 Opaque                                1      6d3h
  hm-elasticsearch-es-internal-users               Opaque                                3      6d3h
  hm-elasticsearch-es-xpack-file-realm             Opaque                                3      6d3h
  hm-elasticsearch-es-default-es-config            Opaque                                1      6d3h
  hm-elasticsearch-es-default-es-transport-certs   Opaque                                3      6d3h
  hm-kibana-kb-config                              Opaque                                2      6d3h
  </code></pre>
  <p>I am able to save <code>ca.crt</code>, <code>tls.crt</code>, <code>tls.key</code> locally by</p>
  <pre><code>kubectl get secret hm-elasticsearch-es-http-certs-public \
    --namespace=elastic \
    --output=go-template='{{index .data &quot;ca.crt&quot; | base64decode }}' \
    &gt; ca.crt
  
  kubectl get secret hm-elasticsearch-es-http-certs-public \
    --namespace=elastic \
    --output=go-template='{{index .data &quot;tls.crt&quot; | base64decode }}' \
    &gt; tls.crt
  
  kubectl get secret hm-elasticsearch-es-http-certs-internal \
    --namespace=elastic \
    --output=go-template='{{index .data &quot;tls.key&quot; | base64decode }}' \
    &gt; tls.key
  </code></pre>
  <p>Before I have to use <code>-k</code> flag to disable the certificate verification to let the <code>curl</code> succeed. Now in Kubernetes, I can succeed without the flag:</p>
  <pre><code>curl --request GET \
       --url https://hm-elasticsearch-es-http.elastic:9200 \
       --cacert ca.crt \
       --key tls.key \
       --cert tls.crt \
       --header 'Content-Type: application/json' \
       -u &quot;elastic:passw0rd&quot;
  </code></pre>
  <p>Also, I succeed generating <code>keystore.jks</code> by</p>
  <pre><code>openssl pkcs12 -export \
    -in tls.crt \
    -inkey tls.key \
    -CAfile ca.crt \
    -caname root \
    -out keystore.p12 \
    -password pass:SFLzyT8DPkGGjDtn \
    -name hm-elasticsearch-keystore
  
  keytool -importkeystore \
    -srckeystore keystore.p12 \
    -srcstoretype PKCS12 \
    -srcstorepass SFLzyT8DPkGGjDtn \
    -deststorepass MPx57vkACsRWKVap \
    -destkeypass MPx57vkACsRWKVap \
    -destkeystore keystore.jks \
    -alias hm-elasticsearch-keystore
  </code></pre>
  <p>I am still not sure how <strong>truststore</strong> in UPDATE 1 is. But now I can fill this part regarding <strong>keystore</strong>:</p>
  <pre><code>&quot;elastic.https.ssl.keystore.location&quot;: &quot;/path/to/keystore.jks&quot;
  &quot;elastic.https.ssl.keystore.password&quot;: &quot;MPx57vkACsRWKVap&quot;
  &quot;elastic.https.ssl.key.password&quot;: &quot;MPx57vkACsRWKVap&quot;
  &quot;elastic.https.ssl.keystore.type&quot;: &quot;JKS&quot;
  </code></pre>
  <p>However, my <code>keystore.jks</code> is currently on my laptop locally. My Elasticsearch and ElasticsearchSinkConnector is running in the Kubernetes.</p>
  <p>Where does this <code>&quot;elastic.https.ssl.keystore.location&quot;: &quot;/path/to/keystore.jks&quot;</code> refer to?</p>