Fix visibility processor panic on add after stop

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

@@ -172,6 +172,11 @@ 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 p.mapToAckFuture == nil { // mapToAckFuture will be nil iff processor has been shut down
+		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
+	}
+
 	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() {