From 0eef127a7ff04225f4205186032512b7ea81051a Mon Sep 17 00:00:00 2001
From: Rob Bavey <rob.bavey@elastic.co>
Date: Thu, 8 Nov 2018 14:50:57 -0500
Subject: [PATCH] Further fixes to integration and unit tests

Fixes #812
---
 ci/run.sh                                     |  3 +++
 lib/logstash/outputs/elasticsearch/common.rb  | 22 +++++++---------
 .../outputs/elasticsearch/http_client.rb      |  2 --
 spec/es_spec_helper.rb                        | 12 +++++++--
 .../outputs/compressed_indexing_spec.rb       |  2 +-
 spec/integration/outputs/delete_spec.rb       |  8 +++---
 .../integration/outputs/groovy_update_spec.rb | 22 ++++++++--------
 spec/integration/outputs/index_spec.rb        |  2 +-
 .../integration/outputs/index_version_spec.rb | 14 +++++-----
 .../outputs/painless_update_spec.rb           | 26 +++++++++----------
 spec/integration/outputs/retry_spec.rb        |  4 +--
 spec/integration/outputs/templates_5x_spec.rb |  2 +-
 spec/integration/outputs/update_spec.rb       | 12 ++++-----
 spec/unit/outputs/elasticsearch_spec.rb       |  8 +++---
 spec/unit/outputs/error_whitelist_spec.rb     |  2 +-
 15 files changed, 73 insertions(+), 68 deletions(-)

diff --git a/ci/run.sh b/ci/run.sh
index f30409d99ad30f..02e03012dcc906 100755
--- a/ci/run.sh
+++ b/ci/run.sh
@@ -132,6 +132,9 @@ else
   fi
 
   case "$ES_VERSION" in
+  # ES_VERSION prefixed with LATEST-SNAPSHOT- will interrogate the artifacts API to find the latest version that matches
+  # the part after the prefix - eg LATEST-SNAPSHOT-7 will pull the latest snapshot release of 7.x of Elasticsearch,
+  # LATEST-SNAPSHOT-6.5 will pull the latest snapshot release of 6.5
     LATEST-SNAPSHOT-*)
       split_latest=${ES_VERSION##*-}
 
diff --git a/lib/logstash/outputs/elasticsearch/common.rb b/lib/logstash/outputs/elasticsearch/common.rb
index 4a207a08f78bd0..3213f10f8b300c 100644
--- a/lib/logstash/outputs/elasticsearch/common.rb
+++ b/lib/logstash/outputs/elasticsearch/common.rb
@@ -67,7 +67,7 @@ def event_action_tuple(event)
         :_id => @document_id ? event.sprintf(@document_id) : nil,
         :_index => event.sprintf(@index),
         :_type => get_event_type(event),
-        routing_key => @routing ? event.sprintf(@routing) : nil
+        routing_field_name => @routing ? event.sprintf(@routing) : nil
       }
 
       if @pipeline
@@ -79,16 +79,16 @@ def event_action_tuple(event)
           join_value = event.get(@join_field)
           parent_value = event.sprintf(@parent)
           event.set(@join_field, { "name" => join_value, "parent" => parent_value })
-          params[routing_key] = event.sprintf(@parent)
+          params[routing_field_name] = event.sprintf(@parent)
         else
           params[:parent] = event.sprintf(@parent)
         end
       end
 
       if action == 'update'
-        params[upsert_key] = LogStash::Json.load(event.sprintf(@upsert)) if @upsert != ""
+        params[:_upsert] = LogStash::Json.load(event.sprintf(@upsert)) if @upsert != ""
         params[:_script] = event.sprintf(@script) if @script != ""
-        params[retry_on_conflict_key] = @retry_on_conflict
+        params[retry_on_conflict_action_name] = @retry_on_conflict
       end
 
       if @version
@@ -115,16 +115,12 @@ def maximum_seen_major_version
     end
 
 
-    def routing_key
-      @routing_key ||= maximum_seen_major_version >= 6 ? :routing : :_routing
+    def routing_field_name
+      maximum_seen_major_version >= 6 ? :routing : :_routing
     end
 
