bugfix goroutine-leak in statsloop in partition table

partition_processor.go

@@ -182,6 +182,7 @@ func (pp *PartitionProcessor) Setup(ctx context.Context) error {
 	defer pp.state.SetState(PPStateRunning)
 
 	if pp.table != nil {
+		go pp.table.RunStatsLoop(runnerCtx)
 		setupErrg.Go(func() error {
 			pp.log.Debugf("catching up table")
 			defer pp.log.Debugf("catching up table done")
@@ -202,6 +203,7 @@ func (pp *PartitionProcessor) Setup(ctx context.Context) error {
 		)
 		pp.joins[join.Topic()] = table
 
+		go table.RunStatsLoop(runnerCtx)
 		setupErrg.Go(func() error {
 			return table.SetupAndRecover(setupCtx, false)
 		})
@@ -358,6 +360,10 @@ func (pp *PartitionProcessor) enqueueStatsUpdate(ctx context.Context, updater fu
 	select {
 	case pp.updateStats <- updater:
 	case <-ctx.Done():
+	default:
+		// going to default indicates the updateStats channel is not read, so so the stats
+		// loop is not actually running.
+		// We must not block here, so we'll skip the update
 	}
 }
 

partition_table.go

@@ -42,11 +42,10 @@ type PartitionTable struct {
 	tmgr           TopicManager
 	updateCallback UpdateCallback
 
-	stats           *TableStats
-	cancelStatsLoop context.CancelFunc
-	requestStats    chan bool
-	responseStats   chan *TableStats
-	updateStats     chan func()
+	stats         *TableStats
+	requestStats  chan bool
+	responseStats chan *TableStats
+	updateStats   chan func()
 
 	offsetM sync.Mutex
 	// current offset
@@ -82,8 +81,6 @@ func newPartitionTable(topic string,
 	backoff Backoff,
 	backoffResetTimeout time.Duration) *PartitionTable {
 
-	statsLoopCtx, cancel := context.WithCancel(context.Background())
-
 	pt := &PartitionTable{
 		partition:      partition,
 		state:          newPartitionTableState(),
@@ -96,23 +93,21 @@ func newPartitionTable(topic string,
 		stallPeriod:    defaultStallPeriod,
 		stalledTimeout: defaultStalledTimeout,
 
-		stats:           newTableStats(),
-		requestStats:    make(chan bool),
-		responseStats:   make(chan *TableStats, 1),
-		updateStats:     make(chan func(), 10),
-		cancelStatsLoop: cancel,
+		stats:         newTableStats(),
+		requestStats:  make(chan bool),
+		responseStats: make(chan *TableStats, 1),
+		updateStats:   make(chan func(), 10),
 
 		backoff:             backoff,
 		backoffResetTimeout: backoffResetTimeout,
 	}
 
-	go pt.runStatsLoop(statsLoopCtx)
-
 	return pt
 }
 
 // SetupAndRecover sets up the partition storage and recovers to HWM
 func (p *PartitionTable) SetupAndRecover(ctx context.Context, restartOnError bool) error {
+
 	err := p.setup(ctx)
 	if err != nil {
 		return err
@@ -538,10 +533,17 @@ func (p *PartitionTable) enqueueStatsUpdate(ctx context.Context, updater func())
 	select {
 	case p.updateStats <- updater:
 	case <-ctx.Done():
+	default:
+		// going to default indicates the updateStats channel is not read, so so the stats
+		// loop is not actually running.
+		// We must not block here, so we'll skip the update
 	}
 }
 
-func (p *PartitionTable) runStatsLoop(ctx context.Context) {
+// RunStatsLoop starts the handler for stats requests. This loop runs detached from the
+// recover/catchup mechanism so clients can always request stats even if the partition table is not
+// running (like a processor table after it's recovered).
+func (p *PartitionTable) RunStatsLoop(ctx context.Context) {
 
 	updateHwmStatsTicker := time.NewTicker(statsHwmUpdateInterval)
 	defer updateHwmStatsTicker.Stop()

view.go

@@ -237,6 +237,7 @@ func (v *View) Run(ctx context.Context) (rerr error) {
 
 	for _, partition := range v.partitions {
 		partition := partition
+		go partition.RunStatsLoop(ctx)
 		recoverErrg.Go(func() error {
 			return partition.SetupAndRecover(recoverCtx, v.opts.autoreconnect)
 		})

view_test.go

@@ -697,7 +697,7 @@ func TestView_WaitRunning(t *testing.T) {
 		view, _, ctrl := createTestView(t, NewMockAutoConsumer(t, DefaultConfig()))
 		defer ctrl.Finish()
 
-		view.state = newViewSignal()
+		view.state = newViewSignal().SetState(State(ViewStateRunning))
 
 		var isRunning bool
 		select {