Fix liveness goroutine leak in TaskListManager (#6563)

What changed?

Liveness.Stop should not call shutdownFn because there is not IsLive check. Even adding IsLive check in Stop would be confusing since the function literally means that shut down liveness only.
Liveness.eventLoop should not call Stop() which caused the deadlock (cyclic wait).
Why?

Liveness can cause deadlock which caused a goroutine leak

How did you test it?
Unit Test with go routine leak detector

service/matching/liveness/liveness.go

@@ -51,6 +51,8 @@ type (
 
 var _ common.Daemon = (*Liveness)(nil)
 
+// NewLiveness creates a Liveness daemon that calls the broadcastShutdownFn if it does not receive MarkAlive() within ttl
+// NOTE: livesness needs to be stopped explicitly to avoid go routine leak
 func NewLiveness(timeSource clock.TimeSource, ttl time.Duration, broadcastShutdownFn func()) *Liveness {
 	return &Liveness{
 		status:              common.DaemonStatusInitialized,
@@ -72,13 +74,13 @@ func (l *Liveness) Start() {
 	go l.eventLoop(checkTimer)
 }
 
+// Stop ONLY shuts down liveness does not block on broadcastShutdownFn
 func (l *Liveness) Stop() {
 	if !atomic.CompareAndSwapInt32(&l.status, common.DaemonStatusStarted, common.DaemonStatusStopped) {
 		return
 	}
 
 	close(l.stopCh)
-	l.broadcastShutdownFn()
 	l.wg.Wait()
 }
 
@@ -90,7 +92,8 @@ func (l *Liveness) eventLoop(ticker clock.Ticker) {
 		select {
 		case <-ticker.Chan():
 			if !l.IsAlive() {
-				l.Stop()
+				go l.broadcastShutdownFn() // do not block shutdown
+				return
 			}
 
 		case <-l.stopCh:

service/matching/liveness/liveness_test.go

@@ -55,7 +55,10 @@ func (s *livenessSuite) SetupTest() {
 	s.ttl = 500 * time.Millisecond
 	s.timeSource = clock.NewMockedTimeSource()
 	s.shutdownFlag = 0
-	s.liveness = NewLiveness(s.timeSource, s.ttl, func() { atomic.CompareAndSwapInt32(&s.shutdownFlag, 0, 1) })
+	s.liveness = NewLiveness(s.timeSource, s.ttl, func() {
+		atomic.CompareAndSwapInt32(&s.shutdownFlag, 0, 1)
+		s.liveness.Stop()
+	})
 }
 
 func (s *livenessSuite) TestIsAlive_No() {

service/matching/tasklist/task_list_manager_test.go

@@ -33,6 +33,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/uber-go/tally"
+	"go.uber.org/goleak"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/uber/cadence/client/history"
@@ -300,6 +301,7 @@ func TestDescribeTaskList(t *testing.T) {
 }
 
 func TestCheckIdleTaskList(t *testing.T) {
+	defer goleak.VerifyNone(t)
 	cfg := config.NewConfig(dynamicconfig.NewNopCollection(), "some random hostname", getIsolationgroupsHelper)
 	cfg.IdleTasklistCheckInterval = dynamicconfig.GetDurationPropertyFnFilteredByTaskListInfo(10 * time.Millisecond)