-    def upsert_key
-      @upsert_key ||= maximum_seen_major_version >= 6 ? :upsert : :_upsert
-    end
-
-    def retry_on_conflict_key
-      @retry_on_conflict_key ||= maximum_seen_major_version >= 6 ? :retry_on_conflict : :_retry_on_conflict
+    def retry_on_conflict_action_name
+      maximum_seen_major_version >= 6 ? :retry_on_conflict : :_retry_on_conflict
     end
 
     def install_template
@@ -319,7 +315,7 @@ def safe_bulk(actions)
         log_hash = {:code => e.response_code, :url => e.url.sanitized.to_s}
         log_hash[:body] = e.response_body if @logger.debug? # Generally this is too verbose
         message = "Encountered a retryable error. Will Retry with exponential backoff "
-puts e.response_body
+
         # We treat 429s as a special case because these really aren't errors, but
         # rather just ES telling us to back off a bit, which we do.
         # The other retryable code is 503, which are true errors
diff --git a/lib/logstash/outputs/elasticsearch/http_client.rb b/lib/logstash/outputs/elasticsearch/http_client.rb
index cd2194d930ca40..b46f4be0a25988 100644
--- a/lib/logstash/outputs/elasticsearch/http_client.rb
+++ b/lib/logstash/outputs/elasticsearch/http_client.rb
@@ -362,7 +362,6 @@ def update_action_builder(args, source)
         elsif @options[:doc_as_upsert]
           source['upsert'] = source_orig
         else
-          source['upsert'] = args.delete(:upsert) if args[:upsert]
           source['upsert'] = args.delete(:_upsert) if args[:_upsert]
         end
         case @options[:script_type]
@@ -380,7 +379,6 @@ def update_action_builder(args, source)
           source['doc_as_upsert'] = true
         else
           source['upsert'] = args.delete(:_upsert) if args[:_upsert]
-          source['upsert'] = args.delete(:upsert) if args[:upsert]
         end
       end
       [args, source]
diff --git a/spec/es_spec_helper.rb b/spec/es_spec_helper.rb
index 40e4e1b7e7a3f6..0666559027a082 100644
--- a/spec/es_spec_helper.rb
+++ b/spec/es_spec_helper.rb
@@ -11,7 +11,7 @@ def get_client
     Elasticsearch::Client.new(:hosts => [get_host_port])
   end
 
-  def get_doc_type
+  def doc_type
     if ESHelper.es_version_satisfies?(">=7")
       "_doc"
     else
@@ -19,7 +19,7 @@ def get_doc_type
     end
   end
 
-  def get_mapping_name
+  def mapping_name
     if ESHelper.es_version_satisfies?(">=7")
       "_doc"
     else
@@ -28,6 +28,14 @@ def get_mapping_name
 
   end
 
+  def routing_field_name
+    if ESHelper.es_version_satisfies?(">=6")
+      :routing
+    else
+      :_routing
+    end
+  end
+
   def self.es_version
     RSpec.configuration.filter[:es_version] || ENV['ES_VERSION']
   end
diff --git a/spec/integration/outputs/compressed_indexing_spec.rb b/spec/integration/outputs/compressed_indexing_spec.rb
index b6f403eb679361..3544f0a3bdd0c1 100644
--- a/spec/integration/outputs/compressed_indexing_spec.rb
+++ b/spec/integration/outputs/compressed_indexing_spec.rb
@@ -50,7 +50,7 @@
         result = LogStash::Json.load(response.body)
         result["hits"]["hits"].each do |doc|
           if ESHelper.es_version_satisfies?(">= 6")
-            expect(doc["_type"]).to eq(get_doc_type)
+            expect(doc["_type"]).to eq(doc_type)
           else
             expect(doc["_type"]).to eq(type)
           end
diff --git a/spec/integration/outputs/delete_spec.rb b/spec/integration/outputs/delete_spec.rb
index cc317948165702..f0dcc4886c1a99 100644
--- a/spec/integration/outputs/delete_spec.rb
+++ b/spec/integration/outputs/delete_spec.rb
@@ -40,12 +40,12 @@
       it "should ignore non-monotonic external version updates" do
         id = "ev2"
         subject.multi_receive([LogStash::Event.new("my_id" => id, "my_action" => "index", "message" => "foo", "my_version" => 99)])
