retry upsert on recoverable error.

Author: photom

lib/fluent/plugin/elasticsearch_error_handler.rb

@@ -60,6 +60,8 @@ def handle_error(response, tag, chunk, bulk_message_count, extracted_values)
         write_operation = @plugin.write_operation
       elsif INDEX_OP == @plugin.write_operation && item.is_a?(Hash) && item.has_key?(CREATE_OP)
         write_operation = CREATE_OP
+      elsif UPSERT_OP == @plugin.write_operation && item.is_a?(Hash) && item.has_key?(UPDATE_OP)
+        write_operation = UPDATE_OP
       elsif item.nil?
         stats[:errors_nil_resp] += 1
         next

test/plugin/test_elasticsearch_error_handler.rb

@@ -7,9 +7,10 @@ class TestElasticsearchErrorHandler < Test::Unit::TestCase
 
   class TestPlugin
     attr_reader :log
-    attr_reader :write_operation, :error_events
+    attr_reader :error_events
     attr_accessor :unrecoverable_error_types
     attr_accessor :log_es_400_reason
+    attr_accessor :write_operation
     def initialize(log, log_es_400_reason = false)
       @log = log
       @write_operation = 'index'
@@ -522,4 +523,125 @@ def test_unrecoverable_error_included_in_responses
 
   end
 
+  def test_retry_error_upsert
+    @plugin.write_operation = 'upsert'
+    records = []
+    error_records = Hash.new(false)
+    error_records.merge!({0=>true, 4=>true, 9=>true})
+    10.times do |i|
+      records << {time: 12345, record: {"message"=>"record #{i}","_id"=>i,"raise"=>error_records[i]}}
+    end
+    chunk = MockChunk.new(records)
+
+    response = parse_response(%({
+      "took" : 1,
+      "errors" : true,
+      "items" : [
+        {
+          "update" : {
+            "_index" : "foo",
+            "_type"  : "bar",
+            "_id" : "1",
+            "status" : 201
+          }
+        },
+        {
+          "update" : {
+            "_index" : "foo",
+            "_type"  : "bar",
+            "_id" : "2",
+            "status" : 500,
+            "error" : {
+              "type" : "some unrecognized type",
+              "reason":"unrecognized error"
+            }
+          }
+        },
+        {
+          "update" : {
+            "_index" : "foo",
+            "_type"  : "bar",
+            "_id" : "3",
+            "status" : 409,
+            "error" : {
+              "type":"version_conflict_engine_exception",
+              "reason":"document already exists"
+            }
+          }
+        },
+        {
+          "update" : {
+            "_index" : "foo",
+            "_type"  : "bar",
+            "_id" : "5",
+            "status" : 500,
+            "error" : {
+              "reason":"unrecognized error - no type field"
+            }
+          }
+        },
+        {
+          "update" : {
+            "_index" : "foo",
+            "_type"  : "bar",
+            "_id" : "6",
+            "status" : 400,
+            "error" : {
+              "type" : "mapper_parsing_exception",
+              "reason":"failed to parse"
+            }
+          }
+        },
+        {
+          "update" : {
+            "_index" : "foo",
+            "_type"  : "bar",
+            "_id" : "7",
+            "status" : 400,
+            "error" : {
+              "type" : "some unrecognized type",
+              "reason":"unrecognized error"
+            }
+          }
+        },
+        {
+          "update" : {
+            "_index" : "foo",
+            "_type"  : "bar",
+            "_id" : "8",
+            "status" : 500,
+            "error" : {
+              "type" : "some unrecognized type",
+              "reason":"unrecognized error"
+            }
+          }
+        }
+      ]
+    }))
+
+    begin
+      failed = false
+      dummy_extracted_values = []
+      @handler.handle_error(response, 'atag', chunk, response['items'].length, dummy_extracted_values)
+    rescue Fluent::Plugin::ElasticsearchErrorHandler::ElasticsearchRequestAbortError, Fluent::Plugin::ElasticsearchOutput::RetryStreamError=>e
+      failed = true
+      records = [].tap do |records|
+        next unless e.respond_to?(:retry_stream)
+        e.retry_stream.each {|time, record| records << record}
+      end
+      puts records
+      assert_equal 3, records.length
+      assert_equal 2, records[0]['_id']
+      # upsert is retried in case of conflict error.
+      assert_equal 3, records[1]['_id']
+      assert_equal 8, records[2]['_id']
+      error_ids = @plugin.error_events.collect {|h| h[:record]['_id']}
+      assert_equal 3, error_ids.length
+      assert_equal [5, 6, 7], error_ids
+      @plugin.error_events.collect {|h| h[:error]}.each do |e|
+        assert_true e.respond_to?(:backtrace)
+      end
+    end
+    assert_true failed
+  end
 end