fix case when aggregation returns an error (#204)

Author: jsvd

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 4.20.2
+  - fix case when aggregation returns an error [#204](https://github.com/logstash-plugins/logstash-input-elasticsearch/pull/204)
+
 ## 4.20.1
   - Fix license header [#203](https://github.com/logstash-plugins/logstash-input-elasticsearch/pull/203)
 

lib/logstash/inputs/elasticsearch/aggregation.rb

@@ -30,14 +30,15 @@ def retryable(job_name, &block)
           error_details = {:message => e.message, :cause => e.cause}
           error_details[:backtrace] = e.backtrace if logger.debug?
           logger.error("Tried #{job_name} unsuccessfully", error_details)
+          false
         end
 
         def do_run(output_queue)
           logger.info("Aggregation starting")
           r = retryable(AGGREGATION_JOB) do
             @client.search(@agg_options)
           end
-          @plugin.push_hit(r, output_queue, 'aggregations')
+          @plugin.push_hit(r, output_queue, 'aggregations') if r
         end
       end
     end

logstash-input-elasticsearch.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-input-elasticsearch'
-  s.version         = '4.20.1'
+  s.version         = '4.20.2'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Reads query results from an Elasticsearch cluster"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"

spec/inputs/elasticsearch_spec.rb

@@ -1071,24 +1071,33 @@ def wait_receive_request
       }
     end
 
+    let(:client) { Elasticsearch::Client.new }
     before(:each) do
-      client = Elasticsearch::Client.new
-
       expect(Elasticsearch::Client).to receive(:new).with(any_args).and_return(client)
-      expect(client).to receive(:search).with(any_args).and_return(mock_response)
       expect(client).to receive(:ping)
     end
 
     before { plugin.register }
 
     it 'creates the events from the aggregations' do
+      expect(client).to receive(:search).with(any_args).and_return(mock_response)
       plugin.run queue
       event = queue.pop
 
       expect(event).to be_a(LogStash::Event)
       expect(event.get("[total_counter][value]")).to eql 10
       expect(event.get("[empty_counter][value]")).to eql 5
     end
+
+    context "when there's an exception" do
+      before(:each) do
+        allow(client).to receive(:search).and_raise RuntimeError
+      end
+      it 'produces no events' do
+        plugin.run queue
+        expect(queue).to be_empty
+      end
+    end
   end
 
   context "retries" do