[BEAM-8730] Chicago taxi example on Beam on Flink sometimes fails with TypeError: 'NoneType' object is not subscriptable

• Type: Bug
• Status: Resolved
• Priority: P2
• Resolution: Duplicate
• Fix Version/s: Not applicable
• Component/s: sdk-py-core
• Assignee: Unassigned
• Reporter: Valentyn Tymofieiev
• Created: 15/Nov/19 22:54
• Updated: 24/Jul/20 19:55
• Resolved: 20/Nov/19 18:05

Description

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.

Issue Links

• is duplicated by: BEAM-8651 Python 3 portable pipelines sometimes fail with errors in StockUnpickler.find_class()