[FEATURE] Add an exported flush function to the BulkIndexer

[ae-ou]

Is your feature request related to a problem?

I have a Lambda that consumes events from a Kinesis stream and posts them to an OpenSearch Cluster - I use BulkIndexer to send in a large number of documents more efficiently (i.e. via the /bulk API).

Currently, I build up most of my dependencies (e.g. DB connection pools, AWS Signer, etc) in the main() function of my Lambda, and then I pass these dependencies to a struct. I create a handler function which receives the struct containing all of these dependencies - this way, as long as my Lambda stays warm, the dependencies are readily available for the handler to use on a subsequent invocation. e.g.:

package main

import (
	"context"
	"database/sql"
	"github.com/aws/aws-lambda-go/lambda"
)

//Dependencies can be reused across subsequent invocations of the Lambda - which speeds up non-cold starts (as we don't have to create the dependencies for injection)
type Dependencies struct {
	DbCon *sql.DB
	//... more dependencies
}

func main() {
	myDb := sql.OpenDB() //an imaginary DB that you still shouldn't call DROP on.

	d := Dependencies{
		DbCon: myDb,
	}

	lambda.Start(d.MyHandler)
}

func (d Dependencies) MyHandler(c context.Context) error {
	tx, err := d.DbCon.Begin()
	//Business logic here
}

I want to be able to pass the BulkIndexer on my dependency struct, but there's an issue. Your documentation for the BulkIndexer says:

// You must call the Close() method after you're done adding items.

The problem is that when you call Close() against BulkIndexer, you close queue (which is the channel that Add() appends requests to).
This creates a problem on subsequent invocations of the Lambda because if the channel is closed (by the call to Close() on a previous invocation), then calling Add() will result in trying to pass data to a nonexistent channel - which results in a panic.

This means that I can't store the BulkIndexer in my dependency struct because the channel underpinning the whole thing may be nonexistent on subsequent invocations. This means that I have to instantiate the BulkIndexer for each invocation of the Lambda (directly on the handler - even if the function is still warm).

What solution would you like?

I would like to be able to call Flush() directly against the BulkIndexer.

The Close() function already calls flush() against the workers, and this fires off whatever requests may be remaining in the queue (in addition to invoking the relevant callback functions).
If we were able to call BulkIndexer.Flush() directly, we could clear out whatever requests may remain in the queue (without closing it), and then call return on the Lambda afterwards - this would allow us to reuse the BulkIndexer on subsequent invocations of the Lambda (without risk of closing the channel/causing a panic).

What alternatives have you considered?

1. Setting BulkIndexerConfig.FlushInterval.
   a. The problem here is that we have to wait for the timer to hit 0 for the flush to occur - this means that we have to keep the Lambda running (which costs money and inflates our execution time metrics).
   b. There's also no clean way to see the amount of time until the next flush.
   c. You could set BulkIndexerConfig.FlushInterval to a low number - but if you're flushing/firing off requests every few seconds, you defeat the purpose of using the /bulk API.
2. Setting BulkIndexerConfig.FlushBytes
   a. The problem here is that we have to wait for the BulkIndexer to hit the threshold - which isn't always guaranteed to happen - e.g. if you've just flushed the data, but then you Add() one small request.
   b. You could set BulkIndexerConfig.FlushBytes to a low number - but if you're flushing/firing off requests after a small number of Add() calls, you defeat the purpose of using the /bulk API.
3. Create the BulkIndexer in the handler - every time that the Lambda is called
   a. This is what I do currently
   b. This means that you can call Close() at the end of your handler function (to flush any outstanding requests in the queue)
   c. This significantly increases the runtime of my Lambda (on every invocation), which has a direct impact on running cost.

Do you have any additional context?

Setting up dependencies outside of a Lambda's function handler is an AWS best practice - see here:

Take advantage of execution environment reuse to improve the performance of your function. Initialize SDK clients and database connections outside of the function handler, and cache static assets locally in the /tmp directory. Subsequent invocations processed by the same instance of your function can reuse these resources. This saves cost by reducing function run time.

[wbeckler]

I've never looked at this before, so assume I might be way off here. If the docs say to call close after "you're done adding items," then might it work if you don't call close until all processes/workers have finished adding items. And since you never finish, you never call close()? Again, I hope I'm not wasting your time with this question, but I thought it worth asking with what little I know.

[ae-ou]

