}

// isOffline returns current liveness result of remote target. Add endpoint to
// healthCheck map if missing and default to online status
func (sys *BucketTargetSys) isOffline(ep *url.URL) bool {
	sys.hMutex.RLock()
	defer sys.hMutex.RUnlock()
	if h, ok := sys.hc[ep.Host]; ok {
		return !h.Online
	}
	go sys.initHC(ep)
	return false
}

// markOffline sets endpoint to offline if network i/o timeout seen.

}

func (z *erasureServerPools) IsRebalanceStarted() bool {
	z.rebalMu.RLock()
	defer z.rebalMu.RUnlock()

	if r := z.rebalMeta; r != nil {
		if r.StoppedAt.IsZero() {
			return true
		}
	}
	return false
}

func (z *erasureServerPools) IsPoolRebalancing(poolIndex int) bool {
	z.rebalMu.RLock()
	defer z.rebalMu.RUnlock()

	if r := z.rebalMeta; r != nil {

	return metrics
}

func (config *TierConfigMgr) refreshedAt() time.Time {
	config.RLock()
	defer config.RUnlock()
	return config.lastRefreshedAt
}

// IsTierValid returns true if there exists a remote tier by name tierName,
// otherwise returns false.
func (config *TierConfigMgr) IsTierValid(tierName string) bool {
	config.RLock()
	defer config.RUnlock()
	_, valid := config.isTierNameInUse(tierName)
	return valid

	return "kafka"
}

// String - kafka string
func (h *Target) String() string {
	return "kafka"
}

// Stats returns the target statistics.
func (h *Target) Stats() types.TargetStats {
	h.logChMu.RLock()
	queueLength := len(h.logCh)
	h.logChMu.RUnlock()

	return types.TargetStats{
		TotalMessages:  atomic.LoadInt64(&h.totalMessages),
		FailedMessages: atomic.LoadInt64(&h.failedMessages),

}

func (z *erasureServerPools) decommissionInBackground(ctx context.Context, idx int) error {
	pool := z.serverPools[idx]
	z.poolMetaMutex.RLock()
	pending := z.poolMeta.PendingBuckets(idx)
	z.poolMetaMutex.RUnlock()

	for _, bucket := range pending {
		z.poolMetaMutex.RLock()
		isDecommissioned := z.poolMeta.isBucketDecommissioned(idx, bucket.String())
		z.poolMetaMutex.RUnlock()
		if isDecommissioned {

func (c *Client) LastConn() time.Time {
	return time.Unix(0, atomic.LoadInt64(&c.lastConn))
}

// LastError returns previous error
func (c *Client) LastError() error {
	c.RLock()
	defer c.RUnlock()
	return fmt.Errorf("[%s] %w", c.lastErrTime.Format(time.RFC3339), c.lastErr)
}

// computes the exponential backoff duration according to

func (h *healSequence) getScannedItemsCount() int64 {
	var count int64
	h.mutex.RLock()
	defer h.mutex.RUnlock()

	for _, v := range h.scannedItemsMap {
		count += v
	}
	return count
}

// getScannedItemsMap - returns map of all scanned items against type
func (h *healSequence) getScannedItemsMap() map[madmin.HealItemType]int64 {
	h.mutex.RLock()
	defer h.mutex.RUnlock()

	// Make a copy before returning the value

func (r *ReplicationStats) updateMovingAvg() {
	r.RLock()
	for _, s := range r.Cache {
		for _, st := range s.Stats {
			st.XferRateLrg.measure.updateExponentialMovingAverage(time.Now())
			st.XferRateSml.measure.updateExponentialMovingAverage(time.Now())
		}
	}
	r.RUnlock()
}

// ActiveWorkers returns worker stats
func (r *ReplicationStats) ActiveWorkers() ActiveWorkerStat {
	r.wlock.RLock()
	defer r.wlock.RUnlock()

//
//	is returned, the caller is expected to choose the right parity
//	at that point.
func (sCfg *Config) GetParityForSC(sc string) (parity int) {
	ConfigLock.RLock()
	defer ConfigLock.RUnlock()
	switch strings.TrimSpace(sc) {
	case RRS:
		if !sCfg.initialized {
			return -1
		}
		return sCfg.RRS.Parity
	default:
		if !sCfg.initialized {
			return -1
		}

	if _, ok := t.lastDayStats[tier]; !ok {
		t.lastDayStats[tier] = &lastDayTierStats{}
	}
	t.lastDayStats[tier].addStats(ts)
}

func (t *transitionState) getDailyAllTierStats() DailyAllTierStats {
	t.lastDayMu.RLock()
	defer t.lastDayMu.RUnlock()

	res := make(DailyAllTierStats, len(t.lastDayStats))
	for tier, st := range t.lastDayStats {
		res[tier] = st.clone()
	}
	return res
}