version: '2'

services:
    kafka-connect-01:
        image: confluentinc/cp-kafka-connect
        container_name: kafka-connect-01
        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-01'
            CONNECT_REST_PORT: 8083
            CONNECT_GROUP_ID: kafka-connect-group-01-v04
            CONNECT_CONFIG_STORAGE_TOPIC: _kafka-connect-group-01-v04-configs
            CONNECT_OFFSET_STORAGE_TOPIC: _kafka-connect-group-01-v04-offsets
            CONNECT_STATUS_STORAGE_TOPIC: _kafka-connect-group-01-v04-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_CONFIG_PROVIDERS: 'file'
            CONNECT_CONFIG_PROVIDERS_FILE_CLASS: 'org.apache.kafka.common.config.provider.FileConfigProvider'
            CONNECT_CONSUMER_SECURITY_PROTOCOL: "SSL"
            CONNECT_CONSUMER_SSL_TRUSTSTORE_LOCATION: /usr/lib/jvm/zulu11-ca/lib/security/cacerts
            CONNECT_CONSUMER_REQUEST_TIMEOUT_MS: "20000"
            CONNECT_CONSUMER_RETRY_BACKOFF_MS: "500"
        command: 
            - bash 
            - -c 
            - |
                echo "Installing connector plugins"                
                confluent-hub install --no-prompt snowflakeinc/snowflake-kafka-connector:1.4.3
                #
                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 Snowflake sink"
                curl -i -X PUT -H  "Content-Type:application/json" \
                    http://localhost:8083/connectors/sink_snowflake_01/config \
                    -d '{"connector.class":"com.snowflake.kafka.connector.SnowflakeSinkConnector",
                    "tasks.max":1,
                    "topics":"default_node_topic",
                    "snowflake.url.name":"$${file:/data/credentials.properties:SNOWFLAKE_HOST}",
                    "snowflake.user.name":"$${file:/data/credentials.properties:SNOWFLAKE_USER}",
                    "snowflake.user.role":"DEV_ENGINEERING",
                    "snowflake.private.key":"$${file:/data/credentials.properties:SNOWFLAKE_PRIVATE_KEY}",
                    "snowflake.private.key.passphrase":"$${file:/data/credentials.properties:SNOWFLAKE_PRIVATE_KEY_PASSPHRASE}",
                    "snowflake.database.name":"DEV_ENGINEERING",
                    "snowflake.schema.name":"TEST_SCHEMA",
                    "key.converter":"org.apache.kafka.connect.storage.StringConverter",
                    "value.converter":"com.snowflake.kafka.connector.records.SnowflakeJsonConverter"
                    }'
                #
                sleep infinity
        volumes: 
        - .env:/data/credentials.properties