-        r = es.get(:index => 'logstash-delete', :type => get_doc_type, :id => id, :refresh => true)
+        r = es.get(:index => 'logstash-delete', :type => doc_type, :id => id, :refresh => true)
         expect(r['_version']).to eq(99)
         expect(r['_source']['message']).to eq('foo')
 
         subject.multi_receive([LogStash::Event.new("my_id" => id, "my_action" => "delete", "message" => "foo", "my_version" => 98)])
-        r2 = es.get(:index => 'logstash-delete', :type => get_doc_type, :id => id, :refresh => true)
+        r2 = es.get(:index => 'logstash-delete', :type => doc_type, :id => id, :refresh => true)
         expect(r2['_version']).to eq(99)
         expect(r2['_source']['message']).to eq('foo')
       end
@@ -53,12 +53,12 @@
       it "should commit monotonic external version updates" do
         id = "ev3"
         subject.multi_receive([LogStash::Event.new("my_id" => id, "my_action" => "index", "message" => "foo", "my_version" => 99)])
-        r = es.get(:index => 'logstash-delete', :type => get_doc_type, :id => id, :refresh => true)
+        r = es.get(:index => 'logstash-delete', :type => doc_type, :id => id, :refresh => true)
         expect(r['_version']).to eq(99)
         expect(r['_source']['message']).to eq('foo')
 
         subject.multi_receive([LogStash::Event.new("my_id" => id, "my_action" => "delete", "message" => "foo", "my_version" => 100)])
-        expect { es.get(:index => 'logstash-delete', :type => get_doc_type, :id => id, :refresh => true) }.to raise_error(Elasticsearch::Transport::Transport::Errors::NotFound)
+        expect { es.get(:index => 'logstash-delete', :type => doc_type, :id => id, :refresh => true) }.to raise_error(Elasticsearch::Transport::Transport::Errors::NotFound)
       end
     end
   end
diff --git a/spec/integration/outputs/groovy_update_spec.rb b/spec/integration/outputs/groovy_update_spec.rb
index 7443887e78a0bc..598491c5d32308 100644
--- a/spec/integration/outputs/groovy_update_spec.rb
+++ b/spec/integration/outputs/groovy_update_spec.rb
@@ -25,7 +25,7 @@ def get_es_output( options={} )
       @es.indices.delete(:index => "*") rescue nil
       @es.index(
         :index => 'logstash-update',
-        :type => 'doc',
+        :type => doc_type,
         :id => "123",
         :body => { :message => 'Test', :counter => 1 }
       )
@@ -37,7 +37,7 @@ def get_es_output( options={} )
         subject = get_es_output({ 'document_id' => "123", 'script' => 'scripted_update', 'script_type' => 'file' })
         subject.register
         subject.multi_receive([LogStash::Event.new("count" => 2)])
-        r = @es.get(:index => 'logstash-update', :type => 'doc', :id => "123", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "123", :refresh => true)
         expect(r["_source"]["counter"]).to eq(3)
       end
 
@@ -45,7 +45,7 @@ def get_es_output( options={} )
         subject = get_es_output({ 'document_id' => "123", 'script' => 'scripted_update_nested', 'script_type' => 'file' })
         subject.register
         subject.multi_receive([LogStash::Event.new("data" => { "count" => 3 })])
-        r = @es.get(:index => 'logstash-update', :type => 'doc', :id => "123", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "123", :refresh => true)
         expect(r["_source"]["counter"]).to eq(4)
       end
 
@@ -58,7 +58,7 @@ def get_es_output( options={} )
         })
         subject.register
         subject.multi_receive([LogStash::Event.new("counter" => 3 )])
-        r = @es.get(:index => 'logstash-update', :type => 'doc', :id => "123", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "123", :refresh => true)
         expect(r["_source"]["counter"]).to eq(4)
       end
 
@@ -72,7 +72,7 @@ def get_es_output( options={} )
         })
         subject.register
         subject.multi_receive([LogStash::Event.new("counter" => 3 )])
