Add require_alias on query string and body with serialization

Author: Anaethelion

esutil/bulk_indexer.go

@@ -76,6 +76,7 @@ type BulkIndexerConfig struct {
 	Pretty              bool
 	Refresh             string
 	Routing             string
+	RequireAlias        *bool
 	Source              []string
 	SourceExcludes      []string
 	SourceIncludes      []string
@@ -101,6 +102,7 @@ type BulkIndexerItem struct {
 	Action          string
 	DocumentID      string
 	Routing         string
+	RequireAlias    *bool
 	Version         *int64
 	VersionType     string
 	Body            io.ReadSeeker
@@ -166,6 +168,15 @@ func (item *BulkIndexerItem) marshallMeta() {
 		item.meta.Write(strconv.AppendInt(aux, int64(*item.RetryOnConflict), 10))
 		aux = aux[:0]
 	}
+	if item.RequireAlias != nil {
+		if item.DocumentID != "" || item.Routing != "" || item.Index != "" || item.RetryOnConflict != nil {
+			item.meta.WriteString(",")
+		}
+		item.meta.WriteString(`"require_alias":`)
+		item.meta.Write(strconv.AppendBool(aux, *item.RequireAlias))
+		aux = aux[:0]
+	}
+
 	item.meta.WriteRune('}')
 	item.meta.WriteRune('}')
 	item.meta.WriteRune('\n')
@@ -519,6 +530,7 @@ func (w *worker) flushBuffer(ctx context.Context) error {
 		Pipeline:            w.bi.config.Pipeline,
 		Refresh:             w.bi.config.Refresh,
 		Routing:             w.bi.config.Routing,
+		RequireAlias:        w.bi.config.RequireAlias,
 		Source:              w.bi.config.Source,
 		SourceExcludes:      w.bi.config.SourceExcludes,
 		SourceIncludes:      w.bi.config.SourceIncludes,

esutil/bulk_indexer_integration_test.go

@@ -24,6 +24,7 @@ import (
 	"compress/gzip"
 	"context"
 	"fmt"
+	"github.com/elastic/go-elasticsearch/v8/esapi"
 	"os"
 	"strconv"
 	"strings"
@@ -257,6 +258,53 @@ func TestBulkIndexerIntegration(t *testing.T) {
 				})
 				bulkIndex(bi, 900)
 			})
+
+			t.Run("Index alias", func(t *testing.T) {
+				var index string = "test-index-a"
+				var alias string = "test-alias-a"
+
+				es, _ := elasticsearch.NewClient(elasticsearch.Config{
+					CompressRequestBody:      tt.CompressRequestBodyEnabled,
+					CompressRequestBodyLevel: tt.CompressRequestBodyLevel,
+					Logger:                   &elastictransport.ColorLogger{Output: os.Stdout, EnableRequestBody: true, EnableResponseBody: true},
+				})
+
+				es.Indices.Delete([]string{index}, es.Indices.Delete.WithIgnoreUnavailable(true))
+				es.Indices.Create(index, es.Indices.Create.WithWaitForActiveShards("1"))
+				es.Indices.PutAlias([]string{index}, alias)
+
+				bi, _ := esutil.NewBulkIndexer(esutil.BulkIndexerConfig{
+					Client: es,
+				})
+				func(bulkIndexer esutil.BulkIndexer) {
+					var countTotal int = 10
+					var countSuccessful uint64
+					for i := 0; i < countTotal; i++ {
+						item := esutil.BulkIndexerItem{
+							Action:       "index",
+							Index:        alias,
+							DocumentID:   strconv.Itoa(i),
+							Body:         strings.NewReader(body),
+							RequireAlias: esapi.BoolPtr(true),
+							OnSuccess: func(ctx context.Context, item esutil.BulkIndexerItem, item2 esutil.BulkIndexerResponseItem) {
+								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/bulk_indexer_internal_test.go

@@ -666,6 +666,28 @@ func TestBulkIndexer(t *testing.T) {
 				}},
 				`{"index":{"_id":"42","version":23,"version_type":"external","_index":"test"}}` + "\n",
 			},
+			{
+				"with require_alias",
+				args{BulkIndexerItem{
+					Action:       "index",
+					DocumentID:   "42",
+					Index:        "test",
+					RequireAlias: esapi.BoolPtr(true),
+				}},
+				`{"index":{"_id":"42","_index":"test","require_alias":true}}` + "\n",
+			},
+			{
+				"with version, version_type and require_alias",
+				args{BulkIndexerItem{
+					Action:       "index",
+					DocumentID:   "42",
+					Index:        "test",
+					Version:      &v,
+					VersionType:  "external",
+					RequireAlias: esapi.BoolPtr(true),
+				}},
+				`{"index":{"_id":"42","version":23,"version_type":"external","_index":"test","require_alias":true}}` + "\n",
+			},
 			{
 				"with retry_on_conflict and bad action",
 				args{BulkIndexerItem{