We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[v2compat] wrong param passed when param contains JSON
### Environment

*  KFP SDK version: master branch

### Steps to reproduce

```python
import kfp
from kfp import dsl
from kfp.components import create_component_from_func


@create_component_from_func
def test(json_param: str):
    print(json_param)


@dsl.pipeline(name='bug-json-param')
def my_pipeline(
    json_param: str = '{"a":"b"}',
    value: str = 'c',
):
    test(json_param).set_display_name('test1')
    test('{"a":"' + f'{value}' + '"}').set_display_name('test2')


if __name__ == '__main__':
    kfp.Client().create_run_from_pipeline_func(
        my_pipeline,
        arguments={},
        mode=dsl.PipelineExecutionMode.V2_COMPATIBLE,
    )
```

I got the following log for `test2` task (`task1` is expected).
```
{\"a\":\"c\"}
```

### Expected result

<!-- What should the correct behavior be? -->
```
{"a":"c"}
```

### Materials and Reference

<!-- Help us debug this issue by providing resources such as: sample code, background context, or links to references. -->

---

<!-- Don't delete message below to encourage users to support your issue! -->
Impacted by this bug? Give it a 👍. We prioritise the issues with the most 👍.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sdk/python/kfp/compiler/v2_compat.py`
Content:
```
1 # Copyright 2021 The Kubeflow Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """Utility functions for enabling v2-compatible pipelines in v1."""
15 import collections
16 import json
17 from typing import Optional
18 
19 from kfp import dsl
20 from kfp.compiler import _default_transformers
21 from kfp.pipeline_spec import pipeline_spec_pb2
22 from kfp.v2 import compiler
23 
24 from kubernetes import client as k8s_client
25 
26 _DEFAULT_LAUNCHER_IMAGE = "gcr.io/ml-pipeline/kfp-launcher:1.6.4"
27 
28 
29 def update_op(op: dsl.ContainerOp,
30               pipeline_name: dsl.PipelineParam,
31               pipeline_root: dsl.PipelineParam,
32               launcher_image: Optional[str] = None) -> None:
33   """Updates the passed in Op for running in v2-compatible mode.
34 
35     Args:
36       op: The Op to update.
37       pipeline_spec: The PipelineSpec for the pipeline under which `op`
38         runs.
39       pipeline_root: The root output directory for pipeline artifacts.
40       launcher_image: An optional launcher image. Useful for tests.
41     """
42   op.is_v2 = True
43   # Inject the launcher binary and overwrite the entrypoint.
44   image_name = launcher_image or _DEFAULT_LAUNCHER_IMAGE
45   launcher_container = dsl.UserContainer(name="kfp-launcher",
46                                          image=image_name,
47                                          command="/bin/mount_launcher.sh",
48                                          mirror_volume_mounts=True)
49 
50   op.add_init_container(launcher_container)
51   op.add_volume(k8s_client.V1Volume(name='kfp-launcher'))
52   op.add_volume_mount(
53       k8s_client.V1VolumeMount(name='kfp-launcher', mount_path='/kfp-launcher'))
54 
55   op.command = [
56       "/kfp-launcher/launch",
57       "--mlmd_server_address",
58       "$(METADATA_GRPC_SERVICE_HOST)",
59       "--mlmd_server_port",
60       "$(METADATA_GRPC_SERVICE_PORT)",
61       "--runtime_info_json",
62       "$(KFP_V2_RUNTIME_INFO)",
63       "--container_image",
64       "$(KFP_V2_IMAGE)",
65       "--task_name",
66       op.name,
67       "--pipeline_name",
68       pipeline_name,
69       "--pipeline_run_id",
70       "$(WORKFLOW_ID)",
71       "--pipeline_task_id",
72       "$(KFP_POD_NAME)",
73       "--pipeline_root",
74       pipeline_root,
75   ]
76 
77   # Mount necessary environment variables.
78   op.apply(_default_transformers.add_kfp_pod_env)
79   op.container.add_env_variable(
80       k8s_client.V1EnvVar(name="KFP_V2_IMAGE", value=op.container.image))
81 
82   config_map_ref = k8s_client.V1ConfigMapEnvSource(
83       name='metadata-grpc-configmap', optional=True)
84   op.container.add_env_from(
85       k8s_client.V1EnvFromSource(config_map_ref=config_map_ref))
86 
87   op.arguments = list(op.container_spec.command) + list(op.container_spec.args)
88 
89   runtime_info = {
90       "inputParameters": collections.OrderedDict(),
91       "inputArtifacts": collections.OrderedDict(),
92       "outputParameters": collections.OrderedDict(),
93       "outputArtifacts": collections.OrderedDict(),
94   }
95 
96   component_spec = op.component_spec
97   for parameter, spec in sorted(
98       component_spec.input_definitions.parameters.items()):
99     parameter_info = {
100         "type":
101             pipeline_spec_pb2.PrimitiveType.PrimitiveTypeEnum.Name(spec.type),
102         "value":
103             "BEGIN-KFP-PARAM[{}]END-KFP-PARAM".format(
104                 op._parameter_arguments[parameter])
105     }
106     runtime_info["inputParameters"][parameter] = parameter_info
107 
108   for artifact_name, spec in sorted(
109       component_spec.input_definitions.artifacts.items()):
110     artifact_info = {
111         "metadataPath": op.input_artifact_paths[artifact_name],
112         "schemaTitle": spec.artifact_type.schema_title,
113         "instanceSchema": spec.artifact_type.instance_schema,
114     }
115     runtime_info["inputArtifacts"][artifact_name] = artifact_info
116 
117   for parameter, spec in sorted(
118       component_spec.output_definitions.parameters.items()):
119     parameter_info = {
120         "type":
121             pipeline_spec_pb2.PrimitiveType.PrimitiveTypeEnum.Name(spec.type),
122         "path":
123             op.file_outputs[parameter],
124     }
125     runtime_info["outputParameters"][parameter] = parameter_info
126 
127   for artifact_name, spec in sorted(
128       component_spec.output_definitions.artifacts.items()):
129     # TODO: Assert instance_schema.
130     artifact_info = {
131         # Type used to register output artifacts.
132         "schemaTitle": spec.artifact_type.schema_title,
133         "instanceSchema": spec.artifact_type.instance_schema,
134         # File used to write out the registered artifact ID.
135         "metadataPath": op.file_outputs[artifact_name],
136     }
137     runtime_info["outputArtifacts"][artifact_name] = artifact_info
138 
139   op.container.add_env_variable(
140       k8s_client.V1EnvVar(name="KFP_V2_RUNTIME_INFO",
141                           value=json.dumps(runtime_info)))
142 
143   op.pod_annotations['pipelines.kubeflow.org/v2_component'] = "true"
144   op.pod_labels['pipelines.kubeflow.org/v2_component']= "true"
145
```
--- 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/sdk/python/kfp/compiler/v2_compat.py b/sdk/python/kfp/compiler/v2_compat.py
--- a/sdk/python/kfp/compiler/v2_compat.py
+++ b/sdk/python/kfp/compiler/v2_compat.py
@@ -52,6 +52,40 @@
   op.add_volume_mount(
       k8s_client.V1VolumeMount(name='kfp-launcher', mount_path='/kfp-launcher'))
 
+  # op.command + op.args will have the following sections:
+  # 1. args passed to kfp-launcher
+  # 2. a separator "--"
+  # 3. parameters in format "key1=value1", "key2=value2", ...
+  # 4. a separator "--" as end of arguments passed to launcher
+  # 5. (start of op.args) arguments of the original user program command + args
+  #
+  # example:
+  # - command:
+  # - /kfp-launcher/launch
+  # - '--mlmd_server_address'
+  # - $(METADATA_GRPC_SERVICE_HOST)
+  # - '--mlmd_server_port'
+  # - $(METADATA_GRPC_SERVICE_PORT)
+  # - ... # more launcher params
+  # - '--pipeline_task_id'
+  # - $(KFP_POD_NAME)
+  # - '--pipeline_root'
+  # - ''
+  # - '--' # start of parameter values
+  # - first=first
+  # - second=second
+  # - '--' # start of user command and args
+  # args:
+  # - sh
+  # - '-ec'
+  # - |
+  #     program_path=$(mktemp)
+  #     printf "%s" "$0" > "$program_path"
+  #     python3 -u "$program_path" "$@"
+  # - >
+  #     import json
+  #     import xxx
+  #     ...
   op.command = [
       "/kfp-launcher/launch",
       "--mlmd_server_address",
@@ -93,17 +127,17 @@
       "outputArtifacts": collections.OrderedDict(),
   }
 
+  op.command += ["--"]
   component_spec = op.component_spec
   for parameter, spec in sorted(
       component_spec.input_definitions.parameters.items()):
     parameter_info = {
         "type":
             pipeline_spec_pb2.PrimitiveType.PrimitiveTypeEnum.Name(spec.type),
-        "value":
-            "BEGIN-KFP-PARAM[{}]END-KFP-PARAM".format(
-                op._parameter_arguments[parameter])
     }
+    op.command += [f"{parameter}={op._parameter_arguments[parameter]}"]
     runtime_info["inputParameters"][parameter] = parameter_info
+  op.command += ["--"]
 
   for artifact_name, spec in sorted(
       component_spec.input_definitions.artifacts.items()):

{"golden_diff": "diff --git a/sdk/python/kfp/compiler/v2_compat.py b/sdk/python/kfp/compiler/v2_compat.py\n--- a/sdk/python/kfp/compiler/v2_compat.py\n+++ b/sdk/python/kfp/compiler/v2_compat.py\n@@ -52,6 +52,40 @@\n   op.add_volume_mount(\n       k8s_client.V1VolumeMount(name='kfp-launcher', mount_path='/kfp-launcher'))\n \n+  # op.command + op.args will have the following sections:\n+  # 1. args passed to kfp-launcher\n+  # 2. a separator \"--\"\n+  # 3. parameters in format \"key1=value1\", \"key2=value2\", ...\n+  # 4. a separator \"--\" as end of arguments passed to launcher\n+  # 5. (start of op.args) arguments of the original user program command + args\n+  #\n+  # example:\n+  # - command:\n+  # - /kfp-launcher/launch\n+  # - '--mlmd_server_address'\n+  # - $(METADATA_GRPC_SERVICE_HOST)\n+  # - '--mlmd_server_port'\n+  # - $(METADATA_GRPC_SERVICE_PORT)\n+  # - ... # more launcher params\n+  # - '--pipeline_task_id'\n+  # - $(KFP_POD_NAME)\n+  # - '--pipeline_root'\n+  # - ''\n+  # - '--' # start of parameter values\n+  # - first=first\n+  # - second=second\n+  # - '--' # start of user command and args\n+  # args:\n+  # - sh\n+  # - '-ec'\n+  # - |\n+  #     program_path=$(mktemp)\n+  #     printf \"%s\" \"$0\" > \"$program_path\"\n+  #     python3 -u \"$program_path\" \"$@\"\n+  # - >\n+  #     import json\n+  #     import xxx\n+  #     ...\n   op.command = [\n       \"/kfp-launcher/launch\",\n       \"--mlmd_server_address\",\n@@ -93,17 +127,17 @@\n       \"outputArtifacts\": collections.OrderedDict(),\n   }\n \n+  op.command += [\"--\"]\n   component_spec = op.component_spec\n   for parameter, spec in sorted(\n       component_spec.input_definitions.parameters.items()):\n     parameter_info = {\n         \"type\":\n             pipeline_spec_pb2.PrimitiveType.PrimitiveTypeEnum.Name(spec.type),\n-        \"value\":\n-            \"BEGIN-KFP-PARAM[{}]END-KFP-PARAM\".format(\n-                op._parameter_arguments[parameter])\n     }\n+    op.command += [f\"{parameter}={op._parameter_arguments[parameter]}\"]\n     runtime_info[\"inputParameters\"][parameter] = parameter_info\n+  op.command += [\"--\"]\n \n   for artifact_name, spec in sorted(\n       component_spec.input_definitions.artifacts.items()):\n", "issue": "[v2compat] wrong param passed when param contains JSON\n### Environment\r\n\r\n*  KFP SDK version: master branch\r\n\r\n### Steps to reproduce\r\n\r\n```python\r\nimport kfp\r\nfrom kfp import dsl\r\nfrom kfp.components import create_component_from_func\r\n\r\n\r\n@create_component_from_func\r\ndef test(json_param: str):\r\n    print(json_param)\r\n\r\n\r\n@dsl.pipeline(name='bug-json-param')\r\ndef my_pipeline(\r\n    json_param: str = '{\"a\":\"b\"}',\r\n    value: str = 'c',\r\n):\r\n    test(json_param).set_display_name('test1')\r\n    test('{\"a\":\"' + f'{value}' + '\"}').set_display_name('test2')\r\n\r\n\r\nif __name__ == '__main__':\r\n    kfp.Client().create_run_from_pipeline_func(\r\n        my_pipeline,\r\n        arguments={},\r\n        mode=dsl.PipelineExecutionMode.V2_COMPATIBLE,\r\n    )\r\n```\r\n\r\nI got the following log for `test2` task (`task1` is expected).\r\n```\r\n{\\\"a\\\":\\\"c\\\"}\r\n```\r\n\r\n### Expected result\r\n\r\n<!-- What should the correct behavior be? -->\r\n```\r\n{\"a\":\"c\"}\r\n```\r\n\r\n### Materials and Reference\r\n\r\n<!-- Help us debug this issue by providing resources such as: sample code, background context, or links to references. -->\r\n\r\n---\r\n\r\n<!-- Don't delete message below to encourage users to support your issue! -->\r\nImpacted by this bug? Give it a \ud83d\udc4d. We prioritise the issues with the most \ud83d\udc4d.\r\n\n", "before_files": [{"content": "# Copyright 2021 The Kubeflow Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Utility functions for enabling v2-compatible pipelines in v1.\"\"\"\nimport collections\nimport json\nfrom typing import Optional\n\nfrom kfp import dsl\nfrom kfp.compiler import _default_transformers\nfrom kfp.pipeline_spec import pipeline_spec_pb2\nfrom kfp.v2 import compiler\n\nfrom kubernetes import client as k8s_client\n\n_DEFAULT_LAUNCHER_IMAGE = \"gcr.io/ml-pipeline/kfp-launcher:1.6.4\"\n\n\ndef update_op(op: dsl.ContainerOp,\n              pipeline_name: dsl.PipelineParam,\n              pipeline_root: dsl.PipelineParam,\n              launcher_image: Optional[str] = None) -> None:\n  \"\"\"Updates the passed in Op for running in v2-compatible mode.\n\n    Args:\n      op: The Op to update.\n      pipeline_spec: The PipelineSpec for the pipeline under which `op`\n        runs.\n      pipeline_root: The root output directory for pipeline artifacts.\n      launcher_image: An optional launcher image. Useful for tests.\n    \"\"\"\n  op.is_v2 = True\n  # Inject the launcher binary and overwrite the entrypoint.\n  image_name = launcher_image or _DEFAULT_LAUNCHER_IMAGE\n  launcher_container = dsl.UserContainer(name=\"kfp-launcher\",\n                                         image=image_name,\n                                         command=\"/bin/mount_launcher.sh\",\n                                         mirror_volume_mounts=True)\n\n  op.add_init_container(launcher_container)\n  op.add_volume(k8s_client.V1Volume(name='kfp-launcher'))\n  op.add_volume_mount(\n      k8s_client.V1VolumeMount(name='kfp-launcher', mount_path='/kfp-launcher'))\n\n  op.command = [\n      \"/kfp-launcher/launch\",\n      \"--mlmd_server_address\",\n      \"$(METADATA_GRPC_SERVICE_HOST)\",\n      \"--mlmd_server_port\",\n      \"$(METADATA_GRPC_SERVICE_PORT)\",\n      \"--runtime_info_json\",\n      \"$(KFP_V2_RUNTIME_INFO)\",\n      \"--container_image\",\n      \"$(KFP_V2_IMAGE)\",\n      \"--task_name\",\n      op.name,\n      \"--pipeline_name\",\n      pipeline_name,\n      \"--pipeline_run_id\",\n      \"$(WORKFLOW_ID)\",\n      \"--pipeline_task_id\",\n      \"$(KFP_POD_NAME)\",\n      \"--pipeline_root\",\n      pipeline_root,\n  ]\n\n  # Mount necessary environment variables.\n  op.apply(_default_transformers.add_kfp_pod_env)\n  op.container.add_env_variable(\n      k8s_client.V1EnvVar(name=\"KFP_V2_IMAGE\", value=op.container.image))\n\n  config_map_ref = k8s_client.V1ConfigMapEnvSource(\n      name='metadata-grpc-configmap', optional=True)\n  op.container.add_env_from(\n      k8s_client.V1EnvFromSource(config_map_ref=config_map_ref))\n\n  op.arguments = list(op.container_spec.command) + list(op.container_spec.args)\n\n  runtime_info = {\n      \"inputParameters\": collections.OrderedDict(),\n      \"inputArtifacts\": collections.OrderedDict(),\n      \"outputParameters\": collections.OrderedDict(),\n      \"outputArtifacts\": collections.OrderedDict(),\n  }\n\n  component_spec = op.component_spec\n  for parameter, spec in sorted(\n      component_spec.input_definitions.parameters.items()):\n    parameter_info = {\n        \"type\":\n            pipeline_spec_pb2.PrimitiveType.PrimitiveTypeEnum.Name(spec.type),\n        \"value\":\n            \"BEGIN-KFP-PARAM[{}]END-KFP-PARAM\".format(\n                op._parameter_arguments[parameter])\n    }\n    runtime_info[\"inputParameters\"][parameter] = parameter_info\n\n  for artifact_name, spec in sorted(\n      component_spec.input_definitions.artifacts.items()):\n    artifact_info = {\n        \"metadataPath\": op.input_artifact_paths[artifact_name],\n        \"schemaTitle\": spec.artifact_type.schema_title,\n        \"instanceSchema\": spec.artifact_type.instance_schema,\n    }\n    runtime_info[\"inputArtifacts\"][artifact_name] = artifact_info\n\n  for parameter, spec in sorted(\n      component_spec.output_definitions.parameters.items()):\n    parameter_info = {\n        \"type\":\n            pipeline_spec_pb2.PrimitiveType.PrimitiveTypeEnum.Name(spec.type),\n        \"path\":\n            op.file_outputs[parameter],\n    }\n    runtime_info[\"outputParameters\"][parameter] = parameter_info\n\n  for artifact_name, spec in sorted(\n      component_spec.output_definitions.artifacts.items()):\n    # TODO: Assert instance_schema.\n    artifact_info = {\n        # Type used to register output artifacts.\n        \"schemaTitle\": spec.artifact_type.schema_title,\n        \"instanceSchema\": spec.artifact_type.instance_schema,\n        # File used to write out the registered artifact ID.\n        \"metadataPath\": op.file_outputs[artifact_name],\n    }\n    runtime_info[\"outputArtifacts\"][artifact_name] = artifact_info\n\n  op.container.add_env_variable(\n      k8s_client.V1EnvVar(name=\"KFP_V2_RUNTIME_INFO\",\n                          value=json.dumps(runtime_info)))\n\n  op.pod_annotations['pipelines.kubeflow.org/v2_component'] = \"true\"\n  op.pod_labels['pipelines.kubeflow.org/v2_component']= \"true\"\n", "path": "sdk/python/kfp/compiler/v2_compat.py"}], "after_files": [{"content": "# Copyright 2021 The Kubeflow Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Utility functions for enabling v2-compatible pipelines in v1.\"\"\"\nimport collections\nimport json\nfrom typing import Optional\n\nfrom kfp import dsl\nfrom kfp.compiler import _default_transformers\nfrom kfp.pipeline_spec import pipeline_spec_pb2\nfrom kfp.v2 import compiler\n\nfrom kubernetes import client as k8s_client\n\n_DEFAULT_LAUNCHER_IMAGE = \"gcr.io/ml-pipeline/kfp-launcher:1.6.4\"\n\n\ndef update_op(op: dsl.ContainerOp,\n              pipeline_name: dsl.PipelineParam,\n              pipeline_root: dsl.PipelineParam,\n              launcher_image: Optional[str] = None) -> None:\n  \"\"\"Updates the passed in Op for running in v2-compatible mode.\n\n    Args:\n      op: The Op to update.\n      pipeline_spec: The PipelineSpec for the pipeline under which `op`\n        runs.\n      pipeline_root: The root output directory for pipeline artifacts.\n      launcher_image: An optional launcher image. Useful for tests.\n    \"\"\"\n  op.is_v2 = True\n  # Inject the launcher binary and overwrite the entrypoint.\n  image_name = launcher_image or _DEFAULT_LAUNCHER_IMAGE\n  launcher_container = dsl.UserContainer(name=\"kfp-launcher\",\n                                         image=image_name,\n                                         command=\"/bin/mount_launcher.sh\",\n                                         mirror_volume_mounts=True)\n\n  op.add_init_container(launcher_container)\n  op.add_volume(k8s_client.V1Volume(name='kfp-launcher'))\n  op.add_volume_mount(\n      k8s_client.V1VolumeMount(name='kfp-launcher', mount_path='/kfp-launcher'))\n\n  # op.command + op.args will have the following sections:\n  # 1. args passed to kfp-launcher\n  # 2. a separator \"--\"\n  # 3. parameters in format \"key1=value1\", \"key2=value2\", ...\n  # 4. a separator \"--\" as end of arguments passed to launcher\n  # 5. (start of op.args) arguments of the original user program command + args\n  #\n  # example:\n  # - command:\n  # - /kfp-launcher/launch\n  # - '--mlmd_server_address'\n  # - $(METADATA_GRPC_SERVICE_HOST)\n  # - '--mlmd_server_port'\n  # - $(METADATA_GRPC_SERVICE_PORT)\n  # - ... # more launcher params\n  # - '--pipeline_task_id'\n  # - $(KFP_POD_NAME)\n  # - '--pipeline_root'\n  # - ''\n  # - '--' # start of parameter values\n  # - first=first\n  # - second=second\n  # - '--' # start of user command and args\n  # args:\n  # - sh\n  # - '-ec'\n  # - |\n  #     program_path=$(mktemp)\n  #     printf \"%s\" \"$0\" > \"$program_path\"\n  #     python3 -u \"$program_path\" \"$@\"\n  # - >\n  #     import json\n  #     import xxx\n  #     ...\n  op.command = [\n      \"/kfp-launcher/launch\",\n      \"--mlmd_server_address\",\n      \"$(METADATA_GRPC_SERVICE_HOST)\",\n      \"--mlmd_server_port\",\n      \"$(METADATA_GRPC_SERVICE_PORT)\",\n      \"--runtime_info_json\",\n      \"$(KFP_V2_RUNTIME_INFO)\",\n      \"--container_image\",\n      \"$(KFP_V2_IMAGE)\",\n      \"--task_name\",\n      op.name,\n      \"--pipeline_name\",\n      pipeline_name,\n      \"--pipeline_run_id\",\n      \"$(WORKFLOW_ID)\",\n      \"--pipeline_task_id\",\n      \"$(KFP_POD_NAME)\",\n      \"--pipeline_root\",\n      pipeline_root,\n  ]\n\n  # Mount necessary environment variables.\n  op.apply(_default_transformers.add_kfp_pod_env)\n  op.container.add_env_variable(\n      k8s_client.V1EnvVar(name=\"KFP_V2_IMAGE\", value=op.container.image))\n\n  config_map_ref = k8s_client.V1ConfigMapEnvSource(\n      name='metadata-grpc-configmap', optional=True)\n  op.container.add_env_from(\n      k8s_client.V1EnvFromSource(config_map_ref=config_map_ref))\n\n  op.arguments = list(op.container_spec.command) + list(op.container_spec.args)\n\n  runtime_info = {\n      \"inputParameters\": collections.OrderedDict(),\n      \"inputArtifacts\": collections.OrderedDict(),\n      \"outputParameters\": collections.OrderedDict(),\n      \"outputArtifacts\": collections.OrderedDict(),\n  }\n\n  op.command += [\"--\"]\n  component_spec = op.component_spec\n  for parameter, spec in sorted(\n      component_spec.input_definitions.parameters.items()):\n    parameter_info = {\n        \"type\":\n            pipeline_spec_pb2.PrimitiveType.PrimitiveTypeEnum.Name(spec.type),\n    }\n    op.command += [f\"{parameter}={op._parameter_arguments[parameter]}\"]\n    runtime_info[\"inputParameters\"][parameter] = parameter_info\n  op.command += [\"--\"]\n\n  for artifact_name, spec in sorted(\n      component_spec.input_definitions.artifacts.items()):\n    artifact_info = {\n        \"metadataPath\": op.input_artifact_paths[artifact_name],\n        \"schemaTitle\": spec.artifact_type.schema_title,\n        \"instanceSchema\": spec.artifact_type.instance_schema,\n    }\n    runtime_info[\"inputArtifacts\"][artifact_name] = artifact_info\n\n  for parameter, spec in sorted(\n      component_spec.output_definitions.parameters.items()):\n    parameter_info = {\n        \"type\":\n            pipeline_spec_pb2.PrimitiveType.PrimitiveTypeEnum.Name(spec.type),\n        \"path\":\n            op.file_outputs[parameter],\n    }\n    runtime_info[\"outputParameters\"][parameter] = parameter_info\n\n  for artifact_name, spec in sorted(\n      component_spec.output_definitions.artifacts.items()):\n    # TODO: Assert instance_schema.\n    artifact_info = {\n        # Type used to register output artifacts.\n        \"schemaTitle\": spec.artifact_type.schema_title,\n        \"instanceSchema\": spec.artifact_type.instance_schema,\n        # File used to write out the registered artifact ID.\n        \"metadataPath\": op.file_outputs[artifact_name],\n    }\n    runtime_info[\"outputArtifacts\"][artifact_name] = artifact_info\n\n  op.container.add_env_variable(\n      k8s_client.V1EnvVar(name=\"KFP_V2_RUNTIME_INFO\",\n                          value=json.dumps(runtime_info)))\n\n  op.pod_annotations['pipelines.kubeflow.org/v2_component'] = \"true\"\n  op.pod_labels['pipelines.kubeflow.org/v2_component']= \"true\"\n", "path": "sdk/python/kfp/compiler/v2_compat.py"}]}