# Full pipeline Logstash with dead letter queue

**URL:** <https://discuss.elastic.co/t/full-pipeline-logstash-with-dead-letter-queue/239855>\
**Category:** Logstash\
**Created:** [July 3, 2020, 8:39pm UTC](https://discuss.elastic.co/t/full-pipeline-logstash-with-dead-letter-queue/239855 "2020-07-03T20:39:06Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![Altamir\_Dias](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/altamir_dias/32/69586_2.png) [@Altamir\_Dias](https://discuss.elastic.co/u/Altamir_Dias)\
**Post date:** [July 3, 2020, 8:39pm UTC](https://discuss.elastic.co/t/full-pipeline-logstash-with-dead-letter-queue/239855/1 "2020-07-03T20:39:07Z")

</div>

Hello Guys,

I have a pipeline in my ELK that get data from Kafka and send Elasticsearch. But I don't know how I can set the dead letter queue. My goal is every errors be send a topic called dlq+date.

I tried several solutions, but not work for me.

Someone can help me?

> > Logstash version 7.7.1

```auto
This is my logstash.conf (without dlq)
    input{
            kafka{
                    id => elkkfaplp1
                    group_id => "kafka1"
                    topics_pattern => ".*"
                    bootstrap_servers => "IP:PORT"
                    sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='${KK_USR}' password='${KK_PWD}';"
                    security_protocol => SASL_PLAINTEXT
                    sasl_mechanism => PLAIN
                    decorate_events => true
                    codec => "json"
                    auto_offset_reset => "earliest"
                    metadata_max_age_ms => 500
            }
            kafka{
                    id => elkkfaplp2
                    group_id => "kafka2"
                    topics_pattern => ".*"
                    bootstrap_servers => "IP2:PORT"
                    sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='${KK_USR}' password='${KK_PWD}';"
                    security_protocol => SASL_PLAINTEXT
                    sasl_mechanism => PLAIN
                    decorate_events => true
                    codec => "json"
                    auto_offset_reset => "earliest"
                    metadata_max_age_ms => 500
            }
    }
    filter{
            mutate {
                    #lowercase => ["[@metadata][kafka][topic]" ]
                    strip => [wildcard]
            }
            if [wildcard] != "" {
                    mutate {
                            add_field => { "new_field" => "%{wildcard}" }
                            remove_field => [wildcard]
                    }

                    json {
                            source => [new_field]
                    }

                    mutate {
                            remove_field => [new_field]
                    }
            } else {
                    mutate {
                            remove_field => [wildcard]
                    }
            }
    }
    output {

            if [deadletter] != "" {
                    elasticsearch{
                            id => energisa_cluster1
                            hosts => ["https://elkesaplp1.comp.com.br", "https://elkesaplp2.comp.com.br", "https://elkesaplp3.comp.com.br"]
                            user => "${ES_USR}"
                            password => "${ES_PWD}"
                            cacert => "/etc/logstash/comp-ca.pem"
                            ssl => true
                            ssl_certificate_verification => false

                            index => "comp-%{[@metadata][kafka][topic]}-%{+YYYY.MM.dd}"
                    }
            }
    }

```

---

<div class="post-metadata">

**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\
**Post date:** [July 3, 2020, 9:59pm UTC](https://discuss.elastic.co/t/full-pipeline-logstash-with-dead-letter-queue/239855/2 "2020-07-03T21:59:15Z")

</div>

You do not need to make any changes to your logstash.conf to enable dead letter queues. They are [enabled](https://www.elastic.co/guide/en/logstash/current/dead-letter-queues.html) by making a change in logstash.yml. If logstash gets a 400 or 404 error when it tries to index the event then it will write the event to the dead letter queue, which is one or more files.

You would then run another logstash instance with a dead\_letter\_queue input plugin configured, which would read those files and do whatever you want with the events. It sounds like you want to write them to a kafka topic. You could certainly do that using a kafka output.

---

<div class="post-metadata">

**Author:** ![Altamir\_Dias](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/altamir_dias/32/69586_2.png) [@Altamir\_Dias](https://discuss.elastic.co/u/Altamir_Dias)\
**Post date:** [July 7, 2020, 11:14pm UTC](https://discuss.elastic.co/t/full-pipeline-logstash-with-dead-letter-queue/239855/3 "2020-07-07T23:14:39Z")

</div>

Thank's @Texugo.

Actually, I'm not had undestand Dead Letter Queue yeat. My problem was resolved.  
For this my configuration is similiar with below:

**pipelines.yml**

```
    - pipeline.id: main
      path.config: "/etc/logstash/conf.d/logstash.conf"

    - pipeline.id: deadletter
      path.config: "/etc/logstash/conf.d/logstash_dlq.conf"

```

**logstash.yml**

```
    dead_letter_queue.enable: true

```

**logstash\_dlq.conf** (created by me)  
input{  
dead\_letter\_queue{  
id =\> deadletter  
path =\> "/var/lib/logstash/dead\_letter\_queue"  
commit\_offsets =\> true  
pipeline\_id =\> "main"  
}  
}  
filter {  
mutate {  
add\_field =\> {  
"error\_reason" =\> "%{[@metadata][dead\_letter\_queue][reason]}"  
"plugin\_id" =\> "%{[@metadata][dead\_letter\_queue][plugin\_id]}"  
"plugin\_type" =\> "%{[@metadata][dead\_letter\_queue][plugin\_type]}"  
"entry\_time" =\> "%{[@metadata][dead\_letter\_queue][entry\_time]}"  
}  
}  
}  
output {  
elasticsearch{  
#Elasticsearch Config  
}  
}

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [August 4, 2020, 11:14pm UTC](https://discuss.elastic.co/t/full-pipeline-logstash-with-dead-letter-queue/239855/4 "2020-08-04T23:14:46Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.