func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job BatchJobRequest) error {
	ri := &batchJobInfo{
		JobID:     job.ID,
		JobType:   string(job.Type()),
		StartTime: job.Started,
	}
	if err := ri.load(ctx, api, job); err != nil {
		return err
	}
	if ri.Complete {
		return nil
	}

	globalBatchJobsMetrics.save(job.ID, ri)
	lastObject := ri.Object

	delay := job.KeyRotate.Flags.Retry.Delay

./mc admin replicate add minio1 minio2 --insecure
# sleep for replication to complete
sleep 30

# Create bucket in source cluster
echo "Create bucket in source MinIO instance"
./mc mb minio1/test-bucket --insecure

# Load objects to source site
echo "Loading objects to source MinIO instance"
set -x
./mc cp /tmp/data/plainfile minio1/test-bucket --insecure

func (r *BatchJobExpire) Start(ctx context.Context, api ObjectLayer, job BatchJobRequest) error {
	ri := &batchJobInfo{
		JobID:     job.ID,
		JobType:   string(job.Type()),
		StartTime: job.Started,
	}
	if err := ri.load(ctx, api, job); err != nil {
		return err
	}

	globalBatchJobsMetrics.save(job.ID, ri)
	lastObject := ri.Object

	now := time.Now().UTC()

			return errFileVersionNotFound
		}
		return err
	}
	defer metaDataPoolPut(buf)

	if !isXL2V1Format(buf) {
		return errFileVersionNotFound
	}

	var xlMeta xlMetaV2
	if err = xlMeta.Load(buf); err != nil {
		return err
	}

	if err = xlMeta.UpdateObjectVersion(fi); err != nil {
		return err
	}

	wbuf, err := xlMeta.AppendTo(metaDataPoolGet())
	if err != nil {
		return err

func (p *ReplicationPool) initResync(ctx context.Context, buckets []BucketInfo, objAPI ObjectLayer) error {
	if objAPI == nil {
		return errServerNotInitialized
	}
	// Load bucket metadata sys in background
	go p.startResyncRoutine(ctx, buckets, objAPI)
	return nil
}

func (p *ReplicationPool) startResyncRoutine(ctx context.Context, buckets []BucketInfo, objAPI ObjectLayer) {

		_, ok1 := err.(BucketExists)
		_, ok2 := err.(BucketAlreadyExists)
		if !ok1 && !ok2 {
			return wrapSRErr(c.annotateErr(makeBucketWithVersion, err))
		}
	} else {
		// Load updated bucket metadata into memory as new
		// bucket was created.
		globalNotificationSys.LoadBucketMetadata(GlobalContext, bucket)
	}

	meta, err := globalBucketMetadataSys.Get(bucket)
	if err != nil {

	logger.FatalIf(lockRPCForceUnlock.Register(globalGrid.Load(), lockServer.ForceUnlockHandler), "unable to register handler")
	logger.FatalIf(lockRPCRefresh.Register(globalGrid.Load(), lockServer.RefreshHandler), "unable to register handler")
	logger.FatalIf(lockRPCLock.Register(globalGrid.Load(), lockServer.LockHandler), "unable to register handler")
	logger.FatalIf(lockRPCUnlock.Register(globalGrid.Load(), lockServer.UnlockHandler), "unable to register handler")

	for _, item := range iamListing[stsListKey] {
		userName := path.Dir(item)
		// loadUser() will delete expired user during the load.
		err := iamOS.loadUser(ctx, userName, stsUser, stsAccountsFromStore)
		if err != nil && !errors.Is(err, errNoSuchUser) {
			iamLogIf(GlobalContext,
				fmt.Errorf("unable to load user during STS purge: %w (%s)", err, item))
		}

	}
	// Loading the STS policy mappings from disk ensures that stale entries

	registerStorageRESTHandlers(router, endpointServerPools, globalGrid.Load())

	// Register peer REST router only if its a distributed setup.
	registerPeerRESTHandlers(router, globalGrid.Load())

	// Register bootstrap REST router for distributed setups.
	registerBootstrapRESTHandlers(globalGrid.Load())

	// Register distributed namespace lock routers.
	registerLockRESTHandlers()

		}

		httpStats := globalBucketHTTPStats.load(bucket)
		for k, v := range httpStats.currentS3Requests.Load(false) {
			m.Set(apiRequestsInFlightTotal, float64(v), "bucket", bucket, "name", k, "type", "s3")
		}

		for k, v := range httpStats.totalS3Requests.Load(false) {
			m.Set(apiRequestsTotal, float64(v), "bucket", bucket, "name", k, "type", "s3")
		}