// persist in-memory state to disk after every 10secs.
				batchLogIf(ctx, ri.updateAfter(ctx, api, 10*time.Second, job))

				if wait := globalBatchConfig.ReplicationWait(); wait > 0 {
					time.Sleep(wait)
				}
			}()
		}
		wk.Wait()

		ri.RetryAttempts = attempts
		ri.Complete = ri.ObjectsFailed == 0
		ri.Failed = ri.ObjectsFailed > 0

		globalBatchJobsMetrics.save(job.ID, ri)

	})
	if err != nil {
		t.Fatal(err)
	}

	select {
	case _, ok := <-w.ResultChan():
		if ok {
			t.Error("ResultChan() should be closed")
		}
	case <-time.After(wait.ForeverTestTimeout):
		t.Errorf("timeout after %v", wait.ForeverTestTimeout)
	}
}

func RunTestWatcherTimeout(ctx context.Context, t *testing.T, store storage.Interface) {
	// initialRV is used to initate the watcher at the beginning of the world.

	return pq
}

// Run starts the goroutine to pump from podBackoffQ to activeQ
func (p *PriorityQueue) Run(logger klog.Logger) {
	go wait.Until(func() {
		p.flushBackoffQCompleted(logger)
	}, 1.0*time.Second, p.stop)
	go wait.Until(func() {
		p.flushUnschedulablePodsLeftover(logger)
	}, 30*time.Second, p.stop)
}

			fakePlugin.GetNewDetacherCallCount())
	}
}

func retryWithExponentialBackOff(initialDuration time.Duration, fn wait.ConditionFunc) error {
	backoff := wait.Backoff{
		Duration: initialDuration,
		Factor:   3,
		Jitter:   0,
		Steps:    6,
	}
	return wait.ExponentialBackoff(backoff, fn)
}

// verifies the force detach metric with reason

    # after some delay, or it may appear to be removed then you'll add it back
    # but then it will be removed once again. So, we first wait a long
    # unfortunate amount of time to ensure that things have quiesced, then we
    # wait until we're sure the route is really gone before re-adding it again.
    Log-Output "Waiting 45 seconds for host network state to quiesce"
    Start-Sleep 45

  }

  /**
   * Marks the given step final, waits for it to be finished, and returns the value.
   *
   * @throws ExecutionException if the step failed
   * @throws CancellationException if the step was cancelled
   */
  abstract <T> T getFinalValue(ClosingFuture<T> closingFuture) throws ExecutionException;

  /** Marks the given step final, cancels it, and waits for the cancellation to happen. */

		// be re-queued until "Done", so no more than one worker handle the same item and
		// no event missed.
		go wait.UntilWithContext(ctx, nc.doNodeProcessingPassWorker, time.Second)
	}

	for i := 0; i < podUpdateWorkerSize; i++ {
		go wait.UntilWithContext(ctx, nc.doPodProcessingWorker, time.Second)
	}

		gracePeriod = gracePeriod - m.executePreStopHook(ctx, pod, containerID, containerSpec, gracePeriod)
	}

	// if we care about termination ordering, then wait for this container's turn to exit if there is
	// time remaining
	if ordering != nil && gracePeriod > 0 {
		// grace period is only in seconds, so the time we've waited gets truncated downward

		}
	}()

	// Note the deferred Wait must be called after the deferred close(done),
	// to ensure the N goroutines have been released even if the main goroutine
	// calls Fatalf. It must be called before the Chdir back to the original
	// directory, and before the deferred deletion implied by TempDir,
	// so as not to interfere while the N goroutines are still running.
	defer wg.Wait()
	defer close(done)

	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/apimachinery/pkg/util/uuid"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/tools/record"
	pluginapi "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1"
	watcherapi "k8s.io/kubelet/pkg/apis/pluginregistration/v1"
	"k8s.io/kubernetes/pkg/kubelet/checkpointmanager"