version: '2'

services:
    kafka-connect-01:
        image: confluentinc/cp-kafka-connect
        container_name: kafka-connect-debezium
        ports: 
            - 8083:8083
        environment:
            CONNECT_BOOTSTRAP_SERVERS: "b-1.dev-managed-kafka-neo4.w2oo2u.c7.kafka.us-east-1.amazonaws.com:9094,b-2.dev-managed-kafka-neo4.w2oo2u.c7.kafka.us-east-1.amazonaws.com:9094,b-3.dev-managed-kafka-neo4.w2oo2u.c7.kafka.us-east-1.amazonaws.com:9094"
            CONNECT_REST_ADVERTISED_HOST_NAME: 'kafka-connect-debezium'
            CONNECT_REST_PORT: 8083
            CONNECT_GROUP_ID: kafka-connect-group-debezium-v01
            CONNECT_CONFIG_STORAGE_TOPIC: _kafka-connect-group-debezium-v01-configs
            CONNECT_OFFSET_STORAGE_TOPIC: _kafka-connect-group-debezium-v01-offsets
            CONNECT_STATUS_STORAGE_TOPIC: _kafka-connect-group-debezium-v01-status
            CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
            CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
            CONNECT_INTERNAL_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
            CONNECT_INTERNAL_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
            CONNECT_PLUGIN_PATH: '/usr/share/java,/usr/share/confluent-hub-components/'
            CONNECT_SECURITY_PROTOCOL: SSL
            CONNECT_SSL_TRUSTSTORE_LOCATION: /usr/lib/jvm/zulu11-ca/lib/security/cacerts
            CONNECT_CONSUMER_SECURITY_PROTOCOL: "SSL"
            CONNECT_CONSUMER_SSL_TRUSTSTORE_LOCATION: /usr/lib/jvm/zulu11-ca/lib/security/cacerts
            CONNECT_PRODUCER_SECURITY_PROTOCOL: "SSL"
            CONNECT_PRODUCER_SSL_TRUSTSTORE_LOCATION: /usr/lib/jvm/zulu11-ca/lib/security/cacerts
            CONNECT_CONSUMER_REQUEST_TIMEOUT_MS: "20000"
            CONNECT_CONSUMER_RETRY_BACKOFF_MS: "500"
            CONNECT_LOG4J_ROOT_LOGLEVEL: "INFO"

        command:
            - bash
            - -c
            - |
                echo "Installing connector plugins"
                confluent-hub install --no-prompt debezium/debezium-connector-mysql:latest
                #
                echo "Launching Kafka Connect worker"
                /etc/confluent/docker/run &
                #
                echo "Waiting for Kafka Connect to start listening on localhost:8083 ⏳"
                while : ; do
                    curl_status=$$(curl -s -o /dev/null -w %{http_code} http://localhost:8083/connectors)
                    echo -e $$(date) " Kafka Connect listener HTTP state: " $$curl_status " (waiting for 200)"
                    if [ $$curl_status -eq 200 ] ; then
                    break
                    fi
                    sleep 5
                done
                #
                echo -e "\n--\n+> Creating Kafka Connect Debezium Mysql Source"
                curl -i -X PUT -H  "Content-Type:application/json" \
                    http://localhost:8083/connectors/mysql-debezium-source/config \
                    -d '
                        {
                            "connector.class": "io.debezium.connector.mysql.MySqlConnector",
                            "database.hostname": "build.dddb.devorch.com",
                            "database.port": "3306",
                            "database.user": "ssavva",
                            "database.password": "***",
                            "database.server.id": "184070",
                            "database.server.name": "dev_direct_delivery",
                            "table.whitelist": "direct_delivery.asset_type",
                            "database.history.kafka.bootstrap.servers": "b-1.dev-managed-kafka-neo4.w2oo2u.c7.kafka.us-east-1.amazonaws.com:9094,b-2.dev-managed-kafka-neo4.w2oo2u.c7.kafka.us-east-1.amazonaws.com:9094,b-3.dev-managed-kafka-neo4.w2oo2u.c7.kafka.us-east-1.amazonaws.com:9094",
                            "database.history.kafka.topic": "_dbhistory_dd",
                            "include.schema.changes": "false",
                            "database.history.producer.security.protocol": "SSL",
                            "database.history.producer.ssl.truststore.location": "/usr/lib/jvm/zulu11-ca/lib/security/cacerts",
                            "database.history.consumer.security.protocol": "SSL",
                            "database.history.consumer.ssl.truststore.location": "/usr/lib/jvm/zulu11-ca/lib/security/cacerts"
                        }
                    '
                #
                sleep infinity