329
bulkRequest *client.BulkableRequest,
330
visibilityTaskKey string,
332
>
s.checkProcessor()
333
>
334
>
// Add method is blocking. If bulk processor is busy flushing previous bulk, request will wait here.
335
>
ackF := s.processor.Add(bulkRequest, visibilityTaskKey)
336
>
337
>
// processorAckTimeout is a maximum duration for bulk processor to commit the bulk and unblock the `ackF`.
338
>
// The default value is 30s, and this timeout should never have happened,
339
>
// because Elasticsearch must process a bulk within the 30s.
340
>
// Parent context is not respected here because it has shorter timeout (3s),
341
>
// which might already expired here due to wait at Add method above.
342
>
ctx, cancel := context.WithTimeout(context.Background(), s.processorAckTimeout())
343
>
defer cancel()
344
>
ack, err := ackF.Get(ctx)
345
>
346
>
if err != nil {
347
if errors.Is(err, context.DeadlineExceeded) {
348
return &persistence.TimeoutError{Msg: fmt.Sprintf("visibility task timed out waiting for ACK after %v", s.processorAckTimeout())}