Fix visibility processor panic on add after stop

common/persistence/visibility/store/elasticsearch/processor.go

@@ -33,6 +33,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -69,6 +70,7 @@ type (
 		logger                  log.Logger
 		metricsHandler          metrics.Handler
 		indexerConcurrency      uint32
+		inflightAdds            sync.WaitGroup
 	}
 
 	// ProcessorConfig contains all configs for processor
@@ -150,6 +152,8 @@ func (p *processorImpl) Stop() {
 		return
 	}
 
+	p.inflightAdds.Wait()
+
 	err := p.bulkProcessor.Stop()
 	if err != nil {
 		// This could happen if ES is down when we're trying to shut down the server.
@@ -172,6 +176,15 @@ func (p *processorImpl) hashFn(key interface{}) uint32 {
 
 // Add request to the bulk and return a future object which will receive ack signal when request is processed.
 func (p *processorImpl) Add(request *client.BulkableRequest, visibilityTaskKey string) *future.FutureImpl[bool] {
+	if atomic.LoadInt32(&p.status) == common.DaemonStatusStopped {
+		p.logger.Warn("Skipping ES request for visibility task key because processor has been shut down.", tag.Key(visibilityTaskKey), tag.ESDocID(request.ID), tag.Value(request.Doc))
+		return nil
+	}
+
+	// count the number of requests to prevent shutting down in the middle of an Add
+	p.inflightAdds.Add(1)
+	defer p.inflightAdds.Done()
+
 	newFuture := newAckFuture()
 	_, isDup, _ := p.mapToAckFuture.PutOrDo(visibilityTaskKey, newFuture, func(key interface{}, value interface{}) error {
 		existingFuture, ok := value.(*ackFuture)

common/persistence/visibility/store/elasticsearch/processor_test.go

@@ -128,6 +128,12 @@ func (s *processorSuite) TestNewESProcessorAndStartStop() {
 	p.Stop()
 	s.Nil(p.mapToAckFuture)
 	s.Nil(p.bulkProcessor)
+
+	// Confirm processor does not panic on new requests after stopping
+	request := &client.BulkableRequest{}
+	visibilityTaskKey := "test-key"
+	future1 := p.Add(request, visibilityTaskKey)
+	s.Nil(future1)
 }
 
 func (s *processorSuite) TestAdd() {