Hi @wbeckler. My handler fires off goroutines for each Kinesis record so that I can call BulkIndexer.Add() for multiple records concurrenttly, I then have a sync.WaitGroup that waits for all of these goroutines to complete before I call Bulkindexer.Close() in the handler. At the point that I call Close(), I already know that there are no more additions to be made (on this invocation of the Lambda).

The reason that I have to call Close() is that it forces a flush of the queue - meaning that the workers send all of the remaining queued up requests to OpenSearch (if I understand correctly). I have to do this before the handler returns because the alternative is for the handler to return with requests (potentially) still in the BulkIndexer.queue - getting stale.

Hypothetically, if the Lambda stays warm and a subsequent invocation occurs, any stale requests (from a previous invocation of the Lambda) should still be in the BulkIndexer.queue, and they should be fired off (I haven't confirmed this). The issue is that there's no guarantee of a subsequent invocation of the Lambda - if there isn't one, the stored data will be killed off (including the BulkIndexer) and the next invocation will be a cold start - potentially losing queued requests in the process.

[dblock]

I think adding .Flush() makes sense, want to give it a try?

[ansssu]

@ae-ou , In my opinion, this feature request doesn't make sense, and your approach to handling the bulk inserts represents a potential risk of data loss. I will try to explain why.

An AWS lambda runs in a container environment. The first time a function executes after being created or having its code or configuration updated, a new container will be created. The code is loaded into this container for execution. Subsequently, your code performs its tasks, and the Lambda finishes.

Later on, when you invoke the AWS Lambda again, it can follow two paths: If the code hasn't changed or not much time has gone by, Lambda MAY reuse the previous container and its shared resources or the Lambda MAY create a new execution container from scratch. This behavior isn't something you can control; it's determined by the Lambda itself.

With that said, you should avoid storing data in shared resources. If you run the Lambda and save the data into shared resources and, in the next Lambda execution, the lambda decides to create a new container, then you will lose all data that hasn't been flushed in the previous Lambda execution.

You are trying to make the Lambda function operate in a stateful manner, whereas the lambda's intended behavior is to be completely stateless.

In this scenario, my recommendation is to create the bulk indexer within the handler each time the Lambda is invoked.

[yves-tutti]

Outside of the described Lambda scenario on which I can't comment, having a Flush() on the bulk-indexer comes handy when writing test-code. It's required when trying port code that was written using https://github.com/olivere/elastic/blob/29ee98974cf1984dfecf53ef772d721fb97cb0b9/bulk_processor.go#L409

[LeeFred3042U]

Can i take up this issue?

[sean-]

@LeeFred3042U, go for it.

Close() panics future Add() calls because it close(bi.queue) (opensearchutil/bulk_indexer.go:263). Goal: drain submitted items without closing, so a warm Lambda can reuse the indexer.

The design to try

Replace the work channel with a ring buffer carrying a monotonic write position, and make Flush a watermark wait: "drain everything submitted before me," not best-effort.

1. Ring buffer of items with a monotonic write sequence (the "clock").
2. Add writes into the ring, advancing the write position; returns its sequence.
3. One worker pops items in order and indexes them, advancing a committed position.
4. Flush(ctx) snapshots the current write position w, then blocks until committed >= w. Items added after the snapshot are not waited on.
5. Concurrent flushers each hold their own w, so they unblock independently as the committed position passes each watermark.

This is why the approach fits (and why the earlier ideas didn't): it's one consumer + multiple commit watermarks, not multiple consumers replaying. The watermark gives a precise "everything before this Flush is sent" guarantee instead of the ticker's best-effort buffer flush.

Pitfalls to investigate before committing to it

• Close — the bug (close(bi.queue))
• Add
• ticker auto-flush (the safe pattern)
• worker.flush (the /_bulk send + Stats)
• worker.run (concurrent buffer access)
• queue init / cap

[LeeFred3042U]

@sean- one question i have to stay in go 1.25 or 1.26 to implement this?

[sean-]

@LeeFred3042U For the next month or two, stay with Go 1.25. What out of Go 1.26 are you trying to use?

[LeeFred3042U]

@LeeFred3042U For the next month or two, stay with Go 1.25. What out of Go 1.26 are you trying to use?

I dont think i need any feature of go 1.26 for this specific issue, since most of whas needed is available in both version as standard
i just wanted to confirm since i am using go 1.25,