-        r = @es.get(:index => 'logstash-update', :type => 'doc', :id => "123", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "123", :refresh => true)
         expect(r["_source"]["counter"]).to eq(4)
       end
 
@@ -86,7 +86,7 @@ def get_es_output( options={} )
         })
         subject.register
         subject.multi_receive([LogStash::Event.new("counter" => 3 )])
-        r = @es.get(:index => 'logstash-update', :type => 'doc', :id => "456", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "456", :refresh => true)
         expect(r["_source"]["counter"]).to eq(3)
       end
 
@@ -100,7 +100,7 @@ def get_es_output( options={} )
         })
         subject.register
         subject.multi_receive([LogStash::Event.new("count" => 4 )])
-        r = @es.get(:index => 'logstash-update', :type => 'doc', :id => "123", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "123", :refresh => true)
         expect(r["_source"]["counter"]).to eq(5)
       end
     end
@@ -110,7 +110,7 @@ def get_es_output( options={} )
         subject = get_es_output({ 'document_id' => "456", 'upsert' => '{"message": "upsert message"}' })
         subject.register
         subject.multi_receive([LogStash::Event.new("message" => "sample message here")])
-        r = @es.get(:index => 'logstash-update', :type => 'doc', :id => "456", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "456", :refresh => true)
         expect(r["_source"]["message"]).to eq('upsert message')
       end
 
@@ -118,7 +118,7 @@ def get_es_output( options={} )
         subject = get_es_output({ 'document_id' => "456", 'doc_as_upsert' => true })
         subject.register
         subject.multi_receive([LogStash::Event.new("message" => "sample message here")])
-        r = @es.get(:index => 'logstash-update', :type => 'doc', :id => "456", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "456", :refresh => true)
         expect(r["_source"]["message"]).to eq('sample message here')
       end
 
@@ -133,7 +133,7 @@ def get_es_output( options={} )
         subject = get_es_output({ 'document_id' => "456", 'script' => 'scripted_update', 'upsert' => '{"message": "upsert message"}', 'script_type' => 'file' })
         subject.register
         subject.multi_receive([LogStash::Event.new("message" => "sample message here")])
-        r = @es.get(:index => 'logstash-update', :type => 'doc', :id => "456", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "456", :refresh => true)
         expect(r["_source"]["message"]).to eq('upsert message')
       end
 
@@ -142,7 +142,7 @@ def get_es_output( options={} )
         subject.register
         subject.multi_receive([LogStash::Event.new("counter" => 1)])
         @es.indices.refresh
-        r = @es.get(:index => 'logstash-update', :type => 'doc', :id => "456", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "456", :refresh => true)
         expect(r["_source"]["counter"]).to eq(1)
       end
     end
diff --git a/spec/integration/outputs/index_spec.rb b/spec/integration/outputs/index_spec.rb
index 4bc8c343dd8b12..48fdff187c583a 100644
--- a/spec/integration/outputs/index_spec.rb
+++ b/spec/integration/outputs/index_spec.rb
@@ -79,7 +79,7 @@
       result = LogStash::Json.load(response.body)
       result["hits"]["hits"].each do |doc|
         if ESHelper.es_version_satisfies?(">= 6")
-          expect(doc["_type"]).to eq(get_doc_type)
+          expect(doc["_type"]).to eq(doc_type)
         else
           expect(doc["_type"]).to eq(type)
         end
diff --git a/spec/integration/outputs/index_version_spec.rb b/spec/integration/outputs/index_version_spec.rb
index 2abefd796c86a3..c23d659d34e32b 100644
--- a/spec/integration/outputs/index_version_spec.rb
+++ b/spec/integration/outputs/index_version_spec.rb
@@ -38,11 +38,11 @@
 
         it "should default to ES version" do
           subject.multi_receive([LogStash::Event.new("my_id" => "123", "message" => "foo")])
-          r = es.get(:index => 'logstash-index', :type => get_doc_type, :id => "123", :refresh => true)
+          r = es.get(:index => 'logstash-index', :type => doc_type, :id => "123", :refresh => true)
           expect(r["_version"]).to eq(1)
           expect(r["_source"]["message"]).to eq('foo')
           subject.multi_receive([LogStash::Event.new("my_id" => "123", "message" => "foobar")])
