func (t *RetentionManager) Stop() {
	t.Lock()
	defer t.Unlock()
	if t.running {
		close(t.stopC)  
		t.wg.Wait()     
		t.running = false
		t.logger.Info("RetentionMgr stopped")
	}
}
Stop stops the retention manager
package retentionmgr
import (
	"fmt"
	"math/rand"
	"sync"
	"time"
	"github.com/uber-common/bark"
	"github.com/uber/cherami-server/common"
	"github.com/uber/cherami-server/common/metrics"
	"github.com/uber/cherami-thrift/.generated/go/metadata"
	"github.com/uber/cherami-thrift/.generated/go/shared"
	"github.com/uber/cherami-thrift/.generated/go/store"
)
const numWorkersDefault = 1                       
const bufferedJobs = 65536                        
const defaultExtentDeleteDeferPeriod = time.Hour  
type (
	Options struct {
		RetentionInterval                time.Duration
		DLQRetentionInterval             time.Duration
		SingleCGVisibleExtentGracePeriod time.Duration
		ExtentDeleteDeferPeriod          time.Duration
		NumWorkers                       int
		LocalZone                        string
	}
	RetentionManager struct {
		*Options
		logger   bark.Logger
		m3Client metrics.Client
		metadata  metadataDep
		storehost storehostDep
		wg sync.WaitGroup
		jobsC chan *retentionJob
		stopC chan struct{}
		sync.Mutex
		running bool
		lastDLQRetentionRun time.Time
	}
)
type (
	metadataDep interface {
		GetDestinations() (destinations []*destinationInfo, err error)
		GetExtents(destID destinationID) (extents []*extentInfo, err error)
		GetConsumerGroups(destID destinationID) (consumerGroups []*consumerGroupInfo, err error)
		DeleteExtent(destID destinationID, extID extentID) (err error)
		GetExtentsForConsumerGroup(dstID destinationID, cgID consumerGroupID) (extIDs []extentID, err error)
		MarkExtentConsumed(destID destinationID, extID extentID) (err error)
		DeleteConsumerGroupUUID(destID destinationID, cgID consumerGroupID) (err error)
		DeleteDestination(destID destinationID) (err error)
		DeleteConsumerGroupExtent(destID destinationID, cgID consumerGroupID, extID extentID) (err error)
		GetAckLevel(destID destinationID, extID extentID, cgID consumerGroupID) (ackLevel int64, err error)
		GetExtentInfo(destID destinationID, extID extentID) (extInfo *extentInfo, err error)
	}
	storehostDep interface {
		GetAddressFromTimestamp(storeID storehostID, extID extentID, timestamp int64) (addr int64, sealed bool, err error)
		PurgeMessages(storeID storehostID, extID extentID, retentionAddr int64) (doneAddr int64, err error)
	}
)
type (
	extentID        string
	consumerGroupID string
	destinationID   string
	storehostID     string
	destinationInfo struct {
		id            destinationID
		destType      shared.DestinationType
		status        shared.DestinationStatus
		extents       []*extentInfo
		softRetention int32  
		hardRetention int32  
		path          string
		isMultiZone   bool
	}
	extentInfo struct {
		id                 extentID
		status             shared.ExtentStatus
		statusUpdatedTime  time.Time
		storehosts         []storehostID
		singleCGVisibility consumerGroupID
		originZone         string
		kafkaPhantomExtent bool
	}
	consumerGroupInfo struct {
		id     consumerGroupID
		status shared.ConsumerGroupStatus
	}
	retentionJob struct {
		runAt         time.Time
		dest          *destinationInfo
		ext           *extentInfo
		consumers     []*consumerGroupInfo
		err           error
		minAckAddr    int64  
		retentionAddr int64  
		deleteExtent  bool   
	}
)
func New(opts *Options, metadata metadata.TChanMetadataService, clientFactory common.ClientFactory, m3Client metrics.Client, logger bark.Logger) *RetentionManager {
	if opts.NumWorkers <= 0 {
		opts.NumWorkers = numWorkersDefault
	}
	if opts.ExtentDeleteDeferPeriod <= 0 {
		opts.ExtentDeleteDeferPeriod = defaultExtentDeleteDeferPeriod
	}
	logger = logger.WithField(common.TagModule, `retMgr`)
	return &RetentionManager{
		Options:             opts,
		logger:              logger,
		m3Client:            m3Client,
		metadata:            newMetadataDep(metadata, logger),
		storehost:           newStorehostDep(clientFactory, logger),
		lastDLQRetentionRun: time.Now().AddDate(0, 0, -1),
	}
}
func tNew(opts *Options, metadata metadataDep, storehost storehostDep, m3Client metrics.Client, logger bark.Logger) *RetentionManager {
	if opts.NumWorkers <= 0 {
		opts.NumWorkers = numWorkersDefault
	}
	logger = logger.WithField(common.TagModule, `retMgr`)
	return &RetentionManager{
		Options:             opts,
		logger:              logger,
		m3Client:            m3Client,
		metadata:            metadata,
		storehost:           storehost,
		lastDLQRetentionRun: time.Now().AddDate(0, 0, -1),
	}
}
func (t *RetentionManager) Start() {
	t.Lock()
	defer t.Unlock()
	if !t.running {
		t.logger.WithFields(bark.Fields{
			`interval`:    t.RetentionInterval,
			`dlqinterval`: t.DLQRetentionInterval,
			`workers`:     t.NumWorkers,
		}).Info(`RetentionMgr starting`)
		t.jobsC = make(chan *retentionJob, bufferedJobs)  
		t.stopC = make(chan struct{})
		for i := 0; i < t.NumWorkers; i++ {
			t.wg.Add(1)
			go t.retentionWorker(i, t.jobsC)
		}
		if t.RetentionInterval > 0 {
			t.wg.Add(1)
			go func() {
				defer t.wg.Done()
				defer close(t.jobsC)  
				timer := time.NewTimer(time.Minute)
				defer timer.Stop()
				for {
					select {
					case <-timer.C:  
					case <-t.stopC:
						return
					}
					if !t.runRetention(t.jobsC) {
						return
					}
					timer.Reset(t.RetentionInterval)
				}
			}()
		}
		t.running = true
	}
}