Messages are not submitted into topic with cleanup policy set to "compact"

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

Hello. When i create topic with cleanup policy set to "compact" (in confluent cloud) and start producing messages into them i get an error:

 FutureMessage exception was never retrieved
future: <FutureMessage finished exception=UnknownError()> 
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/faust/topics.py", line 442, in _on_published
    res: RecordMetadata = fut.result()
kafka.errors.UnknownError: [Error -1] UnknownError

setting the topic to "delete" policy resolves the issue. this is not how i expect faust to work :))))

Expected behavior

i believe faust should not care about compaction policy.

Actual behavior

none of the messages actually appear in the topic - it's empty exception thrown output: http://joxi.ru/4AkQejZTkePK5m

Replicate

test code to replicate the issue:

test_topic = app.topic("test", value_serializer='json')
@app.timer(interval=1.0)
async def test_producer():
    await test_topic.send(
        value="{}",
        key=None,
    )
    print(f"item created")

Versions

• Python version: 3.7.2
• Faust version: Faust 1.10.4
• Operating system: linux
• Kafka version: confluent cloud
• RocksDB version (if applicable)