RocksDB recovery crashes with KeyError

[vytas-screen9]

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

We created a simple test for Faust streaming and ran two or more parallel workers, each using own RocksDB.

Under high traffic, stopping one worker sometimes triggers crashes in the other one.

Expected behavior

Workers can be stopped, and restarted, and even in the unfortunate even they crash, it should not impact other workers operating in parallel.

Actual behavior

Other workers may crash.

Tell us what happened instead.

Full traceback

[^---Recovery]: Crashed reason=KeyError(TP(topic='<our-topic>', partition=1)) 

Traceback (most recent call last):
  File "<our venv>/lib/python3.8/site-packages/mode/services.py", line 802, in _execute_task
    await task
  File "<our venv>/lib/python3.8/site-packages/faust/tables/recovery.py", line 375, in _restart_recovery
    await self._wait(
  File "<our venv>/lib/python3.8/site-packages/faust/tables/recovery.py", line 562, in _wait
    wait_result = await self.wait_first(coro, signal, timeout=timeout)
  File "<our venv>/lib/python3.8/site-packages/mode/services.py", line 715, in wait_first
    f.result()  # propagate exceptions
  File "<our venv>/lib/python3.8/site-packages/faust/tables/recovery.py", line 666, in _build_offsets
    new_value = earliest[tp]
KeyError: TP(topic='<our topic>', partition=1)

Versions

• CPython 3.8.5
• Faust faust-streaming @ git+https://github.com/faust-streaming/faust@fcb6b185105239329fd21e4af0f23e5244a2c8eb
• Ubuntu 20.04.1 LTS
• Kafka 2.6.0
• python-rocksdb 0.7.0

[patkivikram]

Can you verify if this is fixed in 0.4.7?