Merge pull request #22276 Don't try to parse non-flags as retained pipeline options.

Author: robertwb

sdks/python/apache_beam/options/pipeline_options.py

@@ -320,6 +320,9 @@ def get_all_options(
       while i < len(unknown_args):
         # Treat all unary flags as booleans, and all binary argument values as
         # strings.
+        if not unknown_args[i].startswith('-'):
+          i += 1
+          continue
         if i + 1 >= len(unknown_args) or unknown_args[i + 1].startswith('-'):
           split = unknown_args[i].split('=', 1)
           if len(split) == 1:
@@ -330,7 +333,7 @@ def get_all_options(
         else:
           parser.add_argument(unknown_args[i], type=str)
           i += 2
-      parsed_args = parser.parse_args(self._flags)
+      parsed_args, _ = parser.parse_known_args(self._flags)
     else:
       if unknown_args:
         _LOGGER.warning("Discarding unparseable args: %s", unknown_args)

sdks/python/apache_beam/options/pipeline_options_test.py

@@ -363,11 +363,6 @@ def test_retain_unknown_options_unary_single_dash_store_true(self):
     result = options.get_all_options(retain_unknown_options=True)
     self.assertEqual(result['i'], True)
 
-  def test_retain_unknown_options_unary_missing_prefix(self):
-    options = PipelineOptions(['bad_option'])
-    with self.assertRaises(SystemExit):
-      options.get_all_options(retain_unknown_options=True)
-
   def test_override_options(self):
     base_flags = ['--num_workers', '5']
     options = PipelineOptions(base_flags)