-          r2 = es.get(:index => 'logstash-index', :type => get_doc_type, :id => "123", :refresh => true)
+          r2 = es.get(:index => 'logstash-index', :type => doc_type, :id => "123", :refresh => true)
           expect(r2["_version"]).to eq(2)
           expect(r2["_source"]["message"]).to eq('foobar')
         end  
@@ -66,7 +66,7 @@
         it "should respect the external version" do
           id = "ev1"
           subject.multi_receive([LogStash::Event.new("my_id" => id, "my_version" => "99", "message" => "foo")])
-          r = es.get(:index => 'logstash-index', :type => get_doc_type, :id => id, :refresh => true)
+          r = es.get(:index => 'logstash-index', :type => doc_type, :id => id, :refresh => true)
           expect(r["_version"]).to eq(99)
           expect(r["_source"]["message"]).to eq('foo')
         end
@@ -74,12 +74,12 @@
         it "should ignore non-monotonic external version updates" do
           id = "ev2"
           subject.multi_receive([LogStash::Event.new("my_id" => id, "my_version" => "99", "message" => "foo")])
-          r = es.get(:index => 'logstash-index', :type => get_doc_type, :id => id, :refresh => true)
+          r = es.get(:index => 'logstash-index', :type => doc_type, :id => id, :refresh => true)
           expect(r["_version"]).to eq(99)
           expect(r["_source"]["message"]).to eq('foo')
 
           subject.multi_receive([LogStash::Event.new("my_id" => id, "my_version" => "98", "message" => "foo")])
-          r2 = es.get(:index => 'logstash-index', :type => get_doc_type, :id => id, :refresh => true)
+          r2 = es.get(:index => 'logstash-index', :type => doc_type, :id => id, :refresh => true)
           expect(r2["_version"]).to eq(99)
           expect(r2["_source"]["message"]).to eq('foo')
         end
@@ -87,12 +87,12 @@
         it "should commit monotonic external version updates" do
           id = "ev3"
           subject.multi_receive([LogStash::Event.new("my_id" => id, "my_version" => "99", "message" => "foo")])
-          r = es.get(:index => 'logstash-index', :type => get_doc_type, :id => id, :refresh => true)
+          r = es.get(:index => 'logstash-index', :type => doc_type, :id => id, :refresh => true)
           expect(r["_version"]).to eq(99)
           expect(r["_source"]["message"]).to eq('foo')
 
           subject.multi_receive([LogStash::Event.new("my_id" => id, "my_version" => "100", "message" => "foo")])
-          r2 = es.get(:index => 'logstash-index', :type => get_doc_type, :id => id, :refresh => true)
+          r2 = es.get(:index => 'logstash-index', :type => doc_type, :id => id, :refresh => true)
           expect(r2["_version"]).to eq(100)
           expect(r2["_source"]["message"]).to eq('foo')
         end
diff --git a/spec/integration/outputs/painless_update_spec.rb b/spec/integration/outputs/painless_update_spec.rb
index c0c703b3b7e6b0..0f06fecb43bec4 100644
--- a/spec/integration/outputs/painless_update_spec.rb
+++ b/spec/integration/outputs/painless_update_spec.rb
@@ -27,7 +27,7 @@ def get_es_output( options={} )
       @es.indices.delete(:index => "*") rescue nil
       @es.index(
         :index => 'logstash-update',
-        :type => get_doc_type,
+        :type => doc_type,
         :id => "123",
         :body => { :message => 'Test', :counter => 1 }
       )
@@ -41,7 +41,7 @@ def get_es_output( options={} )
             subject = get_es_output({ 'document_id' => "123", 'script' => 'scripted_update', 'script_type' => 'file' })
             subject.register
             subject.multi_receive([LogStash::Event.new("count" => 2)])
-            r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "123", :refresh => true)
+            r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "123", :refresh => true)
             expect(r["_source"]["counter"]).to eq(3)
           end
 
