<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>

Hongbo Miao
<p>First add more background. The way I deployed Kafka is using <a href="https://strimzi.io/quickstarts/" rel="noreferrer">Strimzi</a>:</p>
<pre><code>kubectl create namespace kafka
kubectl apply --filename=&quot;https://strimzi.io/install/latest?namespace=kafka&quot; --namespace=kafka
kubectl apply --filename=https://strimzi.io/examples/latest/kafka/kafka-persistent-single.yaml --namespace=kafka
</code></pre>
<p>Following the UPDATE 2 in the question, once got <strong>keystore.jks</strong>, I created the secret by</p>
<pre><code>kubectl create secret generic hm-elasticsearch-keystore \
  --from-file=keystore.jks \
  --namespace=kafka
</code></pre>
<p>I have files</p>
<p><strong>kafkaconnect.yaml</strong></p>
<pre><code>apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: hm-connect-cluster
  namespace: kafka
  annotations:
    # use-connector-resources configures this KafkaConnect
    # to use KafkaConnector resources to avoid
    # needing to call the Connect REST API directly
    strimzi.io/use-connector-resources: &quot;true&quot;
spec:
  image: hongbomiao/hm-connect-debezium:latest
  replicas: 1
  bootstrapServers: my-cluster-kafka-bootstrap:9093
  tls:
    trustedCertificates:
      - secretName: my-cluster-cluster-ca-cert
        certificate: ca.crt
  config:
    config.storage.replication.factor: 1
    offset.storage.replication.factor: 1
    status.storage.replication.factor: 1
    config.providers: file
    config.providers.file.class: org.apache.kafka.common.config.provider.FileConfigProvider
  externalConfiguration:
    volumes:
      - name: hm-elasticsearch-keystore-volume
        secret:
          secretName: hm-elasticsearch-keystore
</code></pre>
<p><strong>elasticsearch-sink-kafkaconnector.yaml</strong></p>
<pre><code>apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: elasticsearch-sink-connector
  namespace: kafka
  labels:
    strimzi.io/cluster: hm-connect-cluster
spec:
  class: io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
  tasksMax: 1
  # https://docs.confluent.io/kafka-connect-elasticsearch/current/configuration_options.html
  config:
    connector.class: &quot;io.confluent.connect.elasticsearch.ElasticsearchSinkConnector&quot;
    tasks.max: &quot;1&quot;
    topics: &quot;opa_db_server.public.roles&quot;
    connection.url: &quot;https://hm-elasticsearch-es-http.elastic:9200&quot;
    connection.username: &quot;elastic&quot;
    connection.password: &quot;passw0rd&quot;
    transforms: &quot;unwrap,key&quot;
    transforms.unwrap.type: &quot;io.debezium.transforms.ExtractNewRecordState&quot;
    transforms.unwrap.drop.tombstones: &quot;false&quot;
    transforms.key.type: &quot;org.apache.kafka.connect.transforms.ExtractField$Key&quot;
    transforms.key.field: &quot;role_id&quot;
    key.ignore: &quot;false&quot;
    behavior.on.null.values: &quot;delete&quot;
    elastic.security.protocol: &quot;SSL&quot;
    elastic.https.ssl.keystore.location: &quot;/opt/kafka/external-configuration/hm-elasticsearch-volume/keystore.jks&quot;
    elastic.https.ssl.keystore.password: &quot;MPx57vkACsRWKVap&quot;
    elastic.https.ssl.key.password: &quot;MPx57vkACsRWKVap&quot;
    elastic.https.ssl.keystore.type: &quot;JKS&quot;
    elastic.https.ssl.truststore.location: &quot;/opt/kafka/external-configuration/hm-elasticsearch-keystore-volume/keystore.jks&quot;
    elastic.https.ssl.truststore.password: &quot;MPx57vkACsRWKVap&quot;
    elastic.https.ssl.truststore.type: &quot;JKS&quot;
    elastic.https.ssl.protocol: &quot;TLS1.3&quot;
</code></pre>
<p>Note keystore and truststore share the same one.</p>
<p>Then just need run</p>
<pre><code>kubectl apply --filename=kafkaconnect.yaml
kubectl apply --filename=elasticsearch-sink-kafkaconnector.yaml
</code></pre>
<p>Now it works!</p>
