forked from p85947160/gitea
657 lines
19 KiB
Go
657 lines
19 KiB
Go
// Copyright 2012-present Oliver Eilhard. All rights reserved.
|
|
// Use of this source code is governed by a MIT-license.
|
|
// See http://olivere.mit-license.org/license.txt for details.
|
|
|
|
package elastic
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"net"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
)
|
|
|
|
var (
|
|
// ErrBulkItemRetry is returned in BulkProcessor from a worker when
|
|
// a response item needs to be retried.
|
|
ErrBulkItemRetry = errors.New("elastic: uncommitted bulk response items")
|
|
|
|
defaultRetryItemStatusCodes = []int{408, 429, 503, 507}
|
|
)
|
|
|
|
// BulkProcessorService allows to easily process bulk requests. It allows setting
|
|
// policies when to flush new bulk requests, e.g. based on a number of actions,
|
|
// on the size of the actions, and/or to flush periodically. It also allows
|
|
// to control the number of concurrent bulk requests allowed to be executed
|
|
// in parallel.
|
|
//
|
|
// BulkProcessorService, by default, commits either every 1000 requests or when the
|
|
// (estimated) size of the bulk requests exceeds 5 MB. However, it does not
|
|
// commit periodically. BulkProcessorService also does retry by default, using
|
|
// an exponential backoff algorithm. It also will automatically re-enqueue items
|
|
// returned with a status of 408, 429, 503 or 507. You can change this
|
|
// behavior with RetryItemStatusCodes.
|
|
//
|
|
// The caller is responsible for setting the index and type on every
|
|
// bulk request added to BulkProcessorService.
|
|
//
|
|
// BulkProcessorService takes ideas from the BulkProcessor of the
|
|
// Elasticsearch Java API as documented in
|
|
// https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/java-docs-bulk-processor.html.
|
|
type BulkProcessorService struct {
|
|
c *Client
|
|
beforeFn BulkBeforeFunc
|
|
afterFn BulkAfterFunc
|
|
name string // name of processor
|
|
numWorkers int // # of workers (>= 1)
|
|
bulkActions int // # of requests after which to commit
|
|
bulkSize int // # of bytes after which to commit
|
|
flushInterval time.Duration // periodic flush interval
|
|
wantStats bool // indicates whether to gather statistics
|
|
backoff Backoff // a custom Backoff to use for errors
|
|
retryItemStatusCodes []int // array of status codes for bulk response line items that may be retried
|
|
}
|
|
|
|
// NewBulkProcessorService creates a new BulkProcessorService.
|
|
func NewBulkProcessorService(client *Client) *BulkProcessorService {
|
|
return &BulkProcessorService{
|
|
c: client,
|
|
numWorkers: 1,
|
|
bulkActions: 1000,
|
|
bulkSize: 5 << 20, // 5 MB
|
|
backoff: NewExponentialBackoff(
|
|
time.Duration(200)*time.Millisecond,
|
|
time.Duration(10000)*time.Millisecond,
|
|
),
|
|
retryItemStatusCodes: defaultRetryItemStatusCodes,
|
|
}
|
|
}
|
|
|
|
// BulkBeforeFunc defines the signature of callbacks that are executed
|
|
// before a commit to Elasticsearch.
|
|
type BulkBeforeFunc func(executionId int64, requests []BulkableRequest)
|
|
|
|
// BulkAfterFunc defines the signature of callbacks that are executed
|
|
// after a commit to Elasticsearch. The err parameter signals an error.
|
|
type BulkAfterFunc func(executionId int64, requests []BulkableRequest, response *BulkResponse, err error)
|
|
|
|
// Before specifies a function to be executed before bulk requests get committed
|
|
// to Elasticsearch.
|
|
func (s *BulkProcessorService) Before(fn BulkBeforeFunc) *BulkProcessorService {
|
|
s.beforeFn = fn
|
|
return s
|
|
}
|
|
|
|
// After specifies a function to be executed when bulk requests have been
|
|
// committed to Elasticsearch. The After callback executes both when the
|
|
// commit was successful as well as on failures.
|
|
func (s *BulkProcessorService) After(fn BulkAfterFunc) *BulkProcessorService {
|
|
s.afterFn = fn
|
|
return s
|
|
}
|
|
|
|
// Name is an optional name to identify this bulk processor.
|
|
func (s *BulkProcessorService) Name(name string) *BulkProcessorService {
|
|
s.name = name
|
|
return s
|
|
}
|
|
|
|
// Workers is the number of concurrent workers allowed to be
|
|
// executed. Defaults to 1 and must be greater or equal to 1.
|
|
func (s *BulkProcessorService) Workers(num int) *BulkProcessorService {
|
|
s.numWorkers = num
|
|
return s
|
|
}
|
|
|
|
// BulkActions specifies when to flush based on the number of actions
|
|
// currently added. Defaults to 1000 and can be set to -1 to be disabled.
|
|
func (s *BulkProcessorService) BulkActions(bulkActions int) *BulkProcessorService {
|
|
s.bulkActions = bulkActions
|
|
return s
|
|
}
|
|
|
|
// BulkSize specifies when to flush based on the size (in bytes) of the actions
|
|
// currently added. Defaults to 5 MB and can be set to -1 to be disabled.
|
|
func (s *BulkProcessorService) BulkSize(bulkSize int) *BulkProcessorService {
|
|
s.bulkSize = bulkSize
|
|
return s
|
|
}
|
|
|
|
// FlushInterval specifies when to flush at the end of the given interval.
|
|
// This is disabled by default. If you want the bulk processor to
|
|
// operate completely asynchronously, set both BulkActions and BulkSize to
|
|
// -1 and set the FlushInterval to a meaningful interval.
|
|
func (s *BulkProcessorService) FlushInterval(interval time.Duration) *BulkProcessorService {
|
|
s.flushInterval = interval
|
|
return s
|
|
}
|
|
|
|
// Stats tells bulk processor to gather stats while running.
|
|
// Use Stats to return the stats. This is disabled by default.
|
|
func (s *BulkProcessorService) Stats(wantStats bool) *BulkProcessorService {
|
|
s.wantStats = wantStats
|
|
return s
|
|
}
|
|
|
|
// Backoff sets the backoff strategy to use for errors.
|
|
func (s *BulkProcessorService) Backoff(backoff Backoff) *BulkProcessorService {
|
|
s.backoff = backoff
|
|
return s
|
|
}
|
|
|
|
// RetryItemStatusCodes sets an array of status codes that indicate that a bulk
|
|
// response line item should be retried.
|
|
func (s *BulkProcessorService) RetryItemStatusCodes(retryItemStatusCodes ...int) *BulkProcessorService {
|
|
s.retryItemStatusCodes = retryItemStatusCodes
|
|
return s
|
|
}
|
|
|
|
// Do creates a new BulkProcessor and starts it.
|
|
// Consider the BulkProcessor as a running instance that accepts bulk requests
|
|
// and commits them to Elasticsearch, spreading the work across one or more
|
|
// workers.
|
|
//
|
|
// You can interoperate with the BulkProcessor returned by Do, e.g. Start and
|
|
// Stop (or Close) it.
|
|
//
|
|
// Context is an optional context that is passed into the bulk request
|
|
// service calls. In contrast to other operations, this context is used in
|
|
// a long running process. You could use it to pass e.g. loggers, but you
|
|
// shouldn't use it for cancellation.
|
|
//
|
|
// Calling Do several times returns new BulkProcessors. You probably don't
|
|
// want to do this. BulkProcessorService implements just a builder pattern.
|
|
func (s *BulkProcessorService) Do(ctx context.Context) (*BulkProcessor, error) {
|
|
|
|
retryItemStatusCodes := make(map[int]struct{})
|
|
for _, code := range s.retryItemStatusCodes {
|
|
retryItemStatusCodes[code] = struct{}{}
|
|
}
|
|
|
|
p := newBulkProcessor(
|
|
s.c,
|
|
s.beforeFn,
|
|
s.afterFn,
|
|
s.name,
|
|
s.numWorkers,
|
|
s.bulkActions,
|
|
s.bulkSize,
|
|
s.flushInterval,
|
|
s.wantStats,
|
|
s.backoff,
|
|
retryItemStatusCodes)
|
|
|
|
err := p.Start(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return p, nil
|
|
}
|
|
|
|
// -- Bulk Processor Statistics --
|
|
|
|
// BulkProcessorStats contains various statistics of a bulk processor
|
|
// while it is running. Use the Stats func to return it while running.
|
|
type BulkProcessorStats struct {
|
|
Flushed int64 // number of times the flush interval has been invoked
|
|
Committed int64 // # of times workers committed bulk requests
|
|
Indexed int64 // # of requests indexed
|
|
Created int64 // # of requests that ES reported as creates (201)
|
|
Updated int64 // # of requests that ES reported as updates
|
|
Deleted int64 // # of requests that ES reported as deletes
|
|
Succeeded int64 // # of requests that ES reported as successful
|
|
Failed int64 // # of requests that ES reported as failed
|
|
|
|
Workers []*BulkProcessorWorkerStats // stats for each worker
|
|
}
|
|
|
|
// BulkProcessorWorkerStats represents per-worker statistics.
|
|
type BulkProcessorWorkerStats struct {
|
|
Queued int64 // # of requests queued in this worker
|
|
LastDuration time.Duration // duration of last commit
|
|
}
|
|
|
|
// newBulkProcessorStats initializes and returns a BulkProcessorStats struct.
|
|
func newBulkProcessorStats(workers int) *BulkProcessorStats {
|
|
stats := &BulkProcessorStats{
|
|
Workers: make([]*BulkProcessorWorkerStats, workers),
|
|
}
|
|
for i := 0; i < workers; i++ {
|
|
stats.Workers[i] = &BulkProcessorWorkerStats{}
|
|
}
|
|
return stats
|
|
}
|
|
|
|
func (st *BulkProcessorStats) dup() *BulkProcessorStats {
|
|
dst := new(BulkProcessorStats)
|
|
dst.Flushed = st.Flushed
|
|
dst.Committed = st.Committed
|
|
dst.Indexed = st.Indexed
|
|
dst.Created = st.Created
|
|
dst.Updated = st.Updated
|
|
dst.Deleted = st.Deleted
|
|
dst.Succeeded = st.Succeeded
|
|
dst.Failed = st.Failed
|
|
for _, src := range st.Workers {
|
|
dst.Workers = append(dst.Workers, src.dup())
|
|
}
|
|
return dst
|
|
}
|
|
|
|
func (st *BulkProcessorWorkerStats) dup() *BulkProcessorWorkerStats {
|
|
dst := new(BulkProcessorWorkerStats)
|
|
dst.Queued = st.Queued
|
|
dst.LastDuration = st.LastDuration
|
|
return dst
|
|
}
|
|
|
|
// -- Bulk Processor --
|
|
|
|
// BulkProcessor encapsulates a task that accepts bulk requests and
|
|
// orchestrates committing them to Elasticsearch via one or more workers.
|
|
//
|
|
// BulkProcessor is returned by setting up a BulkProcessorService and
|
|
// calling the Do method.
|
|
type BulkProcessor struct {
|
|
c *Client
|
|
beforeFn BulkBeforeFunc
|
|
afterFn BulkAfterFunc
|
|
name string
|
|
bulkActions int
|
|
bulkSize int
|
|
numWorkers int
|
|
executionId int64
|
|
requestsC chan BulkableRequest
|
|
workerWg sync.WaitGroup
|
|
workers []*bulkWorker
|
|
flushInterval time.Duration
|
|
flusherStopC chan struct{}
|
|
wantStats bool
|
|
retryItemStatusCodes map[int]struct{}
|
|
backoff Backoff
|
|
|
|
startedMu sync.Mutex // guards the following block
|
|
started bool
|
|
|
|
statsMu sync.Mutex // guards the following block
|
|
stats *BulkProcessorStats
|
|
|
|
stopReconnC chan struct{} // channel to signal stop reconnection attempts
|
|
}
|
|
|
|
func newBulkProcessor(
|
|
client *Client,
|
|
beforeFn BulkBeforeFunc,
|
|
afterFn BulkAfterFunc,
|
|
name string,
|
|
numWorkers int,
|
|
bulkActions int,
|
|
bulkSize int,
|
|
flushInterval time.Duration,
|
|
wantStats bool,
|
|
backoff Backoff,
|
|
retryItemStatusCodes map[int]struct{}) *BulkProcessor {
|
|
return &BulkProcessor{
|
|
c: client,
|
|
beforeFn: beforeFn,
|
|
afterFn: afterFn,
|
|
name: name,
|
|
numWorkers: numWorkers,
|
|
bulkActions: bulkActions,
|
|
bulkSize: bulkSize,
|
|
flushInterval: flushInterval,
|
|
wantStats: wantStats,
|
|
retryItemStatusCodes: retryItemStatusCodes,
|
|
backoff: backoff,
|
|
}
|
|
}
|
|
|
|
// Start starts the bulk processor. If the processor is already started,
|
|
// nil is returned.
|
|
func (p *BulkProcessor) Start(ctx context.Context) error {
|
|
p.startedMu.Lock()
|
|
defer p.startedMu.Unlock()
|
|
|
|
if p.started {
|
|
return nil
|
|
}
|
|
|
|
// We must have at least one worker.
|
|
if p.numWorkers < 1 {
|
|
p.numWorkers = 1
|
|
}
|
|
|
|
p.requestsC = make(chan BulkableRequest)
|
|
p.executionId = 0
|
|
p.stats = newBulkProcessorStats(p.numWorkers)
|
|
p.stopReconnC = make(chan struct{})
|
|
|
|
// Create and start up workers.
|
|
p.workers = make([]*bulkWorker, p.numWorkers)
|
|
for i := 0; i < p.numWorkers; i++ {
|
|
p.workerWg.Add(1)
|
|
p.workers[i] = newBulkWorker(p, i)
|
|
go p.workers[i].work(ctx)
|
|
}
|
|
|
|
// Start the ticker for flush (if enabled)
|
|
if int64(p.flushInterval) > 0 {
|
|
p.flusherStopC = make(chan struct{})
|
|
go p.flusher(p.flushInterval)
|
|
}
|
|
|
|
p.started = true
|
|
|
|
return nil
|
|
}
|
|
|
|
// Stop is an alias for Close.
|
|
func (p *BulkProcessor) Stop() error {
|
|
return p.Close()
|
|
}
|
|
|
|
// Close stops the bulk processor previously started with Do.
|
|
// If it is already stopped, this is a no-op and nil is returned.
|
|
//
|
|
// By implementing Close, BulkProcessor implements the io.Closer interface.
|
|
func (p *BulkProcessor) Close() error {
|
|
p.startedMu.Lock()
|
|
defer p.startedMu.Unlock()
|
|
|
|
// Already stopped? Do nothing.
|
|
if !p.started {
|
|
return nil
|
|
}
|
|
|
|
// Tell connection checkers to stop
|
|
if p.stopReconnC != nil {
|
|
close(p.stopReconnC)
|
|
p.stopReconnC = nil
|
|
}
|
|
|
|
// Stop flusher (if enabled)
|
|
if p.flusherStopC != nil {
|
|
p.flusherStopC <- struct{}{}
|
|
<-p.flusherStopC
|
|
close(p.flusherStopC)
|
|
p.flusherStopC = nil
|
|
}
|
|
|
|
// Stop all workers.
|
|
close(p.requestsC)
|
|
p.workerWg.Wait()
|
|
|
|
p.started = false
|
|
|
|
return nil
|
|
}
|
|
|
|
// Stats returns the latest bulk processor statistics.
|
|
// Collecting stats must be enabled first by calling Stats(true) on
|
|
// the service that created this processor.
|
|
func (p *BulkProcessor) Stats() BulkProcessorStats {
|
|
p.statsMu.Lock()
|
|
defer p.statsMu.Unlock()
|
|
return *p.stats.dup()
|
|
}
|
|
|
|
// Add adds a single request to commit by the BulkProcessorService.
|
|
//
|
|
// The caller is responsible for setting the index and type on the request.
|
|
func (p *BulkProcessor) Add(request BulkableRequest) {
|
|
p.requestsC <- request
|
|
}
|
|
|
|
// Flush manually asks all workers to commit their outstanding requests.
|
|
// It returns only when all workers acknowledge completion.
|
|
func (p *BulkProcessor) Flush() error {
|
|
p.statsMu.Lock()
|
|
p.stats.Flushed++
|
|
p.statsMu.Unlock()
|
|
|
|
for _, w := range p.workers {
|
|
w.flushC <- struct{}{}
|
|
<-w.flushAckC // wait for completion
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// flusher is a single goroutine that periodically asks all workers to
|
|
// commit their outstanding bulk requests. It is only started if
|
|
// FlushInterval is greater than 0.
|
|
func (p *BulkProcessor) flusher(interval time.Duration) {
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ticker.C: // Periodic flush
|
|
p.Flush() // TODO swallow errors here?
|
|
|
|
case <-p.flusherStopC:
|
|
p.flusherStopC <- struct{}{}
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// -- Bulk Worker --
|
|
|
|
// bulkWorker encapsulates a single worker, running in a goroutine,
|
|
// receiving bulk requests and eventually committing them to Elasticsearch.
|
|
// It is strongly bound to a BulkProcessor.
|
|
type bulkWorker struct {
|
|
p *BulkProcessor
|
|
i int
|
|
bulkActions int
|
|
bulkSize int
|
|
service *BulkService
|
|
flushC chan struct{}
|
|
flushAckC chan struct{}
|
|
}
|
|
|
|
// newBulkWorker creates a new bulkWorker instance.
|
|
func newBulkWorker(p *BulkProcessor, i int) *bulkWorker {
|
|
return &bulkWorker{
|
|
p: p,
|
|
i: i,
|
|
bulkActions: p.bulkActions,
|
|
bulkSize: p.bulkSize,
|
|
service: NewBulkService(p.c),
|
|
flushC: make(chan struct{}),
|
|
flushAckC: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
// work waits for bulk requests and manual flush calls on the respective
|
|
// channels and is invoked as a goroutine when the bulk processor is started.
|
|
func (w *bulkWorker) work(ctx context.Context) {
|
|
defer func() {
|
|
w.p.workerWg.Done()
|
|
close(w.flushAckC)
|
|
close(w.flushC)
|
|
}()
|
|
|
|
var stop bool
|
|
for !stop {
|
|
var err error
|
|
select {
|
|
case req, open := <-w.p.requestsC:
|
|
if open {
|
|
// Received a new request
|
|
if _, err = req.Source(); err == nil {
|
|
w.service.Add(req)
|
|
if w.commitRequired() {
|
|
err = w.commit(ctx)
|
|
}
|
|
}
|
|
} else {
|
|
// Channel closed: Stop.
|
|
stop = true
|
|
if w.service.NumberOfActions() > 0 {
|
|
err = w.commit(ctx)
|
|
}
|
|
}
|
|
|
|
case <-w.flushC:
|
|
// Commit outstanding requests
|
|
if w.service.NumberOfActions() > 0 {
|
|
err = w.commit(ctx)
|
|
}
|
|
w.flushAckC <- struct{}{}
|
|
}
|
|
if err != nil {
|
|
w.p.c.errorf("elastic: bulk processor %q was unable to perform work: %v", w.p.name, err)
|
|
if !stop {
|
|
waitForActive := func() {
|
|
// Add back pressure to prevent Add calls from filling up the request queue
|
|
ready := make(chan struct{})
|
|
go w.waitForActiveConnection(ready)
|
|
<-ready
|
|
}
|
|
if _, ok := err.(net.Error); ok {
|
|
waitForActive()
|
|
} else if IsConnErr(err) {
|
|
waitForActive()
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// commit commits the bulk requests in the given service,
|
|
// invoking callbacks as specified.
|
|
func (w *bulkWorker) commit(ctx context.Context) error {
|
|
var res *BulkResponse
|
|
|
|
// commitFunc will commit bulk requests and, on failure, be retried
|
|
// via exponential backoff
|
|
commitFunc := func() error {
|
|
var err error
|
|
// Save requests because they will be reset in service.Do
|
|
reqs := w.service.requests
|
|
res, err = w.service.Do(ctx)
|
|
if err == nil {
|
|
// Overall bulk request was OK. But each bulk response item also has a status
|
|
if w.p.retryItemStatusCodes != nil && len(w.p.retryItemStatusCodes) > 0 {
|
|
// Check res.Items since some might be soft failures
|
|
if res.Items != nil && res.Errors {
|
|
// res.Items will be 1 to 1 with reqs in same order
|
|
for i, item := range res.Items {
|
|
for _, result := range item {
|
|
if _, found := w.p.retryItemStatusCodes[result.Status]; found {
|
|
w.service.Add(reqs[i])
|
|
if err == nil {
|
|
err = ErrBulkItemRetry
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return err
|
|
}
|
|
// notifyFunc will be called if retry fails
|
|
notifyFunc := func(err error) {
|
|
w.p.c.errorf("elastic: bulk processor %q failed but may retry: %v", w.p.name, err)
|
|
}
|
|
|
|
id := atomic.AddInt64(&w.p.executionId, 1)
|
|
|
|
// Update # documents in queue before eventual retries
|
|
w.p.statsMu.Lock()
|
|
if w.p.wantStats {
|
|
w.p.stats.Workers[w.i].Queued = int64(len(w.service.requests))
|
|
}
|
|
w.p.statsMu.Unlock()
|
|
|
|
// Save requests because they will be reset in commitFunc
|
|
reqs := w.service.requests
|
|
|
|
// Invoke before callback
|
|
if w.p.beforeFn != nil {
|
|
w.p.beforeFn(id, reqs)
|
|
}
|
|
|
|
// Commit bulk requests
|
|
err := RetryNotify(commitFunc, w.p.backoff, notifyFunc)
|
|
w.updateStats(res)
|
|
if err != nil {
|
|
w.p.c.errorf("elastic: bulk processor %q failed: %v", w.p.name, err)
|
|
}
|
|
|
|
// Invoke after callback
|
|
if w.p.afterFn != nil {
|
|
w.p.afterFn(id, reqs, res, err)
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
func (w *bulkWorker) waitForActiveConnection(ready chan<- struct{}) {
|
|
defer close(ready)
|
|
|
|
t := time.NewTicker(5 * time.Second)
|
|
defer t.Stop()
|
|
|
|
client := w.p.c
|
|
stopReconnC := w.p.stopReconnC
|
|
w.p.c.errorf("elastic: bulk processor %q is waiting for an active connection", w.p.name)
|
|
|
|
// loop until a health check finds at least 1 active connection or the reconnection channel is closed
|
|
for {
|
|
select {
|
|
case _, ok := <-stopReconnC:
|
|
if !ok {
|
|
w.p.c.errorf("elastic: bulk processor %q active connection check interrupted", w.p.name)
|
|
return
|
|
}
|
|
case <-t.C:
|
|
client.healthcheck(context.Background(), 3*time.Second, true)
|
|
if client.mustActiveConn() == nil {
|
|
// found an active connection
|
|
// exit and signal done to the WaitGroup
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (w *bulkWorker) updateStats(res *BulkResponse) {
|
|
// Update stats
|
|
if res != nil {
|
|
w.p.statsMu.Lock()
|
|
if w.p.wantStats {
|
|
w.p.stats.Committed++
|
|
if res != nil {
|
|
w.p.stats.Indexed += int64(len(res.Indexed()))
|
|
w.p.stats.Created += int64(len(res.Created()))
|
|
w.p.stats.Updated += int64(len(res.Updated()))
|
|
w.p.stats.Deleted += int64(len(res.Deleted()))
|
|
w.p.stats.Succeeded += int64(len(res.Succeeded()))
|
|
w.p.stats.Failed += int64(len(res.Failed()))
|
|
}
|
|
w.p.stats.Workers[w.i].Queued = int64(len(w.service.requests))
|
|
w.p.stats.Workers[w.i].LastDuration = time.Duration(int64(res.Took)) * time.Millisecond
|
|
}
|
|
w.p.statsMu.Unlock()
|
|
}
|
|
}
|
|
|
|
// commitRequired returns true if the service has to commit its
|
|
// bulk requests. This can be either because the number of actions
|
|
// or the estimated size in bytes is larger than specified in the
|
|
// BulkProcessorService.
|
|
func (w *bulkWorker) commitRequired() bool {
|
|
if w.bulkActions >= 0 && w.service.NumberOfActions() >= w.bulkActions {
|
|
return true
|
|
}
|
|
if w.bulkSize >= 0 && w.service.EstimatedSizeInBytes() >= int64(w.bulkSize) {
|
|
return true
|
|
}
|
|
return false
|
|
}
|