@@ -49,7 +49,7 @@ def get_es_output( options={} )
             subject = get_es_output({ 'document_id' => "123", 'script' => 'scripted_update_nested', 'script_type' => 'file' })
             subject.register
             subject.multi_receive([LogStash::Event.new("data" => { "count" => 3 })])
-            r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "123", :refresh => true)
+            r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "123", :refresh => true)
             expect(r["_source"]["counter"]).to eq(4)
           end
         end
@@ -63,7 +63,7 @@ def get_es_output( options={} )
         })
         subject.register
         subject.multi_receive([LogStash::Event.new("counter" => 3 )])
-        r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "123", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "123", :refresh => true)
         expect(r["_source"]["counter"]).to eq(4)
       end
 
@@ -76,7 +76,7 @@ def get_es_output( options={} )
         })
         subject.register
         subject.multi_receive([LogStash::Event.new("counter" => 3 )])
-        r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "123", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "123", :refresh => true)
         expect(r["_source"]["counter"]).to eq(4)
       end
 
@@ -89,7 +89,7 @@ def get_es_output( options={} )
         })
         subject.register
         subject.multi_receive([LogStash::Event.new("counter" => 3 )])
-        r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "456", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "456", :refresh => true)
         expect(r["_source"]["counter"]).to eq(3)
       end
 
@@ -114,7 +114,7 @@ def get_es_output( options={} )
           subject = get_es_output(plugin_parameters)
           subject.register
           subject.multi_receive([LogStash::Event.new("count" => 4 )])
-          r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "123", :refresh => true)
+          r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "123", :refresh => true)
           expect(r["_source"]["counter"]).to eq(5)
         end
       end
@@ -125,7 +125,7 @@ def get_es_output( options={} )
         subject = get_es_output({ 'document_id' => "456", 'upsert' => '{"message": "upsert message"}' })
         subject.register
         subject.multi_receive([LogStash::Event.new("message" => "sample message here")])
-        r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "456", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "456", :refresh => true)
         expect(r["_source"]["message"]).to eq('upsert message')
       end
 
@@ -133,7 +133,7 @@ def get_es_output( options={} )
         subject = get_es_output({ 'document_id' => "456", 'doc_as_upsert' => true })
         subject.register
         subject.multi_receive([LogStash::Event.new("message" => "sample message here")])
-        r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "456", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "456", :refresh => true)
         expect(r["_source"]["message"]).to eq('sample message here')
       end
 
@@ -150,7 +150,7 @@ def get_es_output( options={} )
             subject = get_es_output({ 'document_id' => "456", 'script' => 'scripted_update', 'upsert' => '{"message": "upsert message"}', 'script_type' => 'file' })
             subject.register
             subject.multi_receive([LogStash::Event.new("message" => "sample message here")])
-            r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "456", :refresh => true)
+            r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "456", :refresh => true)
             expect(r["_source"]["message"]).to eq('upsert message')
           end
 
@@ -159,7 +159,7 @@ def get_es_output( options={} )
             subject.register
             subject.multi_receive([LogStash::Event.new("counter" => 1)])
             @es.indices.refresh
-            r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "456", :refresh => true)
+            r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "456", :refresh => true)
             expect(r["_source"]["counter"]).to eq(1)
           end
         end
@@ -171,7 +171,7 @@ def get_es_output( options={} )
           subject.register
 
           subject.multi_receive([LogStash::Event.new("message" => "sample message here")])
-          r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "456", :refresh => true)
+          r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "456", :refresh => true)
           expect(r["_source"]["message"]).to eq('upsert message')
         end
 
@@ -180,7 +180,7 @@ def get_es_output( options={} )
           subject.register
           subject.multi_receive([LogStash::Event.new("counter" => 1)])
           @es.indices.refresh
-          r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "456", :refresh => true)
+          r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "456", :refresh => true)
           expect(r["_source"]["counter"]).to eq(1)
         end
       end
diff --git a/spec/integration/outputs/retry_spec.rb b/spec/integration/outputs/retry_spec.rb
index e15e23cd4c84ac..817629668fcdb9 100644
--- a/spec/integration/outputs/retry_spec.rb
+++ b/spec/integration/outputs/retry_spec.rb
@@ -4,9 +4,9 @@
 describe "failures in bulk class expected behavior", :integration => true do
   let(:template) { '{"template" : "not important, will be updated by :index"}' }
   let(:event1) { LogStash::Event.new("somevalue" => 100, "@timestamp" => "2014-11-17T20:37:17.223Z", "@metadata" => {"retry_count" => 0}) }
