Unexpected error when registering ES connect

[confluentinc]

Unexpected error when registering ES connect

[ewencp]

This doesn't really look specific to the ES connector. It looks from the stack trace like we're somehow getting a null config map somehow. Is the MQTT connector required to reproduce this? Is it reproducible just by running the single command you provided on a new Connect cluster?

[srijiths]

@ewencp Thank you . Here are the steps which i executed
connect-distributed /etc/kafka/connect-distributed.properties

In connect-distributed properties , i changed below configs

bootstrap.servers=ALL KAFKA BROKERS LISTED HERE
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false

curl -X POST -H "Content-Type: application/json" --data '{"name": "sensor-mqtt-connect", "config": {"connector.class":"com.evokly.kafka.connect.mqtt.MqttSourceConnector", "tasks.max":"1", "kafka.topic":"sensor-input","mqtt.server_uris":"tcp://XXXXXX:1883","mqtt.topic":"sensor","mqtt.qos":"1","message_processor_class":"com.evokly.kafka.connect.mqtt.sample.DumbProcessor"}}'  http://XXXXX:8083/connectors

Then i placed kafka-connect-es jar under /usr/share/java/kafka-connect-es
Then i registered ES Connect

curl -X POST -H "Content-Type: application/json" --data '{"name":"elasticsearch-sink","config":{"connector.class":"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector","type.name":"autodesk","tasks.max":"1","topics":"sensor-output","key.ignore":"true","schema.ignore":"true","connection.url":"http://XXXXXXXX:9200"}}'  http://XXXXXXX:8083/connectors

Thats it. Did i miss any steps in between ? Thank You

[ewencp]

@srijiths That seems fine, but I can't reproduce the issue with the sequence of commands you provided. What version of Kafka/Connect are you running? I doubt the combination of MQTT and ES connectors is the issue since the stacktrace doesn't seem like it has anything to do with a conflict.

[srijiths]

@ewencp i am using confluvent 3.0.0 version. I tried to register above ES connector again.

When i observed it has WARN on Concurrent Modification Exception from BulkProcessor class followed by JAVA HEAP SIZE exception. I have 4G alloted to JAVA.
Then it follows a series of exceptions and kafka connect , rest evrything stopped. Since its a lengthy stack trace , created a gist. Please see here
https://gist.github.com/srijiths/459725df0137c881090f7a03565f784f

Here is the MQTT connector which i am using. https://github.com/evokly/kafka-connect-mqtt

I tested ES connect with no MQTT connectors running. Still the same issues. Any pointers please ?

[shikhar]

@srijiths this appears to be the root of the problem: java.lang.OutOfMemoryError: Java heap space

1. It's possible that the ES connector is buffering so much data that you are running out of heap because of that. max.buffered.records defaults to 100000 which is rather high, you could try lowering that.
2. Are you running with customized heap settings? Using the standard scripts, you'd end up with KAFKA_HEAP_OPTS=-Xmx256M .You may want to consider growing that to say 512M if lowering the buffered records config does not help.

[shikhar]

Please reopen if there are any unexpected issues here. The defaults for max.buffered.records and batch.size have been significantly lowered.