hongbo-miao/hongbomiao.com

---
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: hm-dummy-log-elasticsearch-sink-kafka-connector
  namespace: hm-kafka
  labels:
    strimzi.io/cluster: hm-kafka-logging-kafka-connect
spec:
  class: io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
  tasksMax: 1
  # https://docs.confluent.io/kafka-connect-elasticsearch/current/configuration_options.html
  config:
    connector.class: io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
    tasks.max: 1
    topics: hm.dummy-log
    connection.url: https://hm-elasticsearch-es-http.hm-elastic:9200
    connection.username: ${file:/opt/kafka/external-configuration/hm-elasticsearch-credentials-volume/elasticsearch-credentials.properties:elasticsearch_username}"
    connection.password: "${file:/opt/kafka/external-configuration/hm-elasticsearch-credentials-volume/elasticsearch-credentials.properties:elasticsearch_password}"
    elastic.security.protocol: SSL
    elastic.https.ssl.keystore.location: /opt/kafka/external-configuration/hm-elasticsearch-keystore-volume/elasticsearch-keystore.jks
    elastic.https.ssl.keystore.password: "${file:/opt/kafka/external-configuration/hm-elasticsearch-credentials-volume/elasticsearch-credentials.properties:elasticsearch_keystore_password}"
    elastic.https.ssl.key.password: "${file:/opt/kafka/external-configuration/hm-elasticsearch-credentials-volume/elasticsearch-credentials.properties:elasticsearch_keystore_password}"
    elastic.https.ssl.keystore.type: JKS
    elastic.https.ssl.truststore.location: /opt/kafka/external-configuration/hm-elasticsearch-keystore-volume/elasticsearch-keystore.jks
    elastic.https.ssl.truststore.password: "${file:/opt/kafka/external-configuration/hm-elasticsearch-credentials-volume/elasticsearch-credentials.properties:elasticsearch_keystore_password}"
    elastic.https.ssl.truststore.type: JKS
    elastic.https.ssl.protocol: TLSv1.3
    # schema
    schema.ignore: true
    # key
    key.ignore: true
    # value
    value.converter: org.apache.kafka.connect.json.JsonConverter
    value.converter.schemas.enable: false
    behavior.on.null.values: delete