Add an ability to customize bulk api url

README.md

@@ -103,7 +103,11 @@ config :my_app, MyApp.ElasticsearchCluster,
 
       # By default bulk indexing uses the "create" action. To allow existing
       # documents to be replaced, use the "index" action instead.
-      bulk_action: "create"
+      bulk_action: "create",
+
+      # Path to bulk api url. Should start with a slash.
+      # You may need to use `/_bulk` for newer ElasticSearch versions.
+      bulk_path: "/_doc/_bulk"
     }
   }
 ```
@@ -207,9 +211,9 @@ As AWS does not provide credentials' based http authentication, you can use the
 To use this, just add `sigaws` to your dependencies and add this to your configuration:
 
 ```elixir
-# Add to deps 
+# Add to deps
 def deps do
-  [          
+  [
     # ...
     {:sigaws, ">= 0.0.0"}
   ]

lib/elasticsearch/indexing/bulk.ex

@@ -100,6 +100,7 @@ defmodule Elasticsearch.Index.Bulk do
     bulk_page_size = index_config[:bulk_page_size] || 5000
     bulk_wait_interval = index_config[:bulk_wait_interval] || 0
     action = index_config[:bulk_action] || "create"
+    bulk_path = index_config[:bulk_path] || "/_doc/_bulk"
 
     errors =
       store.transaction(fn ->
@@ -108,20 +109,21 @@ defmodule Elasticsearch.Index.Bulk do
         |> Stream.map(&encode!(config, &1, index_name, action))
         |> Stream.chunk_every(bulk_page_size)
         |> Stream.intersperse(bulk_wait_interval)
-        |> Stream.map(&put_bulk_page(config, index_name, &1))
+        |> Stream.map(&put_bulk_page(config, index_name, bulk_path, &1))
         |> Enum.reduce(errors, &collect_errors(&1, &2, action))
       end)
 
     upload(config, index_name, %{index_config | sources: tail}, errors)
   end
 
-  defp put_bulk_page(_config, _index_name, wait_interval) when is_integer(wait_interval) do
+  defp put_bulk_page(_config, _index_name, _bulk_path, wait_interval)
+       when is_integer(wait_interval) do
     Logger.debug("Pausing #{wait_interval}ms between bulk pages")
     :timer.sleep(wait_interval)
   end
 
-  defp put_bulk_page(config, index_name, items) when is_list(items) do
-    Elasticsearch.put(config, "/#{index_name}/_doc/_bulk", Enum.join(items))
+  defp put_bulk_page(config, index_name, bulk_path, items) when is_list(items) do
+    Elasticsearch.put(config, "/#{index_name}#{bulk_path}", Enum.join(items))
   end
 
   defp collect_errors({:ok, %{"errors" => true} = response}, errors, action) do