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)

	wk, err := workers.New(workerSize)
	if err != nil {
		return err
	}

	for setIdx, set := range pool.sets {

	gopts.NoLock = true // We already have a lock, we can live with it.
	objInfo, err := getObjectInfo(ctx, bucket, object, gopts)
	if err != nil {
		// Versioning enabled quite possibly object is deleted might be delete-marker
		// if present set the headers, no idea why AWS S3 sets these headers.
		if objInfo.VersionID != "" && objInfo.DeleteMarker {
			w.Header()[xhttp.AmzVersionID] = []string{objInfo.VersionID}

		Help:      "Total number of versions (includes delete marker) in a cluster",
		Type:      gaugeMetric,
	}
}

func getClusterUsageDeleteMarkersTotalMD() MetricDescription {
	return MetricDescription{
		Namespace: clusterMetricNamespace,
		Subsystem: usageSubsystem,
		Name:      deleteMarkerTotal,
		Help:      "Total number of delete markers in a cluster",
		Type:      gaugeMetric,
	}
}

      and issue tracking systems that are managed by, or on behalf of, the
      Licensor for the purpose of discussing and improving the Work, but
      excluding communication that is conspicuously marked or otherwise
      designated in writing by the copyright owner as "Not a Contribution."

      "Contributor" shall mean Licensor and any individual or Legal Entity

		}
		if result.Prefixes[0] != "this/" {
			t.Errorf("%s: Expected the prefix to be `%s`, but instead found `%s`", instanceType, "this/", result.Prefixes[0])
		}
	}

	// check results with Marker.
	{

		result, err = obj.ListObjects(context.Background(), "bucket", "", "newPrefix", "", 3)
		if err != nil {
			t.Fatalf("%s: <ERROR> %s", instanceType, err)
		}
		if result.Objects[0].Name != "newPrefix2" {

		HTTPStatusCode: http.StatusBadRequest,
	},
	ErrInvalidPrefixMarker: {
		Code:           "InvalidPrefixMarker",
		Description:    "Invalid marker prefix combination",
		HTTPStatusCode: http.StatusBadRequest,
	},
	ErrBadRequest: {
		Code:           "BadRequest",
		Description:    "400 BadRequest",
		HTTPStatusCode: http.StatusBadRequest,
	},

mkdir -p /tmp/data
for i in $(seq 1 10); do
	echo "T" >/tmp/data/file_${i}.txt
done

./mc mirror /tmp/data sitea/bucket/
./mc version enable sitea/bucket

./mc cp /tmp/data/file_1.txt sitea/bucket/marker
./mc rm sitea/bucket/marker

./mc mb siteb/bucket/
./mc version enable siteb/bucket/

echo "adding replication rule for site a -> site b"
./mc replicate add sitea/bucket/ \

				t.Log("ListObjectVersions, bucket:", testCase.bucketName, "prefix:",
					testCase.prefix, "marker:", testCase.marker, "delimiter:",
					testCase.delimiter, "maxkeys:", testCase.maxKeys)

				resultV, err = obj.ListObjectVersions(context.Background(), testCase.bucketName,
					testCase.prefix, testCase.marker, "", testCase.delimiter, testCase.maxKeys)
			} else {

					continue
				}

				// any object with only single DEL marker we don't need
				// to decommission, just skip it, this also includes
				// any other versions that have already expired.
				remainingVersions := len(fivs.Versions) - expired
				if version.Deleted && remainingVersions == 1 {
					decommissioned++
					stopFn(errors.New("DELETE marked object with no other non-current versions will be skipped"))
					continue

}

// GetDataDir returns an expected dataDir given FileInfo
//   - deleteMarker returns "delete-marker"
//   - returns "legacy" if FileInfo is XLV1 and DataDir is
//     empty, returns DataDir otherwise
//   - returns "dataDir"
func (fi FileInfo) GetDataDir() string {
	if fi.Deleted {
		return "delete-marker"
	}
	if fi.XLV1 && fi.DataDir == "" {
		return "legacy"
	}
	return fi.DataDir
}