fix race condition which causes redis output to lose some messages in…

… batch mode

Scenario:
  - @flush_thread starts writing to redis (@redis.rpush)
  - main thread adds a new message to @pending
  - @flush_thread deletes this message (@pending.delete)

lib/logstash/outputs/redis.rb

@@ -150,7 +150,9 @@ def receive(event)
     return unless output?(event)
 
     if @batch
-      @pending[event.sprintf(@key)] << event.to_json
+      @pending_mutex.synchronize do
+        @pending[event.sprintf(@key)] << event.to_json
+      end
       process_pending
       return
     end

spec/outputs/redis.rb

@@ -76,6 +76,55 @@
     end # agent
   end
 
+  describe "batch mode" do
+    key = 10.times.collect { rand(10).to_s }.join("")
+    event_count = 200000
+
+    config <<-CONFIG
+      input {
+        generator {
+          message => "hello world"
+          count => #{event_count}
+          type => "generator"
+        }
+      }
+      output {
+        redis {
+          host => "127.0.0.1"
+          key => "#{key}"
+          data_type => list
+          batch => true
+          batch_timeout => 5
+          timeout => 5
+        }
+      }
+    CONFIG
+
+    agent do
+      # we have to wait for teardown to execute & flush the last batch.
+      # otherwise we might start doing assertions before everything has been
+      # sent out to redis.
+      sleep 2
+
+      redis = Redis.new(:host => "127.0.0.1")
+
+      # The list should contain the number of elements our agent pushed up.
+      insist { redis.llen(key) } == event_count
+
+      # Now check all events for order and correctness.
+      event_count.times do |value|
+        id, element = redis.blpop(key, 0)
+        event = LogStash::Event.new(JSON.parse(element))
+        insist { event["sequence"] } == value
+        insist { event.message } == "hello world"
+        insist { event.type } == "generator"
+      end
+
+      # The list should now be empty
+      insist { redis.llen(key) } == 0
+    end # agent
+  end
+
   describe "converts US-ASCII to utf-8 without failures" do
     key = 10.times.collect { rand(10).to_s }.join("")