153
154
// Add request to the bulk and return a future object which will receive ack signal when the request is processed.
155
>
func (p *processorImpl) Add(request *client.BulkableRequest, visibilityTaskKey string) *future.FutureImpl[bool] {
processor.go
156
>
newFuture := newAckFuture() // Create future first to measure the impact of following RWLock on latency.
157
>
158
>
p.shutdownLock.RLock()
159
>
defer p.shutdownLock.RUnlock()
160
>
161
>
if atomic.LoadInt32(&p.status) == common.DaemonStatusStopped {
162
p.logger.Warn("Rejecting ES request for visibility task key because processor has been shut down.", tag.Key(visibilityTaskKey), tag.ESDocID(request.ID), tag.Value(request.Doc))
163
newFuture.future.Set(false, errVisibilityShutdown)