Can't get working at all / config woes?

[confluentinc]

Can't get working at all / config woes?

[costa]

I've tried many things, and following https://sematext.com/blog/2017/03/06/kafka-connect-elasticsearch-how-to/ was one of them.
The connector does not seem to get loaded, the log is stuck on:

INFO Creating connector elasticsearch-sink of type io.confluent.connect.elasticsearch.ElasticsearchSinkConnector (org.apache.kafka.connect.runtime.Worker:159)

[WillemZegers]

I've followed the exact same tutorial and also have this problem. Any help or tips would be greatly appreciated.

[ewencp]

@costa Avro is a serialization format (think of JSON, protocol buffers, thrifts, etc). Just to get some data flowing, I'd suggest following the quickstart directly, including using kafka-avro-console-producer. Once you have that working, it will make sense to make tweaks to adjust it to your setup.

The output you provided is actually only a partial listing of all connector configs. There are some shared configs that work for all connectors and some that are only used for specific connectors like the ES connector. You're only seeing some of them. This also explains the confusion around topics, which is a common required field for any sink connector since it needs to know where to load data from. If you don't specify any input topics, it won't have any work to do, so you wouldn't see any output in Elasticsearch.

If you're still running into problems, can you show the full config you're using and more of the log output?

[costa]

Hey, thanks, I'm currently using the datamountaineer's ElasticSinkConnector, but I'll check back the Confluent's if needed or advised. I still think some proven up-to-date Getting Started will help the community greatly here, otherwise, you may close this issue.

[tslater]

@ewencp Is Avro required to use kafka-connect-elasticsearch?

[costa]

Just letting know that I've managed to bring my setup to life, and I'm using this connector over the DataMountaineer's because the latter is binary incompatible with ES5 (this connector uses HTTP API).

I've created a configurator component which configures the DM's fast-data-dev container like this:

schemata_indices.each do |schema, index|
  begin
    $stderr.puts "INFO Creating #{index} ES index..."
    RestClient.put(
      "#{ELASTIC_API_URI}/#{index}",
      JSON.parse(File.read "#{index}-es-index.json").to_json,  # NOTE validates JSON
      **content_headers
    )
  rescue Errno::ECONNREFUSED
    $stderr.puts "WARNING Elastic is down, retrying in 5..."
    sleep 5
    retry
  end
  subject = "#{schema}-value"
  begin
    $stderr.puts "INFO Waiting for #{subject} to register..."
    raise RestClient::NotFound  if
      JSON.parse(
        RestClient.get(
          "#{KAFKA_REGISTRY_URI}/subjects/#{subject}/versions",
          **content_headers
        )
      ).empty?
  rescue Errno::ECONNREFUSED
    $stderr.puts "WARNING Schema registry is down, retrying in 5..."
    sleep 5
    retry
  rescue RestClient::NotFound
    $stderr.puts "WARNING Schema is not registered yet, retrying in 5..."
    sleep 5
    retry
  end
  topic = schema
  connector_name = "#{topic}-#{KAFKA_ELASTIC_CONNECT_CLASS.split('.')[-1]}"
  begin
    $stderr.puts "INFO Connecting elasticsearch with #{connector_name}..."
    RestClient.post(
      "#{KAFKA_MANAGER_URI}/kafka-connect/connectors",
      {
        'name' => connector_name,
        'config' => {
          'connector.class' => KAFKA_ELASTIC_CONNECT_CLASS,
          #ESC 'connect.elastic.url' => ELASTIC_SEARCH_URI,
          #ESC 'connect.elastic.cluster.name' => 'elasticsearch',
          'tasks.max' => "#{KAFKA_CONNECT_TASKS_MAX}",
          'topics' => topic,
          #ESC 'connect.elastic.sink.kcql' => "INSERT INTO #{index} SELECT * FROM #{topic}"
          'topic.index.map' => "#{topic}:#{index}",  #ESC
          'connection.url' => ELASTIC_API_URI,  #ESC
          'type.name' => topic,  #ESC
          'key.ignore' => true,  #ESC
          'schema.ignore' => true  #ESC
        }
      }.to_json,
      **content_headers
    )
  rescue RestClient::Conflict
    $stderr.puts "WARNING Elasticsearch seems to have been connected already, disconnecting..."
    RestClient.delete "#{KAFKA_MANAGER_URI}/kafka-connect/connectors/#{connector_name}"
    retry
  rescue Errno::ECONNREFUSED, RestClient::BadGateway
    $stderr.puts "WARNING Kafka Manager is down, retrying in 5..."
    sleep 5
    retry
  end

Lines marked with ESC are different with ElasticSinkConnector.