backport of commit bf11e39 (#22417)

Co-authored-by: Piotr Kazmierczak <470696+pkazmierczak@users.noreply.github.com>

drivers/docker/driver_test.go

@@ -378,6 +378,60 @@ func TestDockerDriver_Start_StoppedContainer(t *testing.T) {
 	require.NoError(t, d.DestroyTask(task.ID, true))
 }
 
+// TestDockerDriver_ContainerAlreadyExists asserts that when Nomad tries to
+// start a job and the container already exists, it purges it (if it's not in
+// the running state), and starts it again (as opposed to trying to
+// continuously re-create an already existing container)
+func TestDockerDriver_ContainerAlreadyExists(t *testing.T) {
+	ci.Parallel(t)
+	testutil.DockerCompatible(t)
+
+	task, cfg, _ := dockerTask(t)
+	must.NoError(t, task.EncodeConcreteDriverConfig(cfg))
+
+	client := newTestDockerClient(t)
+	driver := dockerDriverHarness(t, nil)
+	cleanup := driver.MkAllocDir(task, true)
+	defer cleanup()
+	copyImage(t, task.TaskDir(), "busybox.tar")
+
+	d, ok := driver.Impl().(*Driver)
+	must.True(t, ok)
+
+	_, err := d.createImage(task, cfg, client)
+	must.NoError(t, err)
+
+	containerCfg, err := d.createContainerConfig(task, cfg, cfg.Image)
+	must.NoError(t, err)
+
+	// create a container
+	c, err := d.createContainer(client, containerCfg, cfg.Image)
+	must.NoError(t, err)
+	defer client.RemoveContainer(docker.RemoveContainerOptions{
+		ID:    c.ID,
+		Force: true,
+	})
+
+	// now that the container has been created, start the task that uses it, and
+	// assert that it doesn't end up in "container already exists" fail loop
+	_, _, err = d.StartTask(task)
+	must.NoError(t, err)
+	d.DestroyTask(task.ID, true)
+
+	// let's try all of the above again, but this time with a created and running
+	// container
+	c, err = d.createContainer(client, containerCfg, cfg.Image)
+	must.NoError(t, err)
+	defer client.RemoveContainer(docker.RemoveContainerOptions{
+		ID:    c.ID,
+		Force: true,
+	})
+	must.NoError(t, d.startContainer(c))
+	_, _, err = d.StartTask(task)
+	must.NoError(t, err)
+	d.DestroyTask(task.ID, true)
+}
+
 func TestDockerDriver_Start_LoadImage(t *testing.T) {
 	ci.Parallel(t)
 	testutil.DockerCompatible(t)
@@ -2760,8 +2814,8 @@ func waitForExist(t *testing.T, client *docker.Client, containerID string) {
 }
 
 // TestDockerDriver_CreationIdempotent asserts that createContainer and
-// and startContainers functions are idempotent, as we have some retry
-// logic there without ensureing we delete/destroy containers
+// startContainers functions are idempotent, as we have some retry logic there
+// without ensuring we delete/destroy containers
 func TestDockerDriver_CreationIdempotent(t *testing.T) {
 	ci.Parallel(t)
 	testutil.DockerCompatible(t)