Message duplicty error during indexing Elasticsearch docs

[MichalMaM]

Hey there, We are using Connect to sink documents from kafka topics to Elasticsearch and I got error "version conflict, document already exists" and connector crashed on this error. Is it possible to add setting to ignore these messages if document already exist?

[mihaitodor]

Hey @MichalMaM, could you please share more details about how you're using Connect? Are you using the legacy elasticsearch output or the new elasticsearch_v8 one? Also, when you say it crashed, do you mean the process died? If yes, could you please provide the log output when it crashed?

In terms of config changes, you can try to use the fallback output with elasticsearch_v8 as the first child and drop as the second one. Then you can attach a log processor to the drop output and have in it message: ${! @ fallback_error } to print the error to the logs (or maybe you'll want to set up Kafka instead of drop to act as a dead letter queue.

[MichalMaM]

Hi @mihaitodor thank you for your answer. We're using Connector in kubernetes and our yaml config looks like this:

config:
  input:
    kafka_franz:
      seed_brokers:
        - xxx
      tls:
        enabled: true
      client_id: xxx
      sasl:
        - mechanism: "PLAIN"
          username: xxx
          password: xxx
      consumer_group: xxx
      regexp_topics: true
      topics:
        - xxx_.*
  pipeline:
    processors:
      - schema_registry_decode:
          url: xxx
          basic_auth:
            enabled: true
            username: xxx
            password: xxx
          avro:
            raw_unions: true
      - catch:
        - mapping: xxx
      - mutation: xxx
  output:
    elasticsearch:
      urls:
      - xxx
      basic_auth:
        enabled: true
        password: xxx
        username: xxx
      sniff: false
      action: create
      id: xxx
      index: xxx

I apologize for my mistake, connector doesn't creash, but can not process another messages (and kafka group lag is constantly increasing) because it can not store the same message to elastic and it logs this error:

... "Failed to send message to elasticsearch: failed to send message" ... "version conflict, document already exists" ...

But before this error I see in log first error:

Failed to send message to elasticsearch: Post "https://xxx/_bulk": context deadline exceeded (Client.Timeout exceeded while awaiting headers)

Your suggestion to use fallback could be a good way for us, can you paste here simple example pls?

[mihaitodor]

Got you, yeah, you'll have to explicitly handle those failed messages and fallback is one option to do so. Please consult the docs for examples. Here's how the drop version would look like:

output:
  fallback:
    - elasticsearch_v8: # Please use this one instead of the legacy `elasticsearch`
        urls:
          - xxx
        basic_auth:
          enabled: true
          password: xxx
          username: xxx
        sniff: false
        action: create
        id: xxx
        index: xxx
    - drop: {}
      processors:
        - log:
            message: ${! @ fallback_error }

[MichalMaM]

Thank you. We will try it.