-  let(:action1) { ["index", {:_id=>nil, :routing=>nil, :_index=>"logstash-2014.11.17", :_type=> get_doc_type }, event1] }
+  let(:action1) { ["index", {:_id=>nil, routing_field_name =>nil, :_index=>"logstash-2014.11.17", :_type=> doc_type }, event1] }
   let(:event2) { LogStash::Event.new("geoip" => { "location" => [ 0.0, 0.0] }, "@timestamp" => "2014-11-17T20:37:17.223Z", "@metadata" => {"retry_count" => 0}) }
-  let(:action2) { ["index", {:_id=>nil, :routing=>nil, :_index=>"logstash-2014.11.17", :_type=> get_doc_type }, event2] }
+  let(:action2) { ["index", {:_id=>nil, routing_field_name =>nil, :_index=>"logstash-2014.11.17", :_type=> doc_type }, event2] }
   let(:invalid_event) { LogStash::Event.new("geoip" => { "location" => "notlatlon" }, "@timestamp" => "2014-11-17T20:37:17.223Z") }
 
   def mock_actions_with_response(*resp)
diff --git a/spec/integration/outputs/templates_5x_spec.rb b/spec/integration/outputs/templates_5x_spec.rb
index a601f0aba5813a..7d354487147975 100644
--- a/spec/integration/outputs/templates_5x_spec.rb
+++ b/spec/integration/outputs/templates_5x_spec.rb
@@ -82,7 +82,7 @@
     end
 
     it "make [geoip][location] a geo_point" do
-      expect(@es.indices.get_template(name: "logstash")["logstash"]["mappings"][get_mapping_name]["properties"]["geoip"]["properties"]["location"]["type"]).to eq("geo_point")
+      expect(@es.indices.get_template(name: "logstash")["logstash"]["mappings"][mapping_name]["properties"]["geoip"]["properties"]["location"]["type"]).to eq("geo_point")
     end
 
     it "aggregate .keyword results correctly " do
diff --git a/spec/integration/outputs/update_spec.rb b/spec/integration/outputs/update_spec.rb
index 2bbed1ba743500..300754276e8534 100644
--- a/spec/integration/outputs/update_spec.rb
+++ b/spec/integration/outputs/update_spec.rb
@@ -24,7 +24,7 @@ def get_es_output( options={} )
       @es.indices.delete(:index => "*") rescue nil
       @es.index(
         :index => 'logstash-update',
-        :type => get_doc_type,
+        :type => doc_type,
         :id => "123",
         :body => { :message => 'Test', :counter => 1 }
       )
@@ -41,14 +41,14 @@ def get_es_output( options={} )
         subject = get_es_output({ 'document_id' => "456" } )
         subject.register
         subject.multi_receive([LogStash::Event.new("message" => "sample message here")])
-        expect {@es.get(:index => 'logstash-update', :type => get_doc_type, :id => "456", :refresh => true)}.to raise_error(Elasticsearch::Transport::Transport::Errors::NotFound)
+        expect {@es.get(:index => 'logstash-update', :type => doc_type, :id => "456", :refresh => true)}.to raise_error(Elasticsearch::Transport::Transport::Errors::NotFound)
       end
 
       it "should update existing document" do
         subject = get_es_output({ 'document_id' => "123" })
         subject.register
         subject.multi_receive([LogStash::Event.new("message" => "updated message here")])
-        r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "123", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "123", :refresh => true)
         expect(r["_source"]["message"]).to eq('updated message here')
       end
 
@@ -58,7 +58,7 @@ def get_es_output( options={} )
         subject = get_es_output({ 'document_id' => "123" })
         subject.register
         subject.multi_receive([LogStash::Event.new("data" => "updated message here", "message" => "foo")])
-        r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "123", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "123", :refresh => true)
         expect(r["_source"]["data"]).to eq('updated message here')
         expect(r["_source"]["message"]).to eq('foo')
       end
