if err != nil {
		return nil, err
	}

	return &sarama.ProducerMessage{
		Topic: target.args.Topic,
		Key:   sarama.StringEncoder(key),
		Value: sarama.ByteEncoder(data),
	}, nil
}

// Close - closes underneath kafka connection.
func (target *KafkaTarget) Close() error {
	close(target.quitCh)

	if target.producer != nil {

	return err == nil
}

// IsDCOS returns true if minio is running in DCOS.
func IsDCOS() bool {
	// http://mesos.apache.org/documentation/latest/docker-containerizer/
	// Mesos docker containerizer sets this value
	return env.Get("MESOS_CONTAINER_NAME", "") != ""
}

// IsKubernetes returns true if minio is running in kubernetes.
func IsKubernetes() bool {
	// Kubernetes env used to validate if we are

				setIndex:  er.setIndex,
				poolIndex: er.poolIndex,
			})
		}

		return
	}()

	validResp := 0
	totalResp := 0

	// minDisks value is only to reduce the number of calls
	// to the disks; this value is not accurate because we do
	// not know the storage class of the object yet
	minDisks := 0
	if p := globalStorageClass.GetParityForSC(""); p > -1 {
		minDisks = er.setDriveCount - p

	Read(dst sql.Record) (sql.Record, error)
	Close() error
}

const (
	csvFormat     = "csv"
	jsonFormat    = "json"
	parquetFormat = "parquet"
)

// CompressionType - represents value inside <CompressionType/> in request XML.
type CompressionType string

const (
	noneType  CompressionType = "none"
	gzipType  CompressionType = "GZIP"
	bzip2Type CompressionType = "BZIP2"

	const envRebalanceWorkers = "_MINIO_REBALANCE_WORKERS"
	workerSize, err := env.GetInt(envRebalanceWorkers, len(pool.sets))
	if err != nil {
		rebalanceLogIf(ctx, fmt.Errorf("invalid workers value err: %v, defaulting to %d", err, len(pool.sets)))
		workerSize = len(pool.sets)
	}

	// Each decom worker needs one List() goroutine/worker
	// add that many extra workers.
	workerSize += len(pool.sets)

	return defaultMeta, nil
}

// getRemoteInstanceTransport contains a roundtripper for external (not peers) servers
var remoteInstanceTransport atomic.Value

func setRemoteInstanceTransport(tr http.RoundTripper) {
	remoteInstanceTransport.Store(tr)
}

func getRemoteInstanceTransport() http.RoundTripper {

	if rangeHeader := r.Header.Get(xhttp.AmzCopySourceRange); rangeHeader != "" {
		rs, parseRangeErr = parseCopyPartRangeSpec(rangeHeader)
	} else {
		// This check is to see if client specified a header but the value
		// is empty for 'x-amz-copy-source-range'
		_, ok := r.Header[xhttp.AmzCopySourceRange]
		if ok {
			parseRangeErr = errInvalidRange
		}
	}

	checkCopyPartPrecondFn := func(o ObjectInfo) bool {

			}
		}
		var maxUsedPct int
		for _, disk := range zinfo {
			if disk == nil || disk.Total == 0 {
				continue
			}
			available += disk.Total - disk.Used

			// set maxUsedPct to the value from the disk with the least space percentage.
			if pctUsed := int(disk.Used * 100 / disk.Total); pctUsed > maxUsedPct {
				maxUsedPct = pctUsed
			}
		}

          "thresholds": {
            "mode": "absolute",
            "steps": [
              {
                "color": "green",
                "value": null
              },
              {
                "color": "red",
                "value": 80
              }
            ]
          },
          "unit": "bytes"
        },
        "overrides": [
          {
            "matcher": {

	// but will fetch a new value in the background.
	NoWait bool
}

// Cache contains a synchronized value that is considered valid
// for a specific amount of time.
// An Update function must be set to provide an updated value when needed.
type Cache[T any] struct {
	// updateFn must return an updated value.
	// If an error is returned the cached value is not set.
	// Only one caller will call this function at any time, others will be blocking.