[
https://issues.apache.org/jira/browse/BEAM-5442?focusedWorklogId=168875&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-168875
]
ASF GitHub Bot logged work on BEAM-5442:
----------------------------------------
Author: ASF GitHub Bot
Created on: 22/Nov/18 17:30
Start Date: 22/Nov/18 17:30
Worklog Time Spent: 10m
Work Description: chamikaramj closed pull request #7117: [DO NOT MERGE]
Revert "[BEAM-5442] Revert #6675 "Revert PRs #6557 #6589 #6600""
URL: https://github.com/apache/beam/pull/7117
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/sdks/python/apache_beam/options/pipeline_options.py
b/sdks/python/apache_beam/options/pipeline_options.py
index 4cd4f014ac0..39701ddaea0 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -20,10 +20,8 @@
from __future__ import absolute_import
import argparse
-import logging
from builtins import list
from builtins import object
-from collections import OrderedDict
from apache_beam.options.value_provider import RuntimeValueProvider
from apache_beam.options.value_provider import StaticValueProvider
@@ -215,35 +213,7 @@ def get_all_options(self, drop_default=False):
subset[str(cls)] = cls
for cls in subset.values():
cls._add_argparse_args(parser) # pylint: disable=protected-access
- known_args, unknown_args = parser.parse_known_args(self._flags)
- # Parse args which are not known at this point but might be recognized
- # at a later point in time, i.e. by the actual Runner.
- if unknown_args and unknown_args[0] != '':
- logging.info("Parsing unknown args: %s", unknown_args)
-
- def enumerate_args(args):
- cleaned_args = OrderedDict()
- for arg in args:
- if arg.startswith('--'):
- # split argument name if it's in arg_name=value syntax
- arg_name = arg.split('=', 1)[0]
- # count identical arg names
- if arg_name not in cleaned_args:
- cleaned_args[arg_name] = 1
- else:
- cleaned_args[arg_name] += 1
- return cleaned_args
-
- for arg_name, num_times in enumerate_args(unknown_args).items():
- parser.add_argument(arg_name,
- nargs='?',
- action='append' if num_times > 1 else 'store')
-
- # repeat parsing with unknown options added
- known_args, unknown_args = parser.parse_known_args(self._flags)
- if unknown_args:
- logging.warning("Discarding unparseable args: %s", unknown_args)
-
+ known_args, _ = parser.parse_known_args(self._flags)
result = vars(known_args)
# Apply the overrides if any
diff --git a/sdks/python/apache_beam/options/pipeline_options_test.py
b/sdks/python/apache_beam/options/pipeline_options_test.py
index 65997536370..9c14c25668e 100644
--- a/sdks/python/apache_beam/options/pipeline_options_test.py
+++ b/sdks/python/apache_beam/options/pipeline_options_test.py
@@ -45,24 +45,16 @@ def tearDown(self):
'display_data': [DisplayDataItemMatcher('num_workers', 5)]},
{
'flags': [
- '--profile_cpu', '--profile_location', 'gs://bucket/',
- 'ignored', '-invalid=arg', '--unknown_arg', 'unknown_value',
- '--unknown_flag'
- ],
+ '--profile_cpu', '--profile_location', 'gs://bucket/',
'ignored'],
'expected': {
'profile_cpu': True, 'profile_location': 'gs://bucket/',
'mock_flag': False, 'mock_option': None,
- 'mock_multi_option': None,
- 'unknown_arg': 'unknown_value',
- 'unknown_flag': None},
+ 'mock_multi_option': None},
'display_data': [
DisplayDataItemMatcher('profile_cpu',
True),
DisplayDataItemMatcher('profile_location',
- 'gs://bucket/'),
- DisplayDataItemMatcher('unknown_arg',
- 'unknown_value')
- ]
+ 'gs://bucket/')]
},
{'flags': ['--num_workers', '5', '--mock_flag'],
'expected': {'num_workers': 5,
@@ -289,15 +281,6 @@ def _add_argparse_args(cls, parser):
with self.assertRaises(RuntimeError):
options.pot_non_vp_arg1.get()
- # Converts duplicate unknown argument values to a single argument
- # with a list value.
- def test_unknown_duplicate_args_converted_to_list(self):
- options = PipelineOptions(['--dup_arg', 'val1',
- '--dup_arg', 'val2',
- '--dup_arg=val3'])
- self.assertEqual(options.get_all_options()['dup_arg'],
- ['val1', 'val2', 'val3'])
-
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
diff --git a/sdks/python/apache_beam/runners/worker/sdk_worker_main_test.py
b/sdks/python/apache_beam/runners/worker/sdk_worker_main_test.py
index e9b584a339f..6b5972e4ac4 100644
--- a/sdks/python/apache_beam/runners/worker/sdk_worker_main_test.py
+++ b/sdks/python/apache_beam/runners/worker/sdk_worker_main_test.py
@@ -56,8 +56,8 @@ def wrapped_method_for_test():
def test_work_count_default_value(self):
self._check_worker_count('{}', 12)
- def test_parse_pipeline_options(self):
- expected_options = PipelineOptions([])
+ def test_parse_pipeine_options(self):
+ expected_options = PipelineOptions()
expected_options.view_as(
SdkWorkerMainTest.MockOptions).m_m_option = [
'worker_threads=1', 'beam_fn_api'
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 168875)
Time Spent: 10h 40m (was: 10.5h)
> PortableRunner swallows custom options for Runner
> -------------------------------------------------
>
> Key: BEAM-5442
> URL: https://issues.apache.org/jira/browse/BEAM-5442
> Project: Beam
> Issue Type: Bug
> Components: sdk-java-core, sdk-py-core
> Reporter: Maximilian Michels
> Assignee: Maximilian Michels
> Priority: Major
> Labels: portability, portability-flink
> Fix For: 2.10.0
>
> Time Spent: 10h 40m
> Remaining Estimate: 0h
>
> The PortableRunner doesn't pass custom PipelineOptions to the executing
> Runner.
> Example: {{--parallelism=4}} won't be forwarded to the FlinkRunner.
> (The option is just removed during proto translation without any warning)
> We should allow some form of customization through the options, even for the
> PortableRunner.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
