We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Better check for codec names
currently, codec name argument is not checked. A typo would result in worker interpreting encoded data.

--- 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/master/main.py`
Content:
```
1 import logging
2 import time
3 import argparse
4 import os
5 
6 import grpc
7 import tensorflow as tf
8 
9 tf.enable_eager_execution()
10 
11 from concurrent import futures
12 from recordio import File
13 from elasticdl.proto import master_pb2_grpc
14 from elasticdl.master.servicer import MasterServicer
15 from elasticdl.master.task_queue import _TaskQueue
16 from elasticdl.master.k8s_worker_manager import WorkerManager
17 from elasticdl.common.model_helper import load_user_model, build_model
18 
19 
20 def _make_task_queue(data_dir, record_per_task, num_epoch):
21     f_records = {}
22     for f in os.listdir(data_dir):
23         p = os.path.join(data_dir, f)
24         with File(p, "r") as rio:
25             f_records[p] = rio.count()
26     return _TaskQueue(f_records, record_per_task, num_epoch)
27 
28 
29 def _parse_args():
30     parser = argparse.ArgumentParser(description="ElasticDL Master")
31     parser.add_argument(
32         "--model_file",
33         help="Full file path of user defined neural model",
34         required=True,
35     )
36     parser.add_argument(
37         "--train_data_dir",
38         help="Training data directory. Files should be in RecordIO format",
39         required=True,
40     )
41     parser.add_argument("--record_per_task", type=int, required=True)
42     parser.add_argument("--num_epoch", type=int, required=True)
43     parser.add_argument(
44         "--grads_to_wait",
45         type=int,
46         help="Number of gradients to wait before updating model",
47         required=True,
48     )
49     parser.add_argument(
50         "--minibatch_size",
51         type=int,
52         help="Minibatch size used by workers to compute gradients",
53         required=True,
54     )
55     parser.add_argument(
56         "--num_worker",
57         type=int,
58         help="the number of workers used in training",
59         default=0,
60     )
61     parser.add_argument(
62         "--worker_image", help="docker image for worker", default=None
63     )
64     parser.add_argument("--job_name", help="job name", required=True)
65     parser.add_argument(
66         "--codec-type",
67         default=None,
68         help="Type of codec(tf_example or None)",
69     )
70     return parser.parse_args()
71 
72 
73 def main():
74     # TODO: pass port via flags.
75     PORT = 50001
76     logger = logging.getLogger("master")
77     args = _parse_args()
78     task_q = _make_task_queue(
79         args.train_data_dir, args.record_per_task, args.num_epoch
80     )
81     model_module = load_user_model(args.model_file)
82     model_inst = model_module.model
83     build_model(model_inst, model_module.feature_columns())
84     optimizer = model_module.optimizer()
85 
86     server = grpc.server(futures.ThreadPoolExecutor(max_workers=64))
87     master_pb2_grpc.add_MasterServicer_to_server(
88         MasterServicer(
89             logger,
90             args.grads_to_wait,
91             args.minibatch_size,
92             optimizer,
93             task_q,
94             init_var=model_inst.trainable_variables,
95         ),
96         server,
97     )
98     server.add_insecure_port("[::]:{}".format(PORT))
99     server.start()
100     logger.warning("Server started at port: %d", PORT)
101 
102     if args.num_worker:
103         master_addr = "%s:%d" % (os.getenv("MY_POD_IP", "localhost"), PORT)
104         worker_command = ["python"]
105         worker_args = [
106             "-m",
107             "elasticdl.worker.main",
108             "--model_file",
109             args.model_file,
110             "--master_addr",
111             master_addr,
112             "--codec-type",
113             args.codec_type
114         ]
115 
116         worker_manager = WorkerManager(
117             job_name=args.job_name,
118             worker_image=args.worker_image,
119             command=worker_command,
120             args=worker_args,
121             namespace="default",
122             num_worker=args.num_worker,
123         )
124         worker_manager.start_workers(restart_policy="Never")
125 
126     try:
127         while True:
128             if task_q.finished():
129                 break
130             time.sleep(30)
131     except KeyboardInterrupt:
132         logger.warning("Server stopping")
133 
134     if args.num_worker:
135         # TODO: worker_manager.remove_workers supports synchronized call
136         worker_manager.remove_workers()
137         # wait for worker pod to be deleted
138         max_check_num = 10
139         for _ in range(max_check_num):
140             time.sleep(3)
141             counters = worker_manager.get_counters()
142             if not counters:
143                 break
144     server.stop(0)
145 
146 
147 if __name__ == "__main__":
148     logging.basicConfig()
149     main()
150
```
--- 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/master/main.py b/elasticdl/master/main.py
--- a/elasticdl/master/main.py
+++ b/elasticdl/master/main.py
@@ -65,6 +65,7 @@
     parser.add_argument(
         "--codec-type",
         default=None,
+        choices=["tf_example"],
         help="Type of codec(tf_example or None)",
     )
     return parser.parse_args()

{"golden_diff": "diff --git a/elasticdl/master/main.py b/elasticdl/master/main.py\n--- a/elasticdl/master/main.py\n+++ b/elasticdl/master/main.py\n@@ -65,6 +65,7 @@\n     parser.add_argument(\n         \"--codec-type\",\n         default=None,\n+        choices=[\"tf_example\"],\n         help=\"Type of codec(tf_example or None)\",\n     )\n     return parser.parse_args()\n", "issue": "Better check for codec names\ncurrently, codec name argument is not checked. A typo would result in worker interpreting encoded data.\n", "before_files": [{"content": "import logging\nimport time\nimport argparse\nimport os\n\nimport grpc\nimport tensorflow as tf\n\ntf.enable_eager_execution()\n\nfrom concurrent import futures\nfrom recordio import File\nfrom elasticdl.proto import master_pb2_grpc\nfrom elasticdl.master.servicer import MasterServicer\nfrom elasticdl.master.task_queue import _TaskQueue\nfrom elasticdl.master.k8s_worker_manager import WorkerManager\nfrom elasticdl.common.model_helper import load_user_model, build_model\n\n\ndef _make_task_queue(data_dir, record_per_task, num_epoch):\n    f_records = {}\n    for f in os.listdir(data_dir):\n        p = os.path.join(data_dir, f)\n        with File(p, \"r\") as rio:\n            f_records[p] = rio.count()\n    return _TaskQueue(f_records, record_per_task, num_epoch)\n\n\ndef _parse_args():\n    parser = argparse.ArgumentParser(description=\"ElasticDL Master\")\n    parser.add_argument(\n        \"--model_file\",\n        help=\"Full file path of user defined neural model\",\n        required=True,\n    )\n    parser.add_argument(\n        \"--train_data_dir\",\n        help=\"Training data directory. Files should be in RecordIO format\",\n        required=True,\n    )\n    parser.add_argument(\"--record_per_task\", type=int, required=True)\n    parser.add_argument(\"--num_epoch\", type=int, required=True)\n    parser.add_argument(\n        \"--grads_to_wait\",\n        type=int,\n        help=\"Number of gradients to wait before updating model\",\n        required=True,\n    )\n    parser.add_argument(\n        \"--minibatch_size\",\n        type=int,\n        help=\"Minibatch size used by workers to compute gradients\",\n        required=True,\n    )\n    parser.add_argument(\n        \"--num_worker\",\n        type=int,\n        help=\"the number of workers used in training\",\n        default=0,\n    )\n    parser.add_argument(\n        \"--worker_image\", help=\"docker image for worker\", default=None\n    )\n    parser.add_argument(\"--job_name\", help=\"job name\", required=True)\n    parser.add_argument(\n        \"--codec-type\",\n        default=None,\n        help=\"Type of codec(tf_example or None)\",\n    )\n    return parser.parse_args()\n\n\ndef main():\n    # TODO: pass port via flags.\n    PORT = 50001\n    logger = logging.getLogger(\"master\")\n    args = _parse_args()\n    task_q = _make_task_queue(\n        args.train_data_dir, args.record_per_task, args.num_epoch\n    )\n    model_module = load_user_model(args.model_file)\n    model_inst = model_module.model\n    build_model(model_inst, model_module.feature_columns())\n    optimizer = model_module.optimizer()\n\n    server = grpc.server(futures.ThreadPoolExecutor(max_workers=64))\n    master_pb2_grpc.add_MasterServicer_to_server(\n        MasterServicer(\n            logger,\n            args.grads_to_wait,\n            args.minibatch_size,\n            optimizer,\n            task_q,\n            init_var=model_inst.trainable_variables,\n        ),\n        server,\n    )\n    server.add_insecure_port(\"[::]:{}\".format(PORT))\n    server.start()\n    logger.warning(\"Server started at port: %d\", PORT)\n\n    if args.num_worker:\n        master_addr = \"%s:%d\" % (os.getenv(\"MY_POD_IP\", \"localhost\"), PORT)\n        worker_command = [\"python\"]\n        worker_args = [\n            \"-m\",\n            \"elasticdl.worker.main\",\n            \"--model_file\",\n            args.model_file,\n            \"--master_addr\",\n            master_addr,\n            \"--codec-type\",\n            args.codec_type\n        ]\n\n        worker_manager = WorkerManager(\n            job_name=args.job_name,\n            worker_image=args.worker_image,\n            command=worker_command,\n            args=worker_args,\n            namespace=\"default\",\n            num_worker=args.num_worker,\n        )\n        worker_manager.start_workers(restart_policy=\"Never\")\n\n    try:\n        while True:\n            if task_q.finished():\n                break\n            time.sleep(30)\n    except KeyboardInterrupt:\n        logger.warning(\"Server stopping\")\n\n    if args.num_worker:\n        # TODO: worker_manager.remove_workers supports synchronized call\n        worker_manager.remove_workers()\n        # wait for worker pod to be deleted\n        max_check_num = 10\n        for _ in range(max_check_num):\n            time.sleep(3)\n            counters = worker_manager.get_counters()\n            if not counters:\n                break\n    server.stop(0)\n\n\nif __name__ == \"__main__\":\n    logging.basicConfig()\n    main()\n", "path": "elasticdl/master/main.py"}], "after_files": [{"content": "import logging\nimport time\nimport argparse\nimport os\n\nimport grpc\nimport tensorflow as tf\n\ntf.enable_eager_execution()\n\nfrom concurrent import futures\nfrom recordio import File\nfrom elasticdl.proto import master_pb2_grpc\nfrom elasticdl.master.servicer import MasterServicer\nfrom elasticdl.master.task_queue import _TaskQueue\nfrom elasticdl.master.k8s_worker_manager import WorkerManager\nfrom elasticdl.common.model_helper import load_user_model, build_model\n\n\ndef _make_task_queue(data_dir, record_per_task, num_epoch):\n    f_records = {}\n    for f in os.listdir(data_dir):\n        p = os.path.join(data_dir, f)\n        with File(p, \"r\") as rio:\n            f_records[p] = rio.count()\n    return _TaskQueue(f_records, record_per_task, num_epoch)\n\n\ndef _parse_args():\n    parser = argparse.ArgumentParser(description=\"ElasticDL Master\")\n    parser.add_argument(\n        \"--model_file\",\n        help=\"Full file path of user defined neural model\",\n        required=True,\n    )\n    parser.add_argument(\n        \"--train_data_dir\",\n        help=\"Training data directory. Files should be in RecordIO format\",\n        required=True,\n    )\n    parser.add_argument(\"--record_per_task\", type=int, required=True)\n    parser.add_argument(\"--num_epoch\", type=int, required=True)\n    parser.add_argument(\n        \"--grads_to_wait\",\n        type=int,\n        help=\"Number of gradients to wait before updating model\",\n        required=True,\n    )\n    parser.add_argument(\n        \"--minibatch_size\",\n        type=int,\n        help=\"Minibatch size used by workers to compute gradients\",\n        required=True,\n    )\n    parser.add_argument(\n        \"--num_worker\",\n        type=int,\n        help=\"the number of workers used in training\",\n        default=0,\n    )\n    parser.add_argument(\n        \"--worker_image\", help=\"docker image for worker\", default=None\n    )\n    parser.add_argument(\"--job_name\", help=\"job name\", required=True)\n    parser.add_argument(\n        \"--codec-type\",\n        default=None,\n        choices=[\"tf_example\"],\n        help=\"Type of codec(tf_example or None)\",\n    )\n    return parser.parse_args()\n\n\ndef main():\n    # TODO: pass port via flags.\n    PORT = 50001\n    logger = logging.getLogger(\"master\")\n    args = _parse_args()\n    task_q = _make_task_queue(\n        args.train_data_dir, args.record_per_task, args.num_epoch\n    )\n    model_module = load_user_model(args.model_file)\n    model_inst = model_module.model\n    build_model(model_inst, model_module.feature_columns())\n    optimizer = model_module.optimizer()\n\n    server = grpc.server(futures.ThreadPoolExecutor(max_workers=64))\n    master_pb2_grpc.add_MasterServicer_to_server(\n        MasterServicer(\n            logger,\n            args.grads_to_wait,\n            args.minibatch_size,\n            optimizer,\n            task_q,\n            init_var=model_inst.trainable_variables,\n        ),\n        server,\n    )\n    server.add_insecure_port(\"[::]:{}\".format(PORT))\n    server.start()\n    logger.warning(\"Server started at port: %d\", PORT)\n\n    if args.num_worker:\n        master_addr = \"%s:%d\" % (os.getenv(\"MY_POD_IP\", \"localhost\"), PORT)\n        worker_command = [\"python\"]\n        worker_args = [\n            \"-m\",\n            \"elasticdl.worker.main\",\n            \"--model_file\",\n            args.model_file,\n            \"--master_addr\",\n            master_addr,\n            \"--codec-type\",\n            args.codec_type\n        ]\n\n        worker_manager = WorkerManager(\n            job_name=args.job_name,\n            worker_image=args.worker_image,\n            command=worker_command,\n            args=worker_args,\n            namespace=\"default\",\n            num_worker=args.num_worker,\n        )\n        worker_manager.start_workers(restart_policy=\"Never\")\n\n    try:\n        while True:\n            if task_q.finished():\n                break\n            time.sleep(30)\n    except KeyboardInterrupt:\n        logger.warning(\"Server stopping\")\n\n    if args.num_worker:\n        # TODO: worker_manager.remove_workers supports synchronized call\n        worker_manager.remove_workers()\n        # wait for worker pod to be deleted\n        max_check_num = 10\n        for _ in range(max_check_num):\n            time.sleep(3)\n            counters = worker_manager.get_counters()\n            if not counters:\n                break\n    server.stop(0)\n\n\nif __name__ == \"__main__\":\n    logging.basicConfig()\n    main()\n", "path": "elasticdl/master/main.py"}]}