Recovery crash

[patkivikram]

Checklist

• I have included information about relevant versions
• I have verified that the issue persists when using the master branch of Faust.

Steps to reproduce

Recovery crashes on a rebalance

Expected behavior

no crash
Tell us what you expected to happen.

Actual behavior

[2020-11-20 20:25:29,427] [22] [ERROR] [^---Recovery]: Crashed reason=IllegalStateError("No current assignment for partition TopicPartition(topic='p999105-enriched-ad', partition=10)")
Traceback (most recent call last):
File "/usr/local/lib/python3.7/dist-packages/mode/services.py", line 802, in _execute_task
await task
File "/usr/local/lib/python3.7/dist-packages/faust/tables/recovery.py", line 349, in _restart_recovery
await T(self._resume_streams)()
File "/usr/local/lib/python3.7/dist-packages/faust/tables/recovery.py", line 297, in _resume_streams
consumer.perform_seek(), timeout=self.app.conf.broker_request_timeout
File "/usr/local/lib/python3.7/dist-packages/faust/tables/recovery.py", line 556, in _wait
wait_result = await self.wait_first(coro, signal, timeout=timeout)
File "/usr/local/lib/python3.7/dist-packages/mode/services.py", line 715, in wait_first
f.result() # propagate exceptions
File "/usr/local/lib/python3.7/dist-packages/faust/transport/consumer.py", line 534, in perform_seek
_committed_offsets = await self.seek_to_committed()
File "/usr/local/lib/python3.7/dist-packages/faust/transport/consumer.py", line 1315, in seek_to_committed
return await self._thread.seek_to_committed()
File "/usr/local/lib/python3.7/dist-packages/faust/transport/drivers/aiokafka.py", line 509, in seek_to_committed
return await self.call_thread(self._ensure_consumer().seek_to_committed)
File "/usr/local/lib/python3.7/dist-packages/mode/threads.py", line 436, in call_thread
result = await promise
File "/usr/local/lib/python3.7/dist-packages/mode/threads.py", line 383, in _process_enqueued
result = await maybe_async(method(*args, **kwargs))
File "/usr/local/lib/python3.7/dist-packages/mode/utils/futures.py", line 134, in maybe_async
return await res
File "/usr/local/lib/python3.7/dist-packages/aiokafka/consumer/consumer.py", line 868, in seek_to_committed
self._fetcher.seek_to(tp, offset)
File "/usr/local/lib/python3.7/dist-packages/aiokafka/consumer/fetcher.py", line 1162, in seek_to
self._subscriptions.seek(tp, offset)
File "/usr/local/lib/python3.7/dist-packages/aiokafka/consumer/subscription_state.py", line 239, in seek
self._assigned_state(tp).seek(offset)
File "/usr/local/lib/python3.7/dist-packages/aiokafka/consumer/subscription_state.py", line 120, in _assigned_state
"No current assignment for partition {}".format(tp))
kafka.errors.IllegalStateError: IllegalStateError: No current assignment for partition TopicPartition(topic='p999105-enriched-ad', partition=10)
[2020-11-20 20:25:29,436] [22] [WARNING] [^-App]: Missing sensor state for rebalance #13
[2020-11-20 20:25:29,439] [22] [INFO] [^Worker]: Stopping...
[2020-11-20 20:25:29,440] [22] [INFO] [^-App]: Stopping...
[2020-11-20 20:25:29,440] [22] [INFO] [^---Fetcher]: Stopping...
[2020-11-20 20:25:29,441] [22] [INFO] [^--Consumer]: Consumer shutting down for user cancel.
[2020-11-20 20:25:29,441] [22] [INFO] [^-App]: Flush producer buffer...
[2020-11-20 20:25:29,442] [22] [INFO] [^--TableManager]: Stopping...

Tell us what happened instead.

Full traceback

Paste the full traceback (if there is any)

Versions

• Python version 3.7
• Faust version 0.2.2
• Operating system
• Kafka version
• RocksDB version (if applicable)

[patkivikram]

This is fixed in 0.3.0