version: '2'

services:
    kafka-connect-06:
        image: confluentinc/cp-kafka-connect
        container_name: kafka-connect-06
        ports:
            - 28083:28083
        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-06'
            CONNECT_REST_PORT: 28083
            CONNECT_GROUP_ID: kafka-connect-group-06-v04
            CONNECT_CONFIG_STORAGE_TOPIC: _kafka-connect-group-06-v04-configs
            CONNECT_OFFSET_STORAGE_TOPIC: _kafka-connect-group-06-v04-offsets
            CONNECT_STATUS_STORAGE_TOPIC: _kafka-connect-group-06-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/zulu-8-amd64/jre/lib/security/cacerts
            CONNECT_CONSUMER_SECURITY_PROTOCOL: "SSL"
            CONNECT_CONSUMER_SSL_TRUSTSTORE_LOCATION:  /usr/lib/jvm/zulu-8-amd64/jre/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 confluentinc/kafka-connect-elasticsearch:latest
                #
                echo "Launching Kafka Connect worker"
                /etc/confluent/docker/run &
                #
                echo "Waiting for Kafka Connect to start listening on localhost:28083 ⏳"
                while : ; do
                    curl_status=$$(curl -s -o /dev/null -w %{http_code} http://localhost:28083/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 elasticsearch sink"
                curl -i -X POST -H  "Content-Type:application/json" \
                    http://localhost:28083/connectors/sink-elastic-06/config \
                    -d '
                    {
                        "name": "sink-elastic-06",
                        "config": {
                          "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
                          "tasks.max": "1",
                          "topics": "pageviews",
                          "connection.url": "https://vpc-dev-mymac-kafka-neo4j-swmab7laolvwt7rp7utfnhtbxm.us-east-1.es.amazonaws.com/",
                          "key.ignore": "false",
                          "type.name": "lp_test",
                          "schema.ignore": "true",
                          "key.converter": "org.apache.kafka.connect.storage.StringConverter",
                          "value.converter": "org.apache.kafka.connect.json.JsonConverter",
                          "value.converter.schemas.enable": "false"
                      }
                    }
                    '
                #
                sleep infinity