Chicago taxi example on Beam on Flink sometimes fails with TypeError: 'NoneType' object is not subscriptable

Sample stack trace:

```
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py", line 158, in _execute
response = task()
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py", line 191, in <lambda>
self._execute(lambda: worker.do_instruction(work), work)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py", line 343, in do_instruction
request.instruction_id)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py", line 363, in process_bundle
instruction_id, request.process_bundle_descriptor_reference)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py", line 306, in get
self.data_channel_factory)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 578, in _init_
self.ops = self.create_execution_tree(self.process_bundle_descriptor)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 622, in create_execution_tree
descriptor.transforms, key=topological_height, reverse=True)])
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 621, in <listcomp>
for transform_id in sorted(
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 546, in wrapper
result = cache[args] = func(*args)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 605, in get_operation
in descriptor.transforms[transform_id].outputs.items()
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 604, in <dictcomp>
for tag, pcoll_id
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 603, in <listcomp>
tag: [get_operation(op) for op in pcoll_consumers[pcoll_id]]
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 546, in wrapper
result = cache[args] = func(*args)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 605, in get_operation
in descriptor.transforms[transform_id].outputs.items()
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 604, in <dictcomp>
for tag, pcoll_id
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 603, in <listcomp>
tag: [get_operation(op) for op in pcoll_consumers[pcoll_id]]
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 546, in wrapper
result = cache[args] = func(*args)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 605, in get_operation
in descriptor.transforms[transform_id].outputs.items()
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 604, in <dictcomp>
for tag, pcoll_id
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 603, in <listcomp>
tag: [get_operation(op) for op in pcoll_consumers[pcoll_id]]
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 546, in wrapper
result = cache[args] = func(*args)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 605, in get_operation
in descriptor.transforms[transform_id].outputs.items()
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 604, in <dictcomp>
for tag, pcoll_id
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 603, in <listcomp>
tag: [get_operation(op) for op in pcoll_consumers[pcoll_id]]
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 546, in wrapper
result = cache[args] = func(*args)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 605, in get_operation
in descriptor.transforms[transform_id].outputs.items()
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 604, in <dictcomp>
for tag, pcoll_id
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 603, in <listcomp>
tag: [get_operation(op) for op in pcoll_consumers[pcoll_id]]
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 546, in wrapper
result = cache[args] = func(*args)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 605, in get_operation
in descriptor.transforms[transform_id].outputs.items()
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 604, in <dictcomp>
for tag, pcoll_id
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 603, in <listcomp>
tag: [get_operation(op) for op in pcoll_consumers[pcoll_id]]
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 546, in wrapper
result = cache[args] = func(*args)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 605, in get_operation
in descriptor.transforms[transform_id].outputs.items()
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 604, in <dictcomp>
for tag, pcoll_id
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 603, in <listcomp>
tag: [get_operation(op) for op in pcoll_consumers[pcoll_id]]
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 546, in wrapper
result = cache[args] = func(*args)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 605, in get_operation
in descriptor.transforms[transform_id].outputs.items()
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 604, in <dictcomp>
for tag, pcoll_id
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 603, in <listcomp>
tag: [get_operation(op) for op in pcoll_consumers[pcoll_id]]
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 546, in wrapper
result = cache[args] = func(*args)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 605, in get_operation
in descriptor.transforms[transform_id].outputs.items()
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 604, in <dictcomp>
for tag, pcoll_id
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 603, in <listcomp>
tag: [get_operation(op) for op in pcoll_consumers[pcoll_id]]
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 546, in wrapper
result = cache[args] = func(*args)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 608, in get_operation
transform_id, transform_consumers)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 867, in create_operation
return creator(self, transform_id, transform_proto, payload, consumers)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1110, in create
serialized_fn, parameter)
File "/usr/local/google/home/valentyn/tmp/tfx_py37_master/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1149, in _create_pardo_operation
if not dofn_data[-1]:
TypeError: 'NoneType' object is not subscriptable
[while running 'Run[Transform]']
```

Possibly shares the rootcause with: https://issues.apache.org/jira/browse/BEAM-8651.