using ElasticSearch as sink for Kafka Connect

[Anurag Phadke]

Hi,
I am using the quickstart tutorial at http://docs.confluent.io/3.0.1/cp-docker-images/docs/quickstart.html#getting-started-with-docker-client and now trying to add a ElasticSearch sink. Here are the steps I tried:

1. Build the ES jar off https://github.com/confluentinc/kafka-connect-elasticsearch
I am now trying to add the jar to classpath for kafka-connect

docker run -d\
  --name=kafka-connect \
  --net=host \
  -e CONNECT_BOOTSTRAP_SERVERS=localhost:29092 \
  -e CONNECT_REST_PORT=28082 \
  -e CONNECT_GROUP_ID="quickstart" \
  -e CONNECT_CONFIG_STORAGE_TOPIC="quickstart-config" \
  -e CONNECT_OFFSET_STORAGE_TOPIC="quickstart-offsets" \
  -e CONNECT_STATUS_STORAGE_TOPIC="quickstart-status" \
  -e CONNECT_KEY_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
  -e CONNECT_VALUE_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
  -e CONNECT_INTERNAL_KEY_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
  -e CONNECT_INTERNAL_VALUE_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
  -e CONNECT_REST_ADVERTISED_HOST_NAME="localhost" \
  -e CONNECT_LOG4J_ROOT_LOGLEVEL=DEBUG \
  -e CLASSPATH="/Users/aphadke/Desktop/kafka-connect-elasticsearch-3.2.0-SNAPSHOT.jar" \
  -v /tmp/quickstart/file:/tmp/quickstart \
  confluentinc/cp-kafka-connect:3.0.1
 

Now create the connector
curl -X POST -H "Content-Type: application/json" \
    --data '{"name": "es", "config": {"connector.class":"io.confluent.connect.elasticsearch", "tasks.max":"1", "topics":"quickstart-data", "elasticsearch.cluster.name":"test-cluster","elasticsearch.index.hosts":"127.0.0.1:9300","elasticsearch.index.prefix":"prefix","elasticsearch.document.name":"document","elasticsearch.bulk.size":"100"}}' \
    http://$CONNECT_HOST:28082/connectors

The command returns back with a 200 response, however, any subsequent calls to the connectors return "Connection refused"

curl -X GET http://$CONNECT_HOST:28082/connectors
curl: (7) Failed to connect to 192.168.99.101 port 28082: Connection refused

Any tutorials / pointers on how to add a custom ES sink?

-anurag

[Shikhar Bhushan]

Hi Anurag,

The connector.class specification in the config is incomplete. You seem to be running into https://issues.apache.org/jira/browse/KAFKA-4042 fixed in the upcoming 0.10.1 release where this kind of config error can cause the Connect worker to become unavailable.

Best,

Shikhar

--
You received this message because you are subscribed to the Google Groups "Confluent Platform" group.
To unsubscribe from this group and stop receiving emails from it, send an email to confluent-platf...@googlegroups.com.
To post to this group, send email to confluent...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/confluent-platform/10172247-3c85-4626-9646-fb3553a584de%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Anurag Phadke]

Thanks Shikhar for pointing out the error that the connector.class spec in config was incomplete..
-anurag