Aiven-Open / Aiven-Open/opensearch-connector-for-apache-kafka

DLQ not working

Abierto
#141 1 comentario 0 reacciones 0 asignados Ver en GitHub
Lenguaje dominante
Java
Estrellas
95
Forks
61
Merge medio
11 h 52 min
PR fusionados (30 d)
7

Descripción

Hello

Thank you for providing a plugin for Opensearch as we recently migrated our clusters from ES to OS and this plugin has been a great help us to continue pushing data to our cluster. However we noticed a problem where the records not matching the OpenSearch schema are not published to our DLQ and they are simply being ignored and lost. Please see below the connector settings that works well with ES connector. i.e the messages are published to DLQ

```
{
"config": {
"admin.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler",
"admin.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
"admin.sasl.mechanism": "AWS_MSK_IAM",
"admin.security.protocol": "SASL_SSL",
"behavior.on.malformed.documents": "WARN",
"config.providers": "secretManager",
"config.providers.secretManager.class": "com.github.jcustenborder.kafka.config.aws.SecretsManagerConfigProvider",
"config.providers.secretManager.param.aws.region": "us-east-1",
"config.providers.secretManager.param.secret.prefix": "",
"connection.compression": true,
"connection.password": "${secretManager:test/ssl/kafka/opensearch:password}",
"connection.url": "https://test-opensearch.comp.io",
"connection.username": "${secretManager:test/ssl/kafka/opensearch:username}",
"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"consumer.reconnect.backoff.max.ms": 5000,
"consumer.reconnect.backoff.ms": 1000,
"consumer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler",
"consumer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
"consumer.sasl.mechanism": "AWS_MSK_IAM",
"consumer.security.protocol": "SASL_SSL",
"elastic.https.ssl.truststore.location": "${secretManager:test/ssl/kafka/others:cert_trust_store}",
"elastic.security.protocol": "SSL",
"errors.deadletterqueue.context.headers.enable": "true",
"errors.deadletterqueue.topic.name": "test.ssl.app.opensearch.dlq",
"errors.deadletterqueue.topic.replication.factor": "2",
"errors.tolerance": "all",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"producer.reconnect.backoff.max.ms": 5000,
"producer.reconnect.backoff.ms": 1000,
"producer.sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler",
"producer.sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
"producer.sasl.mechanism": "AWS_MSK_IAM",
"producer.security.protocol": "SASL_SSL",
"read.timeout.ms": 60000,
"schema.ignore": true,
"tasks.max": 1,
"topics": "test.ssl.app.opensearch",
"transforms": "renameTopicToIndex",
"transforms.renameTopicToIndex.regex": ".*",
"transforms.renameTopicToIndex.replacement": "test.ssl.app.opensearch",
"transforms.renameTopicToIndex.type": "org.apache.kafka.connect.transforms.RegexRouter",
"type.name": "_doc",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"value.converter.basic.auth.credentials.source": "USER_INFO",
"value.converter.schema.registry.basic.auth.user.info": "${secretManager:test/ssl/kafka/schema:credentials}",
"value.converter.schema.registry.url": "${secretManager:test/ssl/kafka/others:schema_reg_url}"
},
"name": "test.ssl.app.opensearch.opensearch"
}
```

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.