aiven-open/elasticsearch-connector-for-apache-kafka

Apache Kafka Connect sink connector for Elasticsearch

elasticsearch-connector-for-apache-kafka's Issues

Deprecate this connecor in favor of the OpenSearch one

Given the change in licensing in Elasticsearch, we should deprecate this in favor of the OpenSearch equivalent.

Reformatting ES Client wrapper

Carry nit suggestions from #94

Support disabling of hostname verification

To use on-premise clusters with Aiven Kafka Connect, users may have self-signed certificates. Adding support in the config for verification to be disabled would help with connections.

Add Support For Upsert

Hi all!

We're trying to update an ElasticSearch document using multiple processes - each updating a part of the document. We do it by sending different messages to the same topic, expecting ElasticSearch to create the full document. Unfortunately, this is not the case. We see that the document is constantly being replaced by the most recent version.

Is there support for upserting/updating documents? Is it planned?

Thanks!

Aiven Connector issue with Elastic Search Version 7.4.0

Hello,

I deployed Elastic search version 7.4.0 on Azure, and I am using Aiven Kafka + Aiven Schema Registry + Aiven Kafka Connect for copying my messages from Kafka topic to the deployed Elastic Search.

connector configuration:

{
    "name": "engagement-aggregate-sink",
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "24",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "transforms": "ExtractTenantId,IndexRouter",
    "errors.retry.timeout": "0",
    "errors.retry.delay.max.ms": "60000",
    "errors.tolerance": "all",
    "errors.log.enable": "true",
    "errors.log.include.messages": "true",
    "topics": "customer-journey-engagement-aggregate",
    "errors.deadletterqueue.topic.name": "customer-journey-engagement-aggregate-deadletterqueue",
    "errors.deadletterqueue.topic.replication.factor": "1",
    "errors.deadletterqueue.context.headers.enable": "true",
    "connection.url": "ES_URL",
    "connection.username":"USERNAME",
    "connection.password":"PASSWORD",
    "batch.size": "10000",
    "max.in.flight.requests": "3",
    "max.buffered.records": "1000",
    "linger.ms": "100",
    "flush.timeout.ms": "20000",
    "max.retries": "10",
    "retry.backoff.ms": "1000",
    "connection.timeout.ms": "1000",
    "read.timeout.ms": "1000",
    "type.name": "_doc",
    "key.ignore": "false",
    "schema.ignore": "true",
    "compact.map.entries": "false",
    "drop.invalid.message": "true",
    "behavior.on.null.values": "ignore",
    "behavior.on.malformed.documents": "warn",
    "transforms.IndexRouter.timestamp.format": "YYYYMM",
    "transforms.ExtractTenantId.type": "io.aiven.kafka.connect.transforms.ExtractTopic$Value",
    "transforms.ExtractTenantId.skip.missing.or.null": "false",
    "transforms.ExtractTenantId.field.name": "tenantId",
    "transforms.IndexRouter.topic.format": "engagement-${topic}-${timestamp}",
    "transforms.IndexRouter.type": "org.apache.kafka.connect.transforms.TimestampRouter",
    "value.converter.schema.registry.url": "KAFKA_URL",
    "value.converter.basic.auth.credentials.source":"USER_INFO",
    "value.converter.basic.auth.user.info":"UNM:PWD",
    "value.converter.max.schemas.per.subject": "1000",
    "value.converter.value.subject.name.strategy": "io.confluent.kafka.serializers.subject.RecordNameStrategy",
    "value.converter.auto.register.schemas": "false"
}

Even if I use both the sink connector versions 5.0.3 and 6.0.3 it throws the same exception as shown below:

java.lang.NullPointerException
	at io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.getServerVersion(JestElasticsearchClient.java:179)
	at io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.<init>(JestElasticsearchClient.java:148)
	at io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.<init>(JestElasticsearchClient.java:117)
	at io.confluent.connect.elasticsearch.ElasticsearchSinkTask.start(ElasticsearchSinkTask.java:123)
	at io.confluent.connect.elasticsearch.ElasticsearchSinkTask.start(ElasticsearchSinkTask.java:52)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.initializeAndStart(WorkerSinkTask.java:304)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:195)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:184)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:234)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:834)

If I use the Aiven Managed Elastic Search everything works fine.
Could you please help figure out, what is the issue in using a elastic search service deployed on Azure?

Adopt ES Java API client

Instead of Jest or REST APIs, now deprecated, use latest Java API.

Compilation

Hi,

How to compile it please ?

Do ES 6 supported?

thanks

Support for ES 8.x

I'm getting the below error while trying to insert records to ES 8.x using latest aiven connector

_Encountered an illegal document error when executing batch 267 of 10 records. Ignoring and will not index record. Error was {"root_cause":[{"type":"illegal_argument_exception","reason":"Action/metadata line [1] contains an unknown parameter [_type]"}],"type":"illegal_argument_exception","reason":"Action/metadata line [1] contains an unknown parameter [type]"} (io.aiven.connect.elasticsearch.bulk.BulkProcessor:414)

Connector config has this set

"schema.ignore": "true",
"type.name": "_doc",
"value.converter.schemas.enable": "false",
"value.converter": "org.apache.kafka.connect.json.JsonConverter"