diff --git a/CHANGELOG.md b/CHANGELOG.md
index e6242e09..88779ffc 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,6 @@
+## 12.0.7
+ - Support both, encoded and non encoded api-key formats on plugin configuration [#1223](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1223)
+
 ## 12.0.6
  - Add headers reporting uncompressed size and doc count for bulk requests [#1217](https://github.com/logstash-plugins/logstash-output-elasticsearch/pull/1217)
 
diff --git a/lib/logstash/outputs/elasticsearch/http_client_builder.rb b/lib/logstash/outputs/elasticsearch/http_client_builder.rb
index b7344499..c1a378c1 100644
--- a/lib/logstash/outputs/elasticsearch/http_client_builder.rb
+++ b/lib/logstash/outputs/elasticsearch/http_client_builder.rb
@@ -188,25 +188,37 @@ def self.setup_basic_auth(logger, params)
     def self.setup_api_key(logger, params)
       api_key = params["api_key"]
 
-      return {} unless (api_key && api_key.value)
+      return {} unless (api_key&.value)
 
-      { "Authorization" => "ApiKey " + Base64.strict_encode64(api_key.value) }
-    end
+      value = is_base64?(api_key.value) ?  api_key.value : Base64.strict_encode64(api_key.value)
 
-    private
-    def self.dedup_slashes(url)
-      url.gsub(/\/+/, "/")
+      { "Authorization" => "ApiKey #{value}" }
     end
 
-    # Set a `filter_path` query parameter if it is not already set to be
-    # `filter_path=errors,items.*.error,items.*.status` to reduce the payload between Logstash and Elasticsearch
-    def self.resolve_filter_path(url)
-      return url if url.match?(/(?:[&|?])filter_path=/)
-      ("#{url}#{query_param_separator(url)}filter_path=errors,items.*.error,items.*.status")
-    end
+    class << self
+      private
+      def dedup_slashes(url)
+        url.gsub(/\/+/, "/")
+      end
+
+      # Set a `filter_path` query parameter if it is not already set to be
+      # `filter_path=errors,items.*.error,items.*.status` to reduce the payload between Logstash and Elasticsearch
+      def resolve_filter_path(url)
+        return url if url.match?(/(?:[&|?])filter_path=/)
+        ("#{url}#{query_param_separator(url)}filter_path=errors,items.*.error,items.*.status")
+      end
 
-    def self.query_param_separator(url)
-      url.match?(/\?[^\s#]+/) ? '&' : '?'
+      def query_param_separator(url)
+        url.match?(/\?[^\s#]+/) ? '&' : '?'
+      end
+
+      def is_base64?(string)
+        begin
+          string == Base64.strict_encode64(Base64.strict_decode64(string))
+        rescue ArgumentError
+          false
+        end
+      end
     end
   end
 end; end; end
diff --git a/spec/unit/http_client_builder_spec.rb b/spec/unit/http_client_builder_spec.rb
index 713b6ee7..d7e6782c 100644
--- a/spec/unit/http_client_builder_spec.rb
+++ b/spec/unit/http_client_builder_spec.rb
@@ -26,6 +26,44 @@
     end
   end
 
+  describe "auth setup with api-key" do
+    let(:klass) { LogStash::Outputs::ElasticSearch::HttpClientBuilder }
+
+    context "when api-key is not encoded (id:api-key)" do
+      let(:api_key) { "id:api-key" }
+      let(:api_key_secured) do
+        secured = double("api_key")
+        allow(secured).to receive(:value).and_return(api_key)
+        secured
+      end
+      let(:options) { { "api_key" => api_key_secured } }
+      let(:logger) { double("logger") }
+      let(:api_key_header) { klass.setup_api_key(logger, options) }
+
+      it "returns the correct encoded api-key header" do
+        expected = "ApiKey #{Base64.strict_encode64(api_key)}"
+        expect(api_key_header["Authorization"]).to eql(expected)
+      end
+    end
+
+    context "when api-key is already encoded" do
+      let(:api_key) { Base64.strict_encode64("id:api-key") }
+      let(:api_key_secured) do
+        secured = double("api_key")
+        allow(secured).to receive(:value).and_return(api_key)
+        secured
+      end
+      let(:options) { { "api_key" => api_key_secured } }
+      let(:logger) { double("logger") }
+      let(:api_key_header) { klass.setup_api_key(logger, options) }
+
+      it "returns the api-key header as is" do
+        expected = "ApiKey #{api_key}"
+        expect(api_key_header["Authorization"]).to eql(expected)
+      end
+    end
+  end
+
   describe "customizing action paths" do
     let(:hosts) { [ ::LogStash::Util::SafeURI.new("http://localhost:9200") ] }
     let(:options) { {"hosts" => hosts } }