Function TestBulkIndexerFlushJitter

esutil/bulk_indexer_internal_test.go:2326–2399

(t *testing.T)

}

func TestBulkIndexerFlushJitter(t *testing.T) {
	t.Run("auto-flush fires with jitter configured", func(t *testing.T) {
		es, err := elasticsearch.New(elasticsearch.WithTransportOptions(
			elastictransport.WithTransport(&mockTransport{
				RoundTripFunc: func(*http.Request) (*http.Response, error) {
					return &http.Response{
						StatusCode: http.StatusOK,
						Status:     "200 OK",
						Body:       io.NopCloser(strings.NewReader(`{"items":[{"index": {}}]}`)),
						Header:     http.Header{"X-Elastic-Product": []string{"Elasticsearch"}},
					}, nil
				},
			}),
		))
		if err != nil {
			t.Fatalf("Unexpected error: %s", err)
		}

		bi, err := NewBulkIndexer(BulkIndexerConfig{
			NumWorkers:    1,
			Client:        es,
			FlushInterval: 50 * time.Millisecond,
			FlushJitter:   50 * time.Millisecond,
		})
		if err != nil {
			t.Fatalf("Unexpected error: %s", err)
		}

		if err := bi.Add(context.Background(),
			BulkIndexerItem{Action: "index", Body: strings.NewReader(`{"title":"foo"}`)}); err != nil {
			t.Fatalf("Unexpected error: %s", err)
		}

		// Upper bound of the jittered interval is 100ms; 300ms is generous.
		time.Sleep(300 * time.Millisecond)

		stats := bi.Stats()
		if stats.NumFlushed != 1 {
			t.Errorf("Unexpected NumFlushed: want=1, got=%d", stats.NumFlushed)
		}

		// Let any in-flight ticker settle before Close to avoid racing the defer flush.
		time.Sleep(150 * time.Millisecond)
		bi.Close(context.Background())
	})

	t.Run("default FlushJitter is zero", func(t *testing.T) {
		es, err := elasticsearch.New(elasticsearch.WithTransportOptions(
			elastictransport.WithTransport(&mockTransport{
				RoundTripFunc: func(*http.Request) (*http.Response, error) {
					return &http.Response{
						StatusCode: http.StatusOK,
						Status:     "200 OK",
						Body:       io.NopCloser(strings.NewReader(`{}`)),
						Header:     http.Header{"X-Elastic-Product": []string{"Elasticsearch"}},
					}, nil
				},
			}),