@@ -95,7 +95,7 @@ def get_es_output( options={} )
         subject = get_es_output({ 'document_id' => "456", 'upsert' => '{"message": "upsert message"}' })
         subject.register
         subject.multi_receive([LogStash::Event.new("message" => "sample message here")])
-        r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "456", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "456", :refresh => true)
         expect(r["_source"]["message"]).to eq('upsert message')
       end
 
@@ -103,7 +103,7 @@ def get_es_output( options={} )
         subject = get_es_output({ 'document_id' => "456", 'doc_as_upsert' => true })
         subject.register
         subject.multi_receive([LogStash::Event.new("message" => "sample message here")])
-        r = @es.get(:index => 'logstash-update', :type => get_doc_type, :id => "456", :refresh => true)
+        r = @es.get(:index => 'logstash-update', :type => doc_type, :id => "456", :refresh => true)
         expect(r["_source"]["message"]).to eq('sample message here')
       end
 
diff --git a/spec/unit/outputs/elasticsearch_spec.rb b/spec/unit/outputs/elasticsearch_spec.rb
index 635d88b2f64de9..6fb32ddaa1ded5 100644
--- a/spec/unit/outputs/elasticsearch_spec.rb
+++ b/spec/unit/outputs/elasticsearch_spec.rb
@@ -301,7 +301,7 @@
       let(:event) { LogStash::Event.new("myactionfield" => "update", "message" => "blah") }
 
       it "should obtain specific action's params from event_action_tuple" do
-        expect(subject.event_action_tuple(event)[1]).to include(subject.upsert_key)
+        expect(subject.event_action_tuple(event)[1]).to include(:_upsert)
       end
     end
 
@@ -362,7 +362,7 @@
       it "should not set the retry_on_conflict parameter when creating an event_action_tuple" do
         allow(subject.client).to receive(:maximum_seen_major_version).and_return(maximum_seen_major_version)
         action, params, event_data = subject.event_action_tuple(event)
-        expect(params).not_to include({subject.retry_on_conflict_key => num_retries})
+        expect(params).not_to include({subject.retry_on_conflict_action_name => num_retries})
       end
     end
 
@@ -371,7 +371,7 @@
 
       it "should set the retry_on_conflict parameter when creating an event_action_tuple" do
         action, params, event_data = subject.event_action_tuple(event)
-        expect(params).to include({subject.retry_on_conflict_key => num_retries})
+        expect(params).to include({subject.retry_on_conflict_action_name => num_retries})
       end
     end
 
@@ -380,7 +380,7 @@
 
       it "should set the retry_on_conflict parameter when creating an event_action_tuple" do
         action, params, event_data = subject.event_action_tuple(event)
-        expect(params).to include({subject.retry_on_conflict_key => num_retries})
+        expect(params).to include({subject.retry_on_conflict_action_name => num_retries})
         expect(action).to eq("update")
       end
     end
diff --git a/spec/unit/outputs/error_whitelist_spec.rb b/spec/unit/outputs/error_whitelist_spec.rb
index 96b4e0821d3f11..0f8a295713dad6 100644
--- a/spec/unit/outputs/error_whitelist_spec.rb
+++ b/spec/unit/outputs/error_whitelist_spec.rb
@@ -4,7 +4,7 @@
 describe "whitelisting error types in expected behavior" do
   let(:template) { '{"template" : "not important, will be updated by :index"}' }
   let(:event1) { LogStash::Event.new("somevalue" => 100, "@timestamp" => "2014-11-17T20:37:17.223Z") }
-  let(:action1) { ["index", {:_id=>1, :routing=>nil, :_index=>"logstash-2014.11.17", :_type=> get_doc_type }, event1] }
+  let(:action1) { ["index", {:_id=>1, :routing=>nil, :_index=>"logstash-2014.11.17", :_type=> doc_type }, event1] }
   let(:settings) { {"manage_template" => true, "index" => "logstash-2014.11.17", "template_overwrite" => true, "hosts" => get_host_port() } }
 
   subject { LogStash::Outputs::ElasticSearch.new(settings) }