title: Logstash agent not indexing anymore
body: <p>I have a Logstash instance running as a service that reads from Redis and outputs to Elasticsearch. I just noticed there was nothing new in Elasticsearch for the last few days, but the Redis lists were increasing.</p>
  
  <p>Logstash log was filled with 2 errors repeated for thousands of lines:</p>
  
  <pre><code>:message=&gt;"Got error to send bulk of actions"
  :message=&gt;"Failed to flush outgoing items"
  </code></pre>
  
  <p>The reason being:  </p>
  
  <pre><code>{"error":"IllegalArgumentException[Malformed action/metadata line [107], expected a simple value for field [_type] but found [START_ARRAY]]","status":500}, 
  </code></pre>
  
  <p>Additionally, trying to stop the service failed repeatedly, I had to kill it. Restarting it emptied the Redis lists and imported everything to Elasticsearch. It seems to work ok now.  </p>
  
  <p>But I have no idea how to prevent that from happening again. The mentioned <code>type</code> field is set as a string for each input directive, so I don't understand how it could have become an array.<br>
  What am I missing?</p>
  
  <p>I'm using Elasticsearch 1.7.1 and Logstash 1.5.3. The <code>logstash.conf</code> file looks like this:</p>
  
  <pre><code>input {
    redis {
      host =&gt; "127.0.0.1"
      port =&gt; 6381
      data_type =&gt; "list"
      key =&gt; "b2c-web"
      type =&gt; "b2c-web"
      codec =&gt; "json"
    }
    redis {
      host =&gt; "127.0.0.1"
      port =&gt; 6381
      data_type =&gt; "list"
      key =&gt; "b2c-web-staging"
      type =&gt; "b2c-web-staging"
      codec =&gt; "json"
    }
  
      /* other redis inputs, only key/type variations */
  }
  filter {
    grok {
      match =&gt; ["msg", "Cache hit %{WORD:query} in %{NUMBER:hit_total:int}ms. Network: %{NUMBER:hit_network:int} ms.     Deserialization %{NUMBER:hit_deserial:int}"]
      add_tag =&gt; ["cache_hit"]
      tag_on_failure =&gt; []
    }
    /* other groks, not related to type field */
  }
  output {
    elasticsearch {
      host =&gt; "[IP]"
      port =&gt; "9200"
      protocol=&gt; "http"
      cluster =&gt; "logstash-prod-2"
    }
  }
  </code></pre>