ConnectException: Cannot create mapping when using RegexRouter/TimestampRouter SMT

[rmoff]

Following config works fine

{
  "config": {
  "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
  "key.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "key.converter.schema.registry.url": "http://localhost:8081",
  "value.converter.schema.registry.url": "http://localhost:8081",
  "topics": "sqlite-foo",
  "key.ignore": "true",
  "type.name": "type.name=kafka-connect",
  "connection.url": "http://localhost:9200"
  },
  "name": "es-sink-sqlite-foo"
}

But if I add org.apache.kafka.connect.transforms.TimestampRouter SMT:

{
  "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
  "key.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "key.converter.schema.registry.url": "http://localhost:8081",
  "value.converter.schema.registry.url": "http://localhost:8081",
  "topics": "sqlite-foo",
  "key.ignore": "true",
  "type.name": "type.name=kafka-connect",
  "connection.url": "http://localhost:9200",
  "transforms":"routeTS",
  "transforms.routeTS.type":"org.apache.kafka.connect.transforms.TimestampRouter"
}

the Elasticsearch Sink fails:

[2017-07-12 16:06:16,869] ERROR Task es-sink-sqlite-foo-0 threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerSinkTask:455)
org.apache.kafka.connect.errors.ConnectException: Cannot create mapping {"type.name=kafka-connect":{"properties":{"c1":{"type":"long"},"c2":{"type":"string"},"messagetopic":{"type":"string"},"messagesource":{"type":"string"}}}} -- {"root_cause":[{"type":"index_not_found_exception","reason":"no such index","resource.type":"index_or_alias","resource.id":"sqlite-foo-20170712","index_uuid":"_na_","index":"sqlite-foo-20170712"}],"type":"index_not_found_exception","reason":"no such index","resource.type":"index_or_alias","resource.id":"sqlite-foo-20170712","index_uuid":"_na_","index":"sqlite-foo-20170712"}
        at io.confluent.connect.elasticsearch.Mapping.createMapping(Mapping.java:65)
        at io.confluent.connect.elasticsearch.ElasticsearchWriter.write(ElasticsearchWriter.java:201)
        at io.confluent.connect.elasticsearch.ElasticsearchSinkTask.put(ElasticsearchSinkTask.java:119)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:435)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:251)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:180)
        at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:148)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:146)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:190)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)
[2017-07-12 16:06:16,871] ERROR Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerSinkTask:456)

Same problem exists with using RegexRouter e.g.:

        "transforms":"dropPrefix",
        "transforms.dropPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.dropPrefix.regex":"sqlite-(.*)",
        "transforms.dropPrefix.replacement":"$1"

It looks like the Sink correctly autocreates indices based on the inbound topic, but if the topic mutates with SMT, it doesn't.

[kkonstantine]

Probably related to: https://issues.apache.org/jira/browse/KAFKA-5567
Can you try with Kafka Connect workers from Apache Kafka trunk repo?

[perryao]

is there a workaround for this?

[jay999123]

Can anyone comment on what is the workaround for this issue? or when can this will be resolved?

[rmoff]

This works fine now as of Apache Kafka 2.0 / Confluent Platform 5.0

[mattferroni]

Hey @rmoff , same problem here :(
Any suggestion on how to handle it with a previous Kafka deploy?

[rmoff]

@mattferroni which version are you running? The options as I see it for you are:

• Create the target Elasticsearch indices another way (manually, or scripted)
• Upgrade your Kafka Connect version

[mattferroni]

Running Kafka 1.0, I finally scripted it (manually) -- thanks anyway!

[gunnarmorling]

FWIW, I'm still seeing this issue with Kafka 2.0 and kafka-connect-elasticsearch-5.0.0.jar. I'd like to apply RegexRouter, but it seems to be ignored. In my case that's causing things to not work at all, as the inbound topic name isn't a valid index name:

Invalid index name [dbserver1.inventory.Hike], must be lowercase\",\"index_uuid\":\"_na_\",\"index\":\"dbserver1.inventory.Hike\"} at 
io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.createIndices(JestElasticsearchClient.java:238) at 
io.confluent.connect.elasticsearch.ElasticsearchWriter.createIndicesForTopics(ElasticsearchWriter.java:330) at 
io.confluent.connect.elasticsearch.ElasticsearchSinkTask.open(ElasticsearchSinkTask.java:157)

[rmoff]

Hey @gunnarmorling :) Can you share your full sink config?

[gunnarmorling]

@rmoff, of course, here it is:

{
    "name": "elastic-sink",
    "config": {
        "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
        "tasks.max": "1",
        "topics": "Customers",
        "connection.url": "http://elastic:9200",
        "transforms": "unwrap,key,route",
        "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
        "transforms.key.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
        "transforms.key.field": "id",
        "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.route.regex": "Customers",
        "transforms.route.replacement": "customers",
        "key.ignore": "false",
        "type.name": "customer"
    }
}

I have also an executable reproducer, if you'd like to take a look: https://github.com/gunnarmorling/debezium-examples/tree/kafka-connect-elasticsearch-99/unwrap-smt

Check out that branch and follow the steps in https://github.com/gunnarmorling/debezium-examples/blob/kafka-connect-elasticsearch-99/unwrap-smt/reproducer.md. Registering the sink connector fails due to an upper-case index name, which should have been transformed by means of the regex router. Just verified it's using Kafka 2.0 and ES sink 5.0.0.

Maybe I'm also doing some silly mistake here?

[rmoff]

@gunnarmorling I reproduced your problem using the steps provided.

However, I think it's different from the original problem in this ticket. Whereas before, Kafka Connect was applying the RegEx SMT but not creating the target ES index, what seems to be happening in your case is that it's simply not applying the RegEx SMT.

You can see this from the Elasticsearch error, where it is trying to use the original topic name in the index Invalid index name [Customers].

The practical step if you just want to force to lowercase is use the topic.index.map config option:

    "topic.index.map":"Customers:customers",

But, it's probably worth raising a separate ticket to address the issue with RegexRouter that you're seeing anyway.