FnApiRunner KeyError

Code to recreate (modified slightly from https://lists.apache.org/thread.html/e910bfe702a3c8c5b0902f5f1c2c51fb7b2574f1b4abc4d9efab4e0f@%3Cdev.beam.apache.org%3E):

import apache_beam as beam
import argparse
from apache_beam import transforms
from apache_beam import pvalue
from apache_beam.options import pipeline_options
import logging

def _copy_number(number, side=None):
print '_copy_number:', number, side
yield number

def fn_sum(values):
#print 'values', values
return sum(values)

def run(argv=None):
parser = argparse.ArgumentParser()
_, pipeline_args = parser.parse_known_args(argv)
options = pipeline_options.PipelineOptions(pipeline_args)
#options.view_as(pipeline_options.StandardOptions).streaming = True
numbers = [1, 2]
with beam.Pipeline(options=options) as p:
sum_1 = (p

sum_2 = (p

_ = ((sum_1, sum_2)

logging.getLogger().setLevel(logging.INFO)
run()

Console:
$ python test.py
INFO:root:Missing pipeline option (runner). Executing pipeline using the default runner: DirectRunner.
INFO:root:==================== <function annotate_downstream_side_inputs at 0x7f2b7088e758> ====================
INFO:root:==================== <function fix_side_input_pcoll_coders at 0x7f2b7088e7d0> ====================
INFO:root:==================== <function lift_combiners at 0x7f2b7088e5f0> ====================
INFO:root:==================== <function expand_gbk at 0x7f2b7088e668> ====================
INFO:root:==================== <function sink_flattens at 0x7f2b7088e6e0> ====================
INFO:root:==================== <function greedily_fuse at 0x7f2b7088e848> ====================
INFO:root:==================== <function impulse_to_input at 0x7f2b7088e578> ====================
INFO:root:==================== <function inject_timer_pcollections at 0x7f2b7088e8c0> ====================
INFO:root:==================== <function sort_stages at 0x7f2b7088e938> ====================
INFO:root:Running ((ref_AppliedPTransform_ReadNumber1/Read_3)((ref_AppliedPTransform_CalculateSum1/KeyWithVoid_5)(CalculateSum1/CombinePerKey/Precombine)))+(CalculateSum1/CombinePerKey/Group/Write)
INFO:root:Running ((CalculateSum1/CombinePerKey/Group/Read)(CalculateSum1/CombinePerKey/Merge))((CalculateSum1/CombinePerKey/ExtractOutputs)((ref_AppliedPTransform_CalculateSum1/UnKey_13)(ref_PCollection_PCollection_7/Write)))
INFO:root:Running ((ref_AppliedPTransform_CalculateSum1/DoOnce/Read_15)(((ref_AppliedPTransform_CalculateSum1/InjectDefault_16)(ref_PCollection_PCollection_9/Write))(Flatten/Transcode/0)))(Flatten/Write/0)
INFO:root:Running ((ref_AppliedPTransform_ReadNumber2/Read_18)((ref_AppliedPTransform_ParDo(CallableWrapperDoFn)_19)((ref_AppliedPTransform_CalculateSum2/KeyWithVoid_21)(CalculateSum2/CombinePerKey/Precombine))))(CalculateSum2/CombinePerKey/Group/Write)
Traceback (most recent call last):
File "test.py", line 41, in <module>
run()
File "test.py", line 38, in run