Deep copy PipelineOptions and keep the input intact. (#34723)

* Deep copy PipelineOptions and keep the input intact.

* Fix failed tests for FlinkRunner.

Author: shunping

sdks/python/apache_beam/pipeline.py

@@ -49,6 +49,7 @@
 
 import abc
 import contextlib
+import copy
 import logging
 import os
 import re
@@ -171,7 +172,9 @@ def __init__(
 
     if options is not None:
       if isinstance(options, PipelineOptions):
-        self._options = options
+        # Make a deep copy of options since they could be overwritten in later
+        # steps.
+        self._options = copy.deepcopy(options)
       else:
         raise ValueError(
             'Parameter options, if specified, must be of type PipelineOptions. '

sdks/python/apache_beam/runners/interactive/interactive_beam_test.py

@@ -395,7 +395,12 @@ def test_cleanup_by_a_pipeline(self):
     # Pipeline association is cleaned up.
     self.assertNotIn(p, self.clusters.pipelines)
     self.assertNotIn(p, dcm.pipelines)
-    self.assertEqual(options.view_as(FlinkRunnerOptions).flink_master, '[auto]')
+    # The internal option in the pipeline is overwritten.
+    self.assertEqual(
+        p.options.view_as(FlinkRunnerOptions).flink_master, '[auto]')
+    # The original option is unchanged.
+    self.assertEqual(
+        options.view_as(FlinkRunnerOptions).flink_master, meta.master_url)
     # The cluster is unknown now.
     self.assertNotIn(meta, self.clusters.dataproc_cluster_managers)
     self.assertNotIn(meta.master_url, self.clusters.master_urls)
@@ -423,10 +428,17 @@ def test_not_cleanup_if_multiple_pipelines_share_a_manager(self):
     # Pipeline association of p is cleaned up.
     self.assertNotIn(p, self.clusters.pipelines)
     self.assertNotIn(p, dcm.pipelines)
-    self.assertEqual(options.view_as(FlinkRunnerOptions).flink_master, '[auto]')
+    # The internal option in the pipeline is overwritten.
+    self.assertEqual(
+        p.options.view_as(FlinkRunnerOptions).flink_master, '[auto]')
+    # The original option is unchanged.
+    self.assertEqual(
+        options.view_as(FlinkRunnerOptions).flink_master, meta.master_url)
     # Pipeline association of p2 still presents.
     self.assertIn(p2, self.clusters.pipelines)
     self.assertIn(p2, dcm.pipelines)
+    self.assertEqual(
+        p2.options.view_as(FlinkRunnerOptions).flink_master, meta.master_url)
     self.assertEqual(
         options2.view_as(FlinkRunnerOptions).flink_master, meta.master_url)
     # The cluster is still known.