Ignore undeclared side outputs of DoFns in cloud executor

DoFns may emit undeclared side outputs, executor is crashing on those
instead they should be ignored.

----Release Notes----
[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=123371414

Author: aaltay

google/cloud/dataflow/transforms/core.py

@@ -575,7 +575,8 @@ def with_outputs(self, *tags, **main_kw):
     the available tags (e.g., for tag in o: ...).
 
     Args:
-      *tags: if non-empty, list of valid tags
+      *tags: if non-empty, list of valid tags. If a list of valid tags is given,
+        it will be an error to use an undeclared tag later in the pipeline.
       **main_kw: dictionary empty or with one key 'main' defining the tag to be
         used for the main output (which will not have a tag associated with it).
 

google/cloud/dataflow/worker/executor.py

@@ -388,6 +388,19 @@ def process(self, o):
     self.receivers[0].update_counters_finish()
 
 
+class _TaggedReceivers(dict):
+
+  class NullReceiver(object):
+
+    def output(self, element):
+      pass
+
+  def __missing__(self, unused_key):
+    if not getattr(self, '_null_receiver', None):
+      self._null_receiver = _TaggedReceivers.NullReceiver()
+    return self._null_receiver
+
+
 class DoOperation(Operation):
   """A Do operation that will execute a custom DoFn for each input element."""
 
@@ -473,7 +486,8 @@ def start(self):
     # Tag to output index map used to dispatch the side output values emitted
     # by the DoFn function to the appropriate receivers. The main output is
     # tagged with None and is associated with its corresponding index.
-    tagged_receivers = {}
+    tagged_receivers = _TaggedReceivers()
+
     output_tag_prefix = PropertyNames.OUT + '_'
     for index, tag in enumerate(self.spec.output_tags):
       if tag == PropertyNames.OUT:

google/cloud/dataflow/worker/executor_test.py

@@ -87,6 +87,14 @@ def finish_bundle(self, context, *args, **kwargs):
       f.write('finish called.')
 
 
+class DoFnUsingWithUndeclaredSideOutput(ptransform.DoFn):
+  """A DoFn class that emits an undeclared side output."""
+
+  def process(self, context, *args, **kwargs):
+    yield pvalue.SideOutputValue('undeclared', context.element)
+    yield context.element
+
+
 class ProgressRequestRecordingInMemoryReader(inmemory.InMemoryReader):
 
   def __init__(self, source):
@@ -175,6 +183,31 @@ def test_read_do_write_with_start_bundle(self):
     with open(finish_path) as f:
       self.assertEqual('finish called.', f.read())
 
+  def test_read_do_write_with_undeclared_output(self):
+    input_path = self.create_temp_file('01234567890123456789\n0123456789')
+    output_path = '%s.out' % input_path
+    work_item = workitem.BatchWorkItem(None)
+    work_item.map_task = make_map_task([
+        maptask.WorkerRead(
+            fileio.TextFileSource(file_path=input_path,
+                                  start_offset=0,
+                                  end_offset=15,
+                                  strip_trailing_newlines=True,
+                                  coder=coders.StrUtf8Coder()),
+            output_coders=[self.OUTPUT_CODER]),
+        maptask.WorkerDoFn(serialized_fn=pickle_with_side_inputs(
+            DoFnUsingWithUndeclaredSideOutput()),
+                           output_tags=['out'],
+                           output_coders=[self.OUTPUT_CODER],
+                           input=(0, 0),
+                           side_inputs=None),
+        make_text_sink(output_path, input=(1, 0))
+    ])
+
+    executor.MapTaskExecutor(work_item.map_task).execute()
+    with open(output_path) as f:
+      self.assertEqual('01234567890123456789\n', f.read())
+
   def test_read_do_shuffle_write(self):
     input_path = self.create_temp_file('a\nb\nc\nd\n')
     work_spec = [