feat: Add WaitWithTimeout to Partition and WaitGroupTimeout

pkg/wg_timeout/wg_timeout.go

@@ -0,0 +1,25 @@
+package wg_timeout
+
+import (
+	"sync"
+	"time"
+)
+
+func WaitGroupTimeout(wg *sync.WaitGroup, timeout time.Duration, emitter func()) {
+	breakoutChan := make(chan struct{})
+
+	go func(c chan struct{}) {
+		defer close(c)
+		wg.Wait()
+	}(breakoutChan)
+
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+
+	select {
+	case <-breakoutChan:
+		return
+	case <-timer.C:
+		emitter()
+	}
+}

tsdb/index/tsi1/partition.go

@@ -357,9 +357,16 @@ func (p *Partition) CurrentCompactionN() int {
 // Wait will block until all compactions are finished.
 // Must only be called while they are disabled.
 func (p *Partition) Wait() {
+	start := time.Now()
+	duration := 24 * time.Hour
 	ticker := time.NewTicker(10 * time.Millisecond)
 	defer ticker.Stop()
 	for {
+		if time.Since(start) > duration {
+			p.logger.Debug("timed out waiting for compaction to finish", zap.Duration("duration", duration), zap.String("path", p.path), zap.String("id", p.id))
+			// Reset start time to Now() so it will capture in the next 24 hours given this is still hanging
+			start = time.Now()
+		}
 		if p.CurrentCompactionN() == 0 {
 			return
 		}

tsdb/shard.go

@@ -28,6 +28,7 @@ import (
 	"github.com/influxdata/influxdb/pkg/file"
 	"github.com/influxdata/influxdb/pkg/limiter"
 	"github.com/influxdata/influxdb/pkg/slices"
+	"github.com/influxdata/influxdb/pkg/wg_timeout"
 	"github.com/influxdata/influxdb/query"
 	internal "github.com/influxdata/influxdb/tsdb/internal"
 	"github.com/influxdata/influxql"
@@ -1814,7 +1815,9 @@ func (fs *MeasurementFieldSet) SetMeasurementFieldSetWriter(queueLength int, log
 func (fscm *measurementFieldSetChangeMgr) Close() {
 	if fscm != nil {
 		close(fscm.writeRequests)
-		fscm.wg.Wait()
+		wg_timeout.WaitGroupTimeout(&fscm.wg, 24*time.Hour, func() {
+			fscm.logger.Debug("timed out waiting for measurementFieldSetChangeMgr to close", zap.String("changeFilePath", fscm.changeFilePath))
+		})
 	}
 }