diff --git a/airflow/providers/apache/beam/operators/beam.py b/airflow/providers/apache/beam/operators/beam.py
index bf75f3caa74fb..9dc4c36b3efc8 100644
--- a/airflow/providers/apache/beam/operators/beam.py
+++ b/airflow/providers/apache/beam/operators/beam.py
@@ -333,6 +333,7 @@ def __init__(
         self.py_interpreter = py_interpreter
         self.py_requirements = py_requirements
         self.py_system_site_packages = py_system_site_packages
+        self.pipeline_options = copy.deepcopy(self.pipeline_options)
         self.pipeline_options.setdefault("labels", {}).update(
             {"airflow-version": "v" + version.replace(".", "-").replace("+", "-")}
         )
@@ -740,7 +741,7 @@ def __init__(
         self.go_file = go_file
         self.launcher_binary = launcher_binary
         self.worker_binary = worker_binary or launcher_binary
-
+        self.pipeline_options = copy.deepcopy(self.pipeline_options)
         self.pipeline_options.setdefault("labels", {}).update(
             {"airflow-version": "v" + version.replace(".", "-").replace("+", "-")}
         )
diff --git a/tests/providers/apache/beam/operators/test_beam.py b/tests/providers/apache/beam/operators/test_beam.py
index 538c2417a0e38..ca4e7558e06f6 100644
--- a/tests/providers/apache/beam/operators/test_beam.py
+++ b/tests/providers/apache/beam/operators/test_beam.py
@@ -60,6 +60,7 @@
     "output": "gs://test/output",
     "labels": {"foo": "bar", "airflow-version": TEST_VERSION},
 }
+
 TEST_IMPERSONATION_ACCOUNT = "test@impersonation.com"
 BEAM_OPERATOR_PATH = "airflow.providers.apache.beam.operators.beam.{}"
 
@@ -286,7 +287,7 @@ def test_exec_dataflow_runner(self, gcs_hook, dataflow_hook_mock, beam_hook_mock
             "jobName": job_name,
             "stagingLocation": "gs://test/staging",
             "region": "us-central1",
-            "labels": {"foo": "bar", "airflow-version": TEST_VERSION},
+            "labels": {"foo": "bar"},
             "output": "gs://test/output",
             "impersonateServiceAccount": TEST_IMPERSONATION_ACCOUNT,
         }
@@ -926,7 +927,7 @@ def test_exec_dataflow_runner(self, gcs_hook, dataflow_hook_mock, beam_hook_mock
             "job_name": job_name,
             "staging_location": "gs://test/staging",
             "output": "gs://test/output",
-            "labels": {"foo": "bar", "airflow-version": TEST_VERSION},
+            "labels": {"foo": "bar"},
             "region": "us-central1",
             "impersonate_service_account": TEST_IMPERSONATION_ACCOUNT,
         }