feat[opensearchutil]: route subsequent bulk requests on the same documentID to the same worker. by yanaix10 · Pull Request #950 · opensearch-project/opensearch-go

yanaix10 · 2026-07-12T22:33:44Z

Changes made :

Struct Update - queue chan BulkIndexerItem -> queues []chan BulkIndexerItem sized to NumWorkers
Implemented OpenSearch shardhash.Hash() for documentID.
Item routed to hashValue % NumWorkers.
If no DocumentID is present, it falls back to a lock-free round-robin distribution via atomic.Uint64.
Updated close function : Iterates through and closes all worker channels.

codecov · 2026-07-12T22:36:12Z

Codecov Report

❌ Patch coverage is 68.18182% with 14 lines in your changes missing coverage. Please review.
✅ Project coverage is 55.46%. Comparing base (2f11b1b) to head (57b0b4f).
⚠️ Report is 11 commits behind head on main.

Files with missing lines	Patch %	Lines
opensearchutil/bulk_indexer.go	68.18%	12 Missing and 2 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #950      +/-   ##
==========================================
- Coverage   55.47%   55.46%   -0.02%     
==========================================
  Files         660      660              
  Lines       64338    64378      +40     
==========================================
+ Hits        35694    35705      +11     
- Misses      26804    26828      +24     
- Partials     1840     1845       +5

Flag	Coverage Δ
integration	`29.54% <68.18%> (+<0.01%)`	⬆️
unit	`54.18% <68.18%> (+<0.01%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
opensearchutil/bulk_indexer.go	`85.18% <68.18%> (-2.85%)`	⬇️

... and 6 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

yanaix10 · 2026-07-12T23:08:35Z

hey @sean- I am not very sure about the test case I wrote. I hardcorded the documentID.

sean-

Admittedly, I haven't looked into the opensearchutil/BulkIndexer much, but I'm not a fan of it in its current incarnation because it's interface is not particularly rigid. That said, I think its internals should be adapted to use the opensearchtransport/DocRouter until a v2 of this package can be put together to tighten up its interface.

sean- · 2026-07-15T15:16:19Z

 //
 // Adding an item after a call to Close() will panic.
 func (bi *bulkIndexer) Add(ctx context.Context, item BulkIndexerItem) error {
+	var targetQueue chan BulkIndexerItem


Here, we'd take a BulkIndexerItem, encapsulate it as an http.Request, then pass it along to DocRouter.Eval(). In fact, there may be a chance that the new routing code in v5 (and if opted-in, v4) may just "do this" out of the box on a best-effort basis, but to force this, we should have opensearchtuil/ construct a DocRouter and firehose things into the DocRouter.

sean- · 2026-07-15T15:17:48Z

 - Default the benchmark pprof server to an ephemeral loopback port so back-to-back `go test -bench` runs no longer collide on a `TIME_WAIT` socket held by the prior run. The startup logic moves into an `internal/pprofutil` package that registers the pprof handlers on a private mux (off `http.DefaultServeMux`); `PPROF_ADDR` pins an explicit `host:port` when needed. ([#864](https://github.com/opensearch-project/opensearch-go/issues/864))

+- Consistently route bulk requests for the same DocumentID to the same worker to prevent race conditions ([#950](https://github.com/opensearch-project/opensearch-go/pull/950)).
+


See https://github.com/opensearch-project/opensearch-go/blob/main/guides/transport-routing.md#document-path-docrouter and https://github.com/opensearch-project/opensearch-go/blob/main/guides/transport-routing.md#shared-connections-across-policies

yanaix10 · 2026-07-25T15:15:49Z

hey @sean- I have made required changes based on the feedback. Please review.

sean- · 2026-07-30T23:28:37Z

+			idx = bi.config.Index
+		}
+
+		u := url.URL{Path: path.Join("/", idx, "_doc", item.DocumentID)}


There's nothing here that escapes idx: we're just trusting the user input blindly. Can you look into url.PathUnescape() and look at the ./internal/build/ package and how it is used.

I have used url.PathEscape() on idx and documentID to handle user input. manually constructing the http.Request and explicitly setting Path (via url.PathUnescape()) and RawPath. Due to manual construction I could drop http.NewRequest entirely - bypassing url.Parse allocation overhead.

Because this is in the bulk-path, I'm not too worried about the performance, but it's up to you.

sean-

Thank you for adapting to using the router!

Signed-off-by: Naitik Yadav <naitik.yadav641@gmail.com>

…retain Eval Signed-off-by: Naitik Yadav <naitik.yadav641@gmail.com>

Signed-off-by: Naitik Yadav <naitik.yadav641@gmail.com>

ryanyuan · 2026-07-31T01:24:04Z

+
+		encodedPath := "/" + url.PathEscape(idx) + "/_doc/" + url.PathEscape(item.DocumentID)
+
+		if p, err := url.PathUnescape(encodedPath); err == nil {


Question about the escape/unescape pair here: does the escaping survive to Eval?
Would putting the escaped form straight into Path work for what you were after, dropping PathUnescape and RawPath?

ryanyuan · 2026-07-31T01:26:20Z

+				},
+			}).WithContext(ctx)
+
+			if hop, evalErr := bi.docRouter.Eval(ctx, req); evalErr == nil && hop.Conn != nil {


Does this Eval ever return a connection? I think NewDocRouter() gets its connections through DiscoveryUpdate(), which only the owning transport calls, so activeConns looks like it stays empty and Eval returns early at policy_doc_router.go:144-150.

yanaix10 requested review from Jakob3xD, VachaShah, VijayanB and sean- as code owners July 12, 2026 22:33

yanaix10 force-pushed the fix-#464 branch 2 times, most recently from 5155e4f to e7d9f58 Compare July 12, 2026 23:52

yanaix10 changed the title ~~feat(opensearchutil): route subsequent bulk requests on the same documentID to the same worker.~~ feat[opensearchutil]: route subsequent bulk requests on the same documentID to the same worker. Jul 12, 2026

sean- added the enhancement New feature or request label Jul 15, 2026

sean- reviewed Jul 15, 2026

View reviewed changes

sean- assigned yanaix10 Jul 15, 2026

sean- marked this pull request as draft July 17, 2026 15:12

yanaix10 force-pushed the fix-#464 branch from f831ba6 to b3f8d87 Compare July 25, 2026 14:56

sean- reviewed Jul 25, 2026

View reviewed changes

Comment thread opensearchutil/bulk_indexer.go Outdated

Comment thread opensearchutil/bulk_indexer.go Outdated

Comment thread opensearchutil/bulk_indexer.go Outdated

Comment thread opensearchutil/bulk_indexer.go Outdated

yanaix10 requested a review from sean- July 25, 2026 19:12

sean- reviewed Jul 30, 2026

View reviewed changes

sean- self-requested a review July 31, 2026 00:45

sean- approved these changes Jul 31, 2026

View reviewed changes

yanaix10 added 7 commits July 30, 2026 17:50

added feature request logic

b5b48db

Signed-off-by: Naitik Yadav <naitik.yadav641@gmail.com>

fixed linter issues

8f03ea9

Signed-off-by: Naitik Yadav <naitik.yadav641@gmail.com>

chore: update changelog for bulk indexer routing fix

01941fa

Signed-off-by: Naitik Yadav <naitik.yadav641@gmail.com>

Fixed .md linter issue.

581abd3

Signed-off-by: Naitik Yadav <naitik.yadav641@gmail.com>

updated changes with feedbacks given.

74fd793

Signed-off-by: Naitik Yadav <naitik.yadav641@gmail.com>

Refactor BulkIndexer routing: use RWMutex map, safe URL parsing, and …

e3f717c

…retain Eval Signed-off-by: Naitik Yadav <naitik.yadav641@gmail.com>

updated

57b0b4f

Signed-off-by: Naitik Yadav <naitik.yadav641@gmail.com>

sean- force-pushed the fix-#464 branch from fd5b446 to 57b0b4f Compare July 31, 2026 00:50

ryanyuan approved these changes Jul 31, 2026

View reviewed changes

		- Default the benchmark pprof server to an ephemeral loopback port so back-to-back `go test -bench` runs no longer collide on a `TIME_WAIT` socket held by the prior run. The startup logic moves into an `internal/pprofutil` package that registers the pprof handlers on a private mux (off `http.DefaultServeMux`); `PPROF_ADDR` pins an explicit `host:port` when needed. ([#864](https://github.com/opensearch-project/opensearch-go/issues/864))

		- Consistently route bulk requests for the same DocumentID to the same worker to prevent race conditions ([#950](https://github.com/opensearch-project/opensearch-go/pull/950)).


		encodedPath := "/" + url.PathEscape(idx) + "/_doc/" + url.PathEscape(item.DocumentID)

		if p, err := url.PathUnescape(encodedPath); err == nil {

Uh oh!

Conversation

yanaix10 commented Jul 12, 2026

Uh oh!

codecov Bot commented Jul 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

yanaix10 commented Jul 12, 2026

Uh oh!

sean- left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

sean- Jul 15, 2026

Choose a reason for hiding this comment

Uh oh!

sean- Jul 15, 2026

Choose a reason for hiding this comment

Uh oh!

yanaix10 commented Jul 25, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

sean- Jul 30, 2026

Choose a reason for hiding this comment

Uh oh!

yanaix10 Jul 31, 2026

Choose a reason for hiding this comment

Uh oh!

sean- Jul 31, 2026

Choose a reason for hiding this comment

Uh oh!

sean- left a comment

Choose a reason for hiding this comment

Uh oh!

ryanyuan Jul 31, 2026

Choose a reason for hiding this comment

Uh oh!

ryanyuan Jul 31, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

codecov Bot commented Jul 12, 2026 •

edited

Loading