We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Create owner references among master and worker pods
Currently we have the following two ways to delete master + worker pods:
* Delete each pod one by one
* Delete all pods related to this elasticdl run via `elasticdl_job_name` label `kubectl delete pod -l elasticdl_job_name=test-job-1559292773-93`

It would be much easier if users could just delete master pod and then worker pods can be delete automatically. This would be possible if there are owner references among master and worker pods.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `elasticdl/python/elasticdl/master/k8s_client.py`
Content:
```
1 import logging
2 import os
3 import threading
4 import traceback
5 
6 from kubernetes import client, config, watch
7 from kubernetes.client import (
8     V1PersistentVolumeClaimVolumeSource as pvcVolumeSource,
9 )
10 
11 WORKER_POD_NAME_PREFIX = "elasticdl-worker-"
12 
13 
14 class Client(object):
15     def __init__(self, *, worker_image, namespace, job_name, event_callback):
16         """
17         ElasticDL k8s client.
18 
19         Args:
20             worker_image: Docker image path for ElasticDL workers.
21             namespace: k8s namespace for ElasticDL pods.
22             job_name: ElasticDL job name, should be unique in the namespace.
23                 Used as worker pod name prefix and value for "elasticdl" label.
24             event_callback: If not None, an event watcher will be created and
25                 events passed to the callback.
26         """
27         if os.getenv("KUBERNETES_SERVICE_HOST"):
28             # We are running inside k8s
29             config.load_incluster_config()
30         else:
31             # Use user's kube config
32             config.load_kube_config()
33 
34         self._v1 = client.CoreV1Api()
35         self._logger = logging.getLogger(__name__)
36         self._image = worker_image
37         self._ns = namespace
38         self._job_name = job_name
39         self._event_cb = event_callback
40         if self._event_cb:
41             threading.Thread(
42                 target=self._watch, name="event_watcher", daemon=True
43             ).start()
44 
45     def _watch(self):
46         stream = watch.Watch().stream(
47             self._v1.list_namespaced_pod,
48             self._ns,
49             label_selector="elasticdl_job_name=" + self._job_name,
50         )
51         for event in stream:
52             try:
53                 self._event_cb(event)
54             except Exception:
55                 traceback.print_exc()
56 
57     def get_worker_pod_name(self, worker_id):
58         return WORKER_POD_NAME_PREFIX + self._job_name + "-" + str(worker_id)
59 
60     def _create_worker_pod(
61         self,
62         worker_id,
63         resource_requests,
64         resource_limits,
65         priority,
66         mount_path,
67         volume_name,
68         image_pull_policy,
69         command,
70         args,
71         restart_policy,
72     ):
73         # Worker container config
74         container = client.V1Container(
75             name=self.get_worker_pod_name(worker_id),
76             image=self._image,
77             command=command,
78             resources=client.V1ResourceRequirements(
79                 requests=resource_requests, limits=resource_limits
80             ),
81             image_pull_policy=image_pull_policy,
82             args=args,
83         )
84 
85         # Pod
86         spec = client.V1PodSpec(
87             containers=[container], restart_policy=restart_policy
88         )
89 
90         # Mount data path
91         if mount_path is not None and volume_name is not None:
92             volume = client.V1Volume(
93                 name="data-volume",
94                 persistent_volume_claim=pvcVolumeSource(
95                     claim_name="fileserver-claim", read_only=False
96                 ),
97             )
98             spec.volumes = [volume]
99             container.volume_mounts = [
100                 client.V1VolumeMount(name=volume_name, mount_path=mount_path)
101             ]
102 
103         if priority is not None:
104             spec.priority_class_name = priority
105 
106         pod = client.V1Pod(
107             spec=spec,
108             metadata=client.V1ObjectMeta(
109                 name=self.get_worker_pod_name(worker_id),
110                 labels={
111                     "app": "elasticdl",
112                     "elasticdl_job_name": self._job_name,
113                 },
114             ),
115         )
116         return pod
117 
118     def create_worker(
119         self,
120         worker_id,
121         resource_requests,
122         resource_limits,
123         priority=None,
124         mount_path=None,
125         volume_name=None,
126         image_pull_policy=None,
127         command=None,
128         args=None,
129         restart_policy="OnFailure",
130     ):
131         self._logger.info("Creating worker: " + str(worker_id))
132         pod = self._create_worker_pod(
133             worker_id,
134             resource_requests,
135             resource_limits,
136             priority,
137             mount_path,
138             volume_name,
139             image_pull_policy,
140             command=command,
141             args=args,
142             restart_policy=restart_policy,
143         )
144         return self._v1.create_namespaced_pod(self._ns, pod)
145 
146     def delete_worker(self, worker_id):
147         self._logger.info("Deleting worker: " + str(worker_id))
148         self._v1.delete_namespaced_pod(
149             self.get_worker_pod_name(worker_id),
150             self._ns,
151             body=client.V1DeleteOptions(grace_period_seconds=0),
152         )
153
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/elasticdl/python/elasticdl/master/k8s_client.py b/elasticdl/python/elasticdl/master/k8s_client.py
--- a/elasticdl/python/elasticdl/master/k8s_client.py
+++ b/elasticdl/python/elasticdl/master/k8s_client.py
@@ -103,6 +103,16 @@
         if priority is not None:
             spec.priority_class_name = priority
 
+        # Find that master pod that will be used as the owner reference
+        # for this worker pod.
+        pods = self._v1.list_namespaced_pod(
+            namespace=self._ns,
+            label_selector="elasticdl_job_name=" + self._job_name
+        ).items
+        master_pod = [pod for pod in pods if (
+                pod.metadata.name == "elasticdl-master-" + self._job_name
+        )][0]
+
         pod = client.V1Pod(
             spec=spec,
             metadata=client.V1ObjectMeta(
@@ -111,6 +121,17 @@
                     "app": "elasticdl",
                     "elasticdl_job_name": self._job_name,
                 },
+                # TODO: Add tests for this once we've done refactoring on
+                # k8s client code and the constant strings
+                owner_references=[
+                    client.V1OwnerReference(
+                        api_version="v1",
+                        block_owner_deletion=True,
+                        kind="Pod",
+                        name=master_pod.metadata.name,
+                        uid=master_pod.metadata.uid,
+                    ),
+                ],
             ),
         )
         return pod

{"golden_diff": "diff --git a/elasticdl/python/elasticdl/master/k8s_client.py b/elasticdl/python/elasticdl/master/k8s_client.py\n--- a/elasticdl/python/elasticdl/master/k8s_client.py\n+++ b/elasticdl/python/elasticdl/master/k8s_client.py\n@@ -103,6 +103,16 @@\n         if priority is not None:\n             spec.priority_class_name = priority\n \n+        # Find that master pod that will be used as the owner reference\n+        # for this worker pod.\n+        pods = self._v1.list_namespaced_pod(\n+            namespace=self._ns,\n+            label_selector=\"elasticdl_job_name=\" + self._job_name\n+        ).items\n+        master_pod = [pod for pod in pods if (\n+                pod.metadata.name == \"elasticdl-master-\" + self._job_name\n+        )][0]\n+\n         pod = client.V1Pod(\n             spec=spec,\n             metadata=client.V1ObjectMeta(\n@@ -111,6 +121,17 @@\n                     \"app\": \"elasticdl\",\n                     \"elasticdl_job_name\": self._job_name,\n                 },\n+                # TODO: Add tests for this once we've done refactoring on\n+                # k8s client code and the constant strings\n+                owner_references=[\n+                    client.V1OwnerReference(\n+                        api_version=\"v1\",\n+                        block_owner_deletion=True,\n+                        kind=\"Pod\",\n+                        name=master_pod.metadata.name,\n+                        uid=master_pod.metadata.uid,\n+                    ),\n+                ],\n             ),\n         )\n         return pod\n", "issue": "Create owner references among master and worker pods\nCurrently we have the following two ways to delete master + worker pods:\r\n* Delete each pod one by one\r\n* Delete all pods related to this elasticdl run via `elasticdl_job_name` label `kubectl delete pod -l elasticdl_job_name=test-job-1559292773-93`\r\n\r\nIt would be much easier if users could just delete master pod and then worker pods can be delete automatically. This would be possible if there are owner references among master and worker pods.\n", "before_files": [{"content": "import logging\nimport os\nimport threading\nimport traceback\n\nfrom kubernetes import client, config, watch\nfrom kubernetes.client import (\n    V1PersistentVolumeClaimVolumeSource as pvcVolumeSource,\n)\n\nWORKER_POD_NAME_PREFIX = \"elasticdl-worker-\"\n\n\nclass Client(object):\n    def __init__(self, *, worker_image, namespace, job_name, event_callback):\n        \"\"\"\n        ElasticDL k8s client.\n\n        Args:\n            worker_image: Docker image path for ElasticDL workers.\n            namespace: k8s namespace for ElasticDL pods.\n            job_name: ElasticDL job name, should be unique in the namespace.\n                Used as worker pod name prefix and value for \"elasticdl\" label.\n            event_callback: If not None, an event watcher will be created and\n                events passed to the callback.\n        \"\"\"\n        if os.getenv(\"KUBERNETES_SERVICE_HOST\"):\n            # We are running inside k8s\n            config.load_incluster_config()\n        else:\n            # Use user's kube config\n            config.load_kube_config()\n\n        self._v1 = client.CoreV1Api()\n        self._logger = logging.getLogger(__name__)\n        self._image = worker_image\n        self._ns = namespace\n        self._job_name = job_name\n        self._event_cb = event_callback\n        if self._event_cb:\n            threading.Thread(\n                target=self._watch, name=\"event_watcher\", daemon=True\n            ).start()\n\n    def _watch(self):\n        stream = watch.Watch().stream(\n            self._v1.list_namespaced_pod,\n            self._ns,\n            label_selector=\"elasticdl_job_name=\" + self._job_name,\n        )\n        for event in stream:\n            try:\n                self._event_cb(event)\n            except Exception:\n                traceback.print_exc()\n\n    def get_worker_pod_name(self, worker_id):\n        return WORKER_POD_NAME_PREFIX + self._job_name + \"-\" + str(worker_id)\n\n    def _create_worker_pod(\n        self,\n        worker_id,\n        resource_requests,\n        resource_limits,\n        priority,\n        mount_path,\n        volume_name,\n        image_pull_policy,\n        command,\n        args,\n        restart_policy,\n    ):\n        # Worker container config\n        container = client.V1Container(\n            name=self.get_worker_pod_name(worker_id),\n            image=self._image,\n            command=command,\n            resources=client.V1ResourceRequirements(\n                requests=resource_requests, limits=resource_limits\n            ),\n            image_pull_policy=image_pull_policy,\n            args=args,\n        )\n\n        # Pod\n        spec = client.V1PodSpec(\n            containers=[container], restart_policy=restart_policy\n        )\n\n        # Mount data path\n        if mount_path is not None and volume_name is not None:\n            volume = client.V1Volume(\n                name=\"data-volume\",\n                persistent_volume_claim=pvcVolumeSource(\n                    claim_name=\"fileserver-claim\", read_only=False\n                ),\n            )\n            spec.volumes = [volume]\n            container.volume_mounts = [\n                client.V1VolumeMount(name=volume_name, mount_path=mount_path)\n            ]\n\n        if priority is not None:\n            spec.priority_class_name = priority\n\n        pod = client.V1Pod(\n            spec=spec,\n            metadata=client.V1ObjectMeta(\n                name=self.get_worker_pod_name(worker_id),\n                labels={\n                    \"app\": \"elasticdl\",\n                    \"elasticdl_job_name\": self._job_name,\n                },\n            ),\n        )\n        return pod\n\n    def create_worker(\n        self,\n        worker_id,\n        resource_requests,\n        resource_limits,\n        priority=None,\n        mount_path=None,\n        volume_name=None,\n        image_pull_policy=None,\n        command=None,\n        args=None,\n        restart_policy=\"OnFailure\",\n    ):\n        self._logger.info(\"Creating worker: \" + str(worker_id))\n        pod = self._create_worker_pod(\n            worker_id,\n            resource_requests,\n            resource_limits,\n            priority,\n            mount_path,\n            volume_name,\n            image_pull_policy,\n            command=command,\n            args=args,\n            restart_policy=restart_policy,\n        )\n        return self._v1.create_namespaced_pod(self._ns, pod)\n\n    def delete_worker(self, worker_id):\n        self._logger.info(\"Deleting worker: \" + str(worker_id))\n        self._v1.delete_namespaced_pod(\n            self.get_worker_pod_name(worker_id),\n            self._ns,\n            body=client.V1DeleteOptions(grace_period_seconds=0),\n        )\n", "path": "elasticdl/python/elasticdl/master/k8s_client.py"}], "after_files": [{"content": "import logging\nimport os\nimport threading\nimport traceback\n\nfrom kubernetes import client, config, watch\nfrom kubernetes.client import (\n    V1PersistentVolumeClaimVolumeSource as pvcVolumeSource,\n)\n\nWORKER_POD_NAME_PREFIX = \"elasticdl-worker-\"\n\n\nclass Client(object):\n    def __init__(self, *, worker_image, namespace, job_name, event_callback):\n        \"\"\"\n        ElasticDL k8s client.\n\n        Args:\n            worker_image: Docker image path for ElasticDL workers.\n            namespace: k8s namespace for ElasticDL pods.\n            job_name: ElasticDL job name, should be unique in the namespace.\n                Used as worker pod name prefix and value for \"elasticdl\" label.\n            event_callback: If not None, an event watcher will be created and\n                events passed to the callback.\n        \"\"\"\n        if os.getenv(\"KUBERNETES_SERVICE_HOST\"):\n            # We are running inside k8s\n            config.load_incluster_config()\n        else:\n            # Use user's kube config\n            config.load_kube_config()\n\n        self._v1 = client.CoreV1Api()\n        self._logger = logging.getLogger(__name__)\n        self._image = worker_image\n        self._ns = namespace\n        self._job_name = job_name\n        self._event_cb = event_callback\n        if self._event_cb:\n            threading.Thread(\n                target=self._watch, name=\"event_watcher\", daemon=True\n            ).start()\n\n    def _watch(self):\n        stream = watch.Watch().stream(\n            self._v1.list_namespaced_pod,\n            self._ns,\n            label_selector=\"elasticdl_job_name=\" + self._job_name,\n        )\n        for event in stream:\n            try:\n                self._event_cb(event)\n            except Exception:\n                traceback.print_exc()\n\n    def get_worker_pod_name(self, worker_id):\n        return WORKER_POD_NAME_PREFIX + self._job_name + \"-\" + str(worker_id)\n\n    def _create_worker_pod(\n        self,\n        worker_id,\n        resource_requests,\n        resource_limits,\n        priority,\n        mount_path,\n        volume_name,\n        image_pull_policy,\n        command,\n        args,\n        restart_policy,\n    ):\n        # Worker container config\n        container = client.V1Container(\n            name=self.get_worker_pod_name(worker_id),\n            image=self._image,\n            command=command,\n            resources=client.V1ResourceRequirements(\n                requests=resource_requests, limits=resource_limits\n            ),\n            image_pull_policy=image_pull_policy,\n            args=args,\n        )\n\n        # Pod\n        spec = client.V1PodSpec(\n            containers=[container], restart_policy=restart_policy\n        )\n\n        # Mount data path\n        if mount_path is not None and volume_name is not None:\n            volume = client.V1Volume(\n                name=\"data-volume\",\n                persistent_volume_claim=pvcVolumeSource(\n                    claim_name=\"fileserver-claim\", read_only=False\n                ),\n            )\n            spec.volumes = [volume]\n            container.volume_mounts = [\n                client.V1VolumeMount(name=volume_name, mount_path=mount_path)\n            ]\n\n        if priority is not None:\n            spec.priority_class_name = priority\n\n        # Find that master pod that will be used as the owner reference\n        # for this worker pod.\n        pods = self._v1.list_namespaced_pod(\n            namespace=self._ns,\n            label_selector=\"elasticdl_job_name=\" + self._job_name\n        ).items\n        master_pod = [pod for pod in pods if (\n                pod.metadata.name == \"elasticdl-master-\" + self._job_name\n        )][0]\n\n        pod = client.V1Pod(\n            spec=spec,\n            metadata=client.V1ObjectMeta(\n                name=self.get_worker_pod_name(worker_id),\n                labels={\n                    \"app\": \"elasticdl\",\n                    \"elasticdl_job_name\": self._job_name,\n                },\n                # TODO: Add tests for this once we've done refactoring on\n                # k8s client code and the constant strings\n                owner_references=[\n                    client.V1OwnerReference(\n                        api_version=\"v1\",\n                        block_owner_deletion=True,\n                        kind=\"Pod\",\n                        name=master_pod.metadata.name,\n                        uid=master_pod.metadata.uid,\n                    ),\n                ],\n            ),\n        )\n        return pod\n\n    def create_worker(\n        self,\n        worker_id,\n        resource_requests,\n        resource_limits,\n        priority=None,\n        mount_path=None,\n        volume_name=None,\n        image_pull_policy=None,\n        command=None,\n        args=None,\n        restart_policy=\"OnFailure\",\n    ):\n        self._logger.info(\"Creating worker: \" + str(worker_id))\n        pod = self._create_worker_pod(\n            worker_id,\n            resource_requests,\n            resource_limits,\n            priority,\n            mount_path,\n            volume_name,\n            image_pull_policy,\n            command=command,\n            args=args,\n            restart_policy=restart_policy,\n        )\n        return self._v1.create_namespaced_pod(self._ns, pod)\n\n    def delete_worker(self, worker_id):\n        self._logger.info(\"Deleting worker: \" + str(worker_id))\n        self._v1.delete_namespaced_pod(\n            self.get_worker_pod_name(worker_id),\n            self._ns,\n            body=client.V1DeleteOptions(grace_period_seconds=0),\n        )\n", "path": "elasticdl/python/elasticdl/master/k8s_client.py"}]}