Halting and re-triggering in short succession causes premature state termination
When an actively running Execution is halted (through the cli or api) Styx will `delete` the Pod in k8s. This process takes some time to complete as the container have to receive the `SIGTERM` signal and exit. It will get some time to exit gracefully (the so called grace period).

If the same Workflow Instance is triggered and a new Execution is started during this shutdown period, there will be two Pods in k8s which are associated with the same Workflow Instance (through a Pod annotation). When finally the first Pod exits (usually with status code 137, 128 + `SIGTERM(9)`), a `terminate(137)` event is generated for that Workflow Instance. This event is dispatched to the currently active state and we see a premature state transition.

At this point, the second execution Pod is orphaned.

A sequence of events outlining this scenario:
```
2016-11-28T14:39:57       triggerExecution          Trigger id: ad-hoc-cli-1480343996990-18517
2016-11-28T14:39:57       submit                    Execution description: ExecutionDescription{dockerImage=stagger:3, dockerArgs=[{}], secret=Optional.empty, commitSha=Optional.empty}
2016-11-28T14:39:57       submitted                 Execution id: styx-run-0ab59615-979b-4717-80f3-deab984d1074
2016-11-28T14:40:00       started
2016-11-28T14:41:17       halt
2016-11-28T14:41:22       triggerExecution          Trigger id: ad-hoc-cli-1480344082837-41704
2016-11-28T14:41:23       submit                    Execution description: ExecutionDescription{dockerImage=stagger:3, dockerArgs=[{}], secret=Optional.empty, commitSha=Optional.empty}
2016-11-28T14:41:23       submitted                 Execution id: styx-run-12994ebb-c7f7-4083-8844-952ecd63b3f8
2016-11-28T14:41:27       started
2016-11-28T14:41:48       terminate                 Exit code: 137
2016-11-28T14:41:48       retryAfter                Delay (seconds): 180
```

The `terminate` event at the end is actually coming from the first execution, `styx-run-0ab59615-979b-4717-80f3-deab984d1074`.

diff --git a/styx-scheduler-service/src/main/java/com/spotify/styx/docker/KubernetesDockerRunner.java b/styx-scheduler-service/src/main/java/com/spotify/styx/docker/KubernetesDockerRunner.java
index e735b7f5..4515be8e 100644
--- a/styx-scheduler-service/src/main/java/com/spotify/styx/docker/KubernetesDockerRunner.java
+++ b/styx-scheduler-service/src/main/java/com/spotify/styx/docker/KubernetesDockerRunner.java
@@ -49,6 +49,7 @@ import java.net.ProtocolException;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -192,8 +193,9 @@ class KubernetesDockerRunner implements DockerRunner {
 
   private void inspectPod(Watcher.Action action, Pod pod) {
     final Map<String, String> annotations = pod.getMetadata().getAnnotations();
+    final String podName = pod.getMetadata().getName();
     if (!annotations.containsKey(KubernetesDockerRunner.STYX_WORKFLOW_INSTANCE_ANNOTATION)) {
-      LOG.warn("Got pod without workflow instance annotation {}", pod.getMetadata().getName());
+      LOG.warn("Got pod without workflow instance annotation {}", podName);
       return;
     }
 
@@ -206,6 +208,19 @@ class KubernetesDockerRunner implements DockerRunner {
       return;
     }
 
+    final Optional<String> executionIdOpt = runState.executionId();
+    if (!executionIdOpt.isPresent()) {
+      LOG.warn("Pod event for state with no current executionId: {}", podName);
+      return;
+    }
+
+    final String executionId = executionIdOpt.get();
+    if (!podName.equals(executionId)) {
+      LOG.warn("Pod event not matching current exec id, current:{} != pod:{}",
+          executionId, podName);
+      return;
+    }
+
     final List<Event> events = translate(workflowInstance, runState, action, pod);
 
     for (Event event : events) {
diff --git a/styx-scheduler-service/src/test/java/com/spotify/styx/docker/KubernetesDockerRunnerPodResourceTest.java b/styx-scheduler-service/src/test/java/com/spotify/styx/docker/KubernetesDockerRunnerPodResourceTest.java
index b20489ae..589866f2 100644
--- a/styx-scheduler-service/src/test/java/com/spotify/styx/docker/KubernetesDockerRunnerPodResourceTest.java
+++ b/styx-scheduler-service/src/test/java/com/spotify/styx/docker/KubernetesDockerRunnerPodResourceTest.java
@@ -20,10 +20,12 @@
 
 package com.spotify.styx.docker;
 
+import static com.jcabi.matchers.RegexMatchers.matchesPattern;
 import static com.spotify.styx.docker.KubernetesDockerRunner.STYX_WORKFLOW_INSTANCE_ANNOTATION;
 import static java.util.Optional.empty;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.hasEntry;
+import static org.hamcrest.Matchers.hasItem;
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.assertThat;
 
@@ -32,6 +34,7 @@ import com.spotify.styx.model.DataEndpoint;
 import com.spotify.styx.model.WorkflowInstance;
 import com.spotify.styx.testdata.TestData;
 import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.EnvVar;
 import io.fabric8.kubernetes.api.model.Pod;
 import io.fabric8.kubernetes.api.model.Volume;
 import io.fabric8.kubernetes.api.model.VolumeMount;
@@ -42,6 +45,9 @@ import org.junit.Test;
 
 public class KubernetesDockerRunnerPodResourceTest {
 
+  private static final String UUID_REGEX =
+      "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}";
+
   private static final WorkflowInstance WORKFLOW_INSTANCE =
       WorkflowInstance.create(TestData.WORKFLOW_ID, "2016-04-04");
 
@@ -155,4 +161,35 @@ public class KubernetesDockerRunnerPodResourceTest {
     assertThat(volumeMount.getMountPath(), is("/etc/secrets"));
     assertThat(volumeMount.getReadOnly(), is(true));
   }
+
+  @Test
+  public void shouldConfigureEnvironmentVariables() throws Exception {
+    Pod pod = KubernetesDockerRunner.createPod(
+        WORKFLOW_INSTANCE,
+        DockerRunner.RunSpec.create("busybox", ImmutableList.of(), Optional.empty()));
+    List<EnvVar> envVars = pod.getSpec().getContainers().get(0).getEnv();
+
+    EnvVar endpoint = new EnvVar();
+    endpoint.setName(KubernetesDockerRunner.ENDPOINT_ID);
+    endpoint.setValue(WORKFLOW_INSTANCE.workflowId().endpointId());
+    EnvVar workflow = new EnvVar();
+    workflow.setName(KubernetesDockerRunner.WORKFLOW_ID);
+    workflow.setValue(WORKFLOW_INSTANCE.workflowId().endpointId());
+    EnvVar component = new EnvVar();
+    component.setName(KubernetesDockerRunner.COMPONENT_ID);
+    component.setValue(WORKFLOW_INSTANCE.workflowId().componentId());
+    EnvVar parameter = new EnvVar();
+    parameter.setName(KubernetesDockerRunner.PARAMETER);
+    parameter.setValue(WORKFLOW_INSTANCE.parameter());
+    EnvVar execution = envVars.get(4);
+
+    assertThat(envVars.size(), is(5));
+    assertThat(envVars, hasItem(component));
+    assertThat(envVars, hasItem(workflow));
+    assertThat(envVars, hasItem(endpoint));
+    assertThat(envVars, hasItem(parameter));
+    assertThat(execution.getName(),is(KubernetesDockerRunner.EXECUTION_ID));
+    assertThat(execution.getValue(),
+        matchesPattern(KubernetesDockerRunner.STYX_RUN + "-" + UUID_REGEX));
+  }
 }
diff --git a/styx-scheduler-service/src/test/java/com/spotify/styx/docker/KubernetesDockerRunnerTest.java b/styx-scheduler-service/src/test/java/com/spotify/styx/docker/KubernetesDockerRunnerTest.java
index 24e10b19..309bd4ec 100644
--- a/styx-scheduler-service/src/test/java/com/spotify/styx/docker/KubernetesDockerRunnerTest.java
+++ b/styx-scheduler-service/src/test/java/com/spotify/styx/docker/KubernetesDockerRunnerTest.java
@@ -20,12 +20,10 @@
 
 package com.spotify.styx.docker;
 
-import static com.jcabi.matchers.RegexMatchers.matchesPattern;
 import static com.spotify.styx.docker.KubernetesPodEventTranslatorTest.podStatusNoContainer;
 import static com.spotify.styx.docker.KubernetesPodEventTranslatorTest.running;
 import static com.spotify.styx.docker.KubernetesPodEventTranslatorTest.terminated;
 import static com.spotify.styx.docker.KubernetesPodEventTranslatorTest.waiting;
-import static org.hamcrest.Matchers.hasItem;
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.assertThat;
 import static org.mockito.Matchers.any;
@@ -44,7 +42,6 @@ import com.spotify.styx.state.StateManager;
 import com.spotify.styx.state.SyncStateManager;
 import com.spotify.styx.testdata.TestData;
 import io.fabric8.kubernetes.api.model.DoneablePod;
-import io.fabric8.kubernetes.api.model.EnvVar;
 import io.fabric8.kubernetes.api.model.ListMeta;
 import io.fabric8.kubernetes.api.model.Pod;
 import io.fabric8.kubernetes.api.model.PodList;
@@ -55,7 +52,6 @@ import io.fabric8.kubernetes.client.dsl.ClientMixedOperation;
 import io.fabric8.kubernetes.client.dsl.ClientPodResource;
 import io.fabric8.kubernetes.client.dsl.Watchable;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import org.junit.After;
@@ -73,8 +69,6 @@ import org.mockito.runners.MockitoJUnitRunner;
 @RunWith(MockitoJUnitRunner.class)
 public class KubernetesDockerRunnerTest {
 
-  private static final String UUID_REGEX =
-      "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}";
   private static final String POD_NAME = "test-pod-1";
   private static final WorkflowInstance WORKFLOW_INSTANCE = WorkflowInstance.create(TestData.WORKFLOW_ID, "foo");
   private static final RunSpec RUN_SPEC = RunSpec.create("busybox", ImmutableList.of(), Optional.empty());
@@ -123,7 +117,7 @@ public class KubernetesDockerRunnerTest {
     createdPod.getMetadata().setName(POD_NAME);
     createdPod.getMetadata().setResourceVersion("1001");
 
-    stateManager.initialize(RunState.create(WORKFLOW_INSTANCE, RunState.State.SUBMITTED));
+    stateManager.initialize(RunState.create(WORKFLOW_INSTANCE, POD_NAME, RunState.State.SUBMITTED));
 
     when(pods.create(any(Pod.class))).thenReturn(createdPod);
 
@@ -136,33 +130,6 @@ public class KubernetesDockerRunnerTest {
     kdr.close();
   }
 
-  @Test
-  public void shouldReturnEnvironmentVariablesForCreatedPod() throws Exception {
-    List<EnvVar> envVars = createdPod.getSpec().getContainers().get(0).getEnv();
-    EnvVar endpoint = new EnvVar();
-    endpoint.setName(KubernetesDockerRunner.ENDPOINT_ID);
-    endpoint.setValue(WORKFLOW_INSTANCE.workflowId().endpointId());
-    EnvVar workflow = new EnvVar();
-    workflow.setName(KubernetesDockerRunner.WORKFLOW_ID);
-    workflow.setValue(WORKFLOW_INSTANCE.workflowId().endpointId());
-    EnvVar component = new EnvVar();
-    component.setName(KubernetesDockerRunner.COMPONENT_ID);
-    component.setValue(WORKFLOW_INSTANCE.workflowId().componentId());
-    EnvVar parameter = new EnvVar();
-    parameter.setName(KubernetesDockerRunner.PARAMETER);
-    parameter.setValue(WORKFLOW_INSTANCE.parameter());
-    EnvVar execution = envVars.get(4);
-
-    assertThat(envVars.size(), is(5));
-    assertThat(envVars, hasItem(component));
-    assertThat(envVars, hasItem(workflow));
-    assertThat(envVars, hasItem(endpoint));
-    assertThat(envVars, hasItem(parameter));
-    assertThat(execution.getName(),is(KubernetesDockerRunner.EXECUTION_ID));
-    assertThat(execution.getValue(),
-        matchesPattern(KubernetesDockerRunner.STYX_RUN + "-" + UUID_REGEX));
-  }
-
   @Test
   public void shouldCompleteWithStatusCodeOnSucceeded() throws Exception {
     createdPod.setStatus(terminated("Succeeded", 20));
@@ -230,7 +197,7 @@ public class KubernetesDockerRunnerTest {
 
   @Test
   public void shouldGenerateStartedWhenContainerIsReady() throws Exception {
-    stateManager.initialize(RunState.create(WORKFLOW_INSTANCE, RunState.State.SUBMITTED));
+    stateManager.initialize(RunState.create(WORKFLOW_INSTANCE, POD_NAME, RunState.State.SUBMITTED));
     createdPod.setStatus(running(/* ready= */ false));
     podWatcher.eventReceived(Watcher.Action.MODIFIED, createdPod);
     assertThat(stateManager.get(WORKFLOW_INSTANCE).state(), is(RunState.State.SUBMITTED));
@@ -239,4 +206,15 @@ public class KubernetesDockerRunnerTest {
     podWatcher.eventReceived(Watcher.Action.MODIFIED, createdPod);
     assertThat(stateManager.get(WORKFLOW_INSTANCE).state(), is(RunState.State.RUNNING));
   }
+
+  @Test
+  public void shouldDiscardChangesForOldExecutions() throws Exception {
+    // simulate event from different pod, but still with the same workflow instance annotation
+    createdPod.getMetadata().setName(POD_NAME + "-other");
+    createdPod.setStatus(terminated("Succeeded", 20));
+
+    podWatcher.eventReceived(Watcher.Action.MODIFIED, createdPod);
+
+    assertThat(stateManager.get(WORKFLOW_INSTANCE).state(), is(RunState.State.RUNNING));
+  }
 }
['styx-scheduler-service/src/test/java/com/spotify/styx/docker/KubernetesDockerRunnerTest.java', 'styx-scheduler-service/src/test/java/com/spotify/styx/docker/KubernetesDockerRunnerPodResourceTest.java', 'styx-scheduler-service/src/main/java/com/spotify/styx/docker/KubernetesDockerRunner.java']