fixing to work with cloudwatch logs kinesis subscriptions which gzip …

…the records
haroldwoo · May 11, 2016 · 07eaaf0 · 07eaaf0
1 parent d33a4fc
commit 07eaaf0
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/lib/logstash/inputs/kinesis/worker.rb b/lib/logstash/inputs/kinesis/worker.rb
@@ -45,7 +45,10 @@ def checkpoint(checkpointer)
   end
 
   def process_record(record)
-    raw = @decoder.decode(record.getData).to_s
+    #raw = @decoder.decode(record.getData).to_s
+    is = com.fasterxml.jackson.databind.util::ByteBufferBackedInputStream.new(record.getData)
+    gzip = Java::java.util.zip::GZIPInputStream.new(is)
+    raw = gzip.to_io.read
     @codec.decode(raw) do |event|
       @decorator.call(event)
       @output_queue << event