time.Sleep(wait)
			}
		}()
	}
	wk.Wait()

	ri.Complete = !failed && ri.ObjectsFailed == 0
	ri.Failed = failed || ri.ObjectsFailed > 0
	globalBatchJobsMetrics.save(job.ID, ri)
	// persist in-memory state to disk.
	batchLogIf(ctx, ri.updateAfter(ctx, api, 0, job))

	if err := r.Notify(ctx, ri); err != nil {
		batchLogIf(ctx, fmt.Errorf("unable to notify %v", err))
	}

	cancel()

}

// ToMRFEntry returns the relevant info needed by MRF
func (ri ReplicateObjectInfo) ToMRFEntry() MRFReplicateEntry {
	return MRFReplicateEntry{
		Bucket:     ri.Bucket,
		Object:     ri.Name,
		versionID:  ri.VersionID,
		sz:         ri.Size,
		RetryCount: int(ri.RetryCount),
	}

		Version:             ri.Version,
		JobID:               ri.JobID,
		JobType:             ri.JobType,
		RetryAttempts:       ri.RetryAttempts,
		Complete:            ri.Complete,
		Failed:              ri.Failed,
		StartTime:           ri.StartTime,
		LastUpdate:          ri.LastUpdate,
		Bucket:              ri.Bucket,
		Object:              ri.Object,
		Objects:             ri.Objects,
		ObjectsFailed:       ri.ObjectsFailed,

	xioutil.SafeClose(expireCh)

	<-expireDoneCh // waits for the expire goroutine to complete
	wk.Wait()      // waits for all expire workers to retire

	ri.Complete = !failed && ri.ObjectsFailed == 0
	ri.Failed = failed || ri.ObjectsFailed > 0
	globalBatchJobsMetrics.save(job.ID, ri)

	// Close the saverQuitCh - this also triggers saving in-memory state
	// immediately one last time before we exit this method.

			}
		}
		return
	}

	var ch, healCh chan<- ReplicationWorkerOperation
	switch ri.OpType {
	case replication.HealReplicationType, replication.ExistingObjectReplicationType:
		ch = p.mrfReplicaCh
		healCh = p.getWorkerCh(ri.Name, ri.Bucket, ri.Size)
	default:
		ch = p.getWorkerCh(ri.Name, ri.Bucket, ri.Size)
	}
	if ch == nil && healCh == nil {
		return
	}

	select {

// function is merely and optimization.
func (ri ReplicateObjectInfo) ToObjectInfo() ObjectInfo {
	return ObjectInfo{
		Name:                       ri.Name,
		Bucket:                     ri.Bucket,
		VersionID:                  ri.VersionID,
		ModTime:                    ri.ModTime,
		UserTags:                   ri.UserTags,
		Size:                       ri.Size,
		ActualSize:                 &ri.ActualSize,

		}

		if loi.NextMarker != "" {
			loi.NextMarker = opts.encodeMarker(loi.NextMarker)
		}
		return loi, nil
	}
	ri := logger.GetReqInfo(ctx)
	hadoop := ri != nil && strings.Contains(ri.UserAgent, "Hadoop ") && strings.Contains(ri.UserAgent, "scala/")
	matches := func() bool {
		if prefix == "" {
			return false
		}
		// List of standard files supported by s3a

		} else {
			errStr = "<nil>"
		}
		tags[fmt.Sprintf("ddisk-%d", index)] = errStr
	}

	return m, nil
}

func fileInfoFromRaw(ri RawFileInfo, bucket, object string, readData, inclFreeVers bool) (FileInfo, error) {
	return getFileInfo(ri.Buf, bucket, object, "", fileInfoOpts{
		Data:             readData,
		InclFreeVersions: inclFreeVers,
	})
}