[Backport] bulkdindexer version and routing (#373)

* Add per item routing for bulk requests (#265)

* Remove docuemnt type from bulk indexer

* Util: Allow setting version metadata in Bulk Indexer

* Fix version type

* Update tests

Co-authored-by: Mike Luu <github@munkyboy.com>

* CI: Util: Add integration tests to bulkindexer for external versioning (#360)

Co-authored-by: Mehran Poursadeghi <mehr.prs@gmail.com>
Co-authored-by: Mike Luu <github@munkyboy.com>

Author: 3 people

esutil/bulk_indexer.go

@@ -106,6 +106,9 @@ type BulkIndexerItem struct {
 	Index           string
 	Action          string
 	DocumentID      string
+	Routing         string
+	Version         *int64
+	VersionType     string
 	Body            io.Reader
 	RetryOnConflict *int
 
@@ -408,10 +411,34 @@ func (w *worker) writeMeta(item BulkIndexerItem) error {
 		w.buf.Write(w.aux)
 		w.aux = w.aux[:0]
 	}
-	if item.Index != "" {
+
+	if item.DocumentID != "" && item.Version != nil {
+		w.buf.WriteRune(',')
+		w.buf.WriteString(`"version":`)
+		w.buf.WriteString(strconv.FormatInt(*item.Version, 10))
+	}
+
+	if item.DocumentID != "" && item.VersionType != "" {
+		w.buf.WriteRune(',')
+		w.buf.WriteString(`"version_type":`)
+		w.aux = strconv.AppendQuote(w.aux, item.VersionType)
+		w.buf.Write(w.aux)
+		w.aux = w.aux[:0]
+	}
+
+	if item.Routing != "" {
 		if item.DocumentID != "" {
 			w.buf.WriteRune(',')
 		}
+		w.buf.WriteString(`"_routing":`)
+		w.aux = strconv.AppendQuote(w.aux, item.Routing)
+		w.buf.Write(w.aux)
+		w.aux = w.aux[:0]
+	}
+	if item.Index != "" {
+		if item.DocumentID != "" || item.Routing != "" {
+			w.buf.WriteRune(',')
+		}
 		w.buf.WriteString(`"_index":`)
 		w.aux = strconv.AppendQuote(w.aux, item.Index)
 		w.buf.Write(w.aux)

esutil/bulk_indexer_integration_test.go

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//go:build integration
 // +build integration
 
 package esutil_test
@@ -187,6 +188,65 @@ func TestBulkIndexerIntegration(t *testing.T) {
 					t.Errorf("Expected indices to exist, but got a [%s] response", res.Status())
 				}
 			})
+
+			t.Run("External version", func(t *testing.T) {
+				var index string = "test-index-a"
+
+				es, _ := elasticsearch.NewClient(elasticsearch.Config{
+					CompressRequestBody: tt.CompressRequestBodyEnabled,
+					Logger:              &estransport.ColorLogger{Output: os.Stdout},
+				})
+
+				es.Indices.Delete([]string{index}, es.Indices.Delete.WithIgnoreUnavailable(true))
+				es.Indices.Create(index, es.Indices.Create.WithWaitForActiveShards("1"))
+
+				bulkIndex := func(bulkIndexer esutil.BulkIndexer, baseVersion int) {
+					var countTotal int = 500
+					var countSuccessful uint64
+					for i := 0; i < countTotal; i++ {
+						version := int64(baseVersion + i)
+						item := esutil.BulkIndexerItem{
+							Action:      "index",
+							Index:       index,
+							DocumentID:  strconv.Itoa(i),
+							Body:        strings.NewReader(body),
+							Version:     &version,
+							VersionType: "external",
+							OnSuccess: func(ctx context.Context, item esutil.BulkIndexerItem, item2 esutil.BulkIndexerResponseItem) {
+								if version != item2.Version &&
+									version != *item.Version &&
+									item2.Result != "created" {
+									t.Fatalf("version mismatch, expected: %d, got: %d && %d", version, item.Version, item2.Version)
+								}
+								atomic.AddUint64(&countSuccessful, 1)
+							},
+						}
+						err := bulkIndexer.Add(context.Background(), item)
+						if err != nil {
+							t.Fatal(err)
+						}
+					}
+					if err := bulkIndexer.Close(context.Background()); err != nil {
+						t.Fatal(err)
+					}
+
+					if int(countSuccessful) != countTotal {
+						t.Fatalf("Unexpected countSuccessful, expected %d, got: %d", countTotal, countSuccessful)
+					}
+				}
+
+				bi, _ := esutil.NewBulkIndexer(esutil.BulkIndexerConfig{
+					Index:  index,
+					Client: es,
+				})
+				bulkIndex(bi, 500)
+
+				bi, _ = esutil.NewBulkIndexer(esutil.BulkIndexerConfig{
+					Index:  index,
+					Client: es,
+				})
+				bulkIndex(bi, 900)
+			})
 		})
 	}
-}
+}

esutil/bulk_indexer_internal_test.go

@@ -644,8 +644,8 @@ func TestBulkIndexer(t *testing.T) {
 			t.Errorf("Unexpected NumAdded: %d", stats.NumAdded)
 		}
 	})
-
 	t.Run("Worker.writeMeta()", func(t *testing.T) {
+		v:=int64(23)
 		type args struct {
 			item BulkIndexerItem
 		}
@@ -684,6 +684,36 @@ func TestBulkIndexer(t *testing.T) {
 				}},
 				`{"index":{"_id":"42","_index":"test"}}` + "\n",
 			},
+			{
+				"with version and no document",
+				args{BulkIndexerItem{
+					Action:     "index",
+					Index:      "test",
+					Version:    &v,
+				}},
+				`{"index":{"_index":"test"}}` + "\n",
+			},
+			{
+				"with version",
+				args{BulkIndexerItem{
+					Action:     "index",
+					DocumentID: "42",
+					Index:      "test",
+					Version:    &v,
+				}},
+				`{"index":{"_id":"42","version":23,"_index":"test"}}` + "\n",
+			},
+			{
+				"with version and version_type",
+				args{BulkIndexerItem{
+					Action:      "index",
+					DocumentID:  "42",
+					Index:       "test",
+					Version:     &v,
+					VersionType: "external",
+				}},
+				`{"index":{"_id":"42","version":23,"version_type":"external","_index":"test"}}` + "\n",
+			},
 		}
 		for _, tt := range tests {
 			tt := tt