title: Logstash Dead Letter Queue Event previously submitted to dead letter queue. Skipping
body: <p>I am trying to enable logstash dead letter queue but, I am getting this warning message and elasticsearch is not receiving the logs from logstash.</p>
  <p><code>[WARN ][org.logstash.common.io.DeadLetterQueueWriter][main] Event previously submitted to dead letter queue. Skipping...</code></p>
  <p>The scenario I am trying,</p>
  <p>Limiting elasticsearch shards so in this case, it can't create a new shard and log will be in the dead letter queue</p>
  <p><code>curl -X PUT 'http://elasticsearch:9200/_cluster/settings' -H &quot;Content-Type: application/json&quot; -d '{ &quot;persistent&quot;: { &quot;cluster.max_shards_per_node&quot;: &quot;3&quot; } }'</code></p>
  <p>end sending log to the logstash from rabbitmq:</p>
  <p><code>{index:&quot;12345&quot;,id:1,message:&quot;hello world&quot;}</code></p>
  <p>Increasing the shard limit:</p>
  <p><code>curl -X PUT 'http://elasticsearch:9200/_cluster/settings' -H &quot;Content-Type: application/json&quot; -d '{ &quot;persistent&quot;: { &quot;cluster.max_shards_per_node&quot;: &quot;10&quot; } }'</code></p>
  <p>when I check the the dead letter queue from logstash api I can see the log in the queue but it's not sending the log to the elasticsearch.</p>
  <p>logstash.yml:</p>
  <pre><code>pipeline:
    batch:
      size: 125
      delay: 5
  dead_letter_queue.enable: true
  </code></pre>
  <p>logstash.conf</p>
  <pre><code>input {
      udp {
          port =&gt; &quot;${INPUT_UDP_PORT}&quot;
          type =&gt; syslog
          codec =&gt; json
      }
      tcp {
          port =&gt; &quot;${INPUT_TCP_PORT}&quot;
          type =&gt; syslog
          codec =&gt; json_lines
      }
      http {
          port =&gt; &quot;${INPUT_HTTP_PORT}&quot;
          codec =&gt; &quot;json&quot;
      }
      rabbitmq {
          host =&gt; &quot;${RABBIT_MQ_HOST}&quot;
          queue =&gt; &quot;log&quot;
          durable =&gt; true
          exchange =&gt; &quot;log&quot;
          key =&gt; &quot;log&quot;
          threads =&gt; 3
          prefetch_count =&gt; 50
          port =&gt; 5672
          user =&gt; &quot;${RABBIT_MQ_USERNAME}&quot;
          password =&gt; &quot;${RABBIT_MQ_PASSWORD}&quot;
          type =&gt; &quot;log&quot;
      }
      dead_letter_queue {
          path =&gt; &quot;/usr/share/logstash/data/dead_letter_queue&quot;
          commit_offsets =&gt; false
          pipeline_id =&gt; &quot;main&quot;
      }
  }
  
  filter {
      if [logger_name] =~ &quot;metrics&quot; {
          kv {
              source =&gt; &quot;message&quot;
              field_split_pattern =&gt; &quot;, &quot;
              prefix =&gt; &quot;metric_&quot;
          }
          mutate {
              convert =&gt; { &quot;metric_value&quot; =&gt; &quot;float&quot; }
              convert =&gt; { &quot;metric_count&quot; =&gt; &quot;integer&quot; }
              convert =&gt; { &quot;metric_min&quot; =&gt; &quot;float&quot; }
              convert =&gt; { &quot;metric_max&quot; =&gt; &quot;float&quot; }
              convert =&gt; { &quot;metric_mean&quot; =&gt; &quot;float&quot; }
              convert =&gt; { &quot;metric_stddev&quot; =&gt; &quot;float&quot; }
              convert =&gt; { &quot;metric_median&quot; =&gt; &quot;float&quot; }
              convert =&gt; { &quot;metric_p75&quot; =&gt; &quot;float&quot; }
              convert =&gt; { &quot;metric_p95&quot; =&gt; &quot;float&quot; }
              convert =&gt; { &quot;metric_p98&quot; =&gt; &quot;float&quot; }
              convert =&gt; { &quot;metric_p99&quot; =&gt; &quot;float&quot; }
              convert =&gt; { &quot;metric_p999&quot; =&gt; &quot;float&quot; }
              convert =&gt; { &quot;metric_mean_rate&quot; =&gt; &quot;float&quot; }
              convert =&gt; { &quot;metric_m1&quot; =&gt; &quot;float&quot; }
              convert =&gt; { &quot;metric_m5&quot; =&gt; &quot;float&quot; }
              convert =&gt; { &quot;metric_m15&quot; =&gt; &quot;float&quot; }
              # No need to keep message field after it has been parsed
              remove_field =&gt; [&quot;message&quot;]
          }
      }
      if [type] == &quot;syslog&quot; {
          mutate {
              add_field =&gt; { &quot;instance_name&quot; =&gt; &quot;%{app_name}-%{host}:%{app_port}&quot; }
          }
      }
      mutate {
          # workaround from https://github.com/elastic/logstash/issues/5115
          add_field =&gt; { &quot;[@metadata][LOGSTASH_DEBUG]&quot; =&gt; &quot;${LOGSTASH_DEBUG:false}&quot; }
      }
      mutate {
        remove_field =&gt; &quot;[geoip][location]&quot;
      }
  }
  
  output {
      if [logger_name] =~ &quot;metrics&quot; {
          elasticsearch {
              hosts =&gt; [&quot;${ELASTICSEARCH_HOST}:${ELASTICSEARCH_PORT}&quot;]
              index =&gt; &quot;metrics-%{+YYYY.MM.dd}&quot;
          }
      } else {
          elasticsearch {
              hosts =&gt; [&quot;${ELASTICSEARCH_HOST}:${ELASTICSEARCH_PORT}&quot;]
              index =&gt; &quot;logs-%{+YYYY.MM.dd}&quot;
          }
      }
      if [@metadata][LOGSTASH_DEBUG] == &quot;true&quot; {
          stdout {
              codec =&gt; rubydebug
          }
      }
  
  }
  </code></pre>
  <p>elasticsearch and logstash versions are: 7.9.2</p>