Я хочу создать топи c, которая автоматически сжимается kafka, используя поток faust. Я использую код как:
import asyncio
import random
import faust
app = faust.App(
'mysample',
broker=<...>,
value_serializer='raw',
key_serializer='raw',
topic_replication_factor=3,
)
topic = app.topic('mytest5', partitions=1, compacting=True, acks=False)
count = 1
@app.timer(2.0, on_leader=True)
async def publish_greetings():
global count
print('PUBLISHING ON LEADER!')
await topic.send(key=str(count), value=str(random.random()))
count += 1
@app.agent(topic)
async def say(greetings):
async for k, v in greetings.items():
print(f'{k}, {v}')
Я пробовал также опции internal
, retention
, deleting
с созданием topi c и каждый раз, когда я менял имя topi c, так новый топи c должен был быть создан. Но я не вижу, что кафка на самом деле сжимает топи c. Каждый раз, когда я читаю топи c, я вижу полную историю событий.
Как создать компактную топи c с faust?