Offset committed for second message when first is still being processed

[bohdantan]

Checklist

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

Steps to reproduce

• Create new python 3.8 virtualenv
• Install Faust pip install git+https://github.com/robinhood/faust@b6ebdd955c1de5ceed81f53ade76b4b04561151e
• Create worker with agent (with 2 actors) and with task that adds messages to topic. Add monitor that prints offset commits. Save it to example.py

  
  import asyncio

import faust from faust.sensors.monitor import Monitor

class PrintCommitMonitor(Monitor): def on_tp_commit(self, tp_offsets): print(f'Commit offsets {tp_offsets}')

app = faust.App( 'test-app', broker='kafka://localhost:9092', value_serializer='raw', Monitor=PrintCommitMonitor, )

test_topic = app.topic('test')

@app.agent(test_topic, concurrency=2) async def process(stream): async for msg in stream: print(f'Start processing {msg} - by {stream.concurrency_index}') if msg == b'Message i=0': await asyncio.sleep(20) else: await asyncio.sleep(3) print(f'Finish processing {msg} - by {stream.concurrency_index}')

@app.task(on_leader=True) async def on_started(): print('Add messages') for i in range(10): msg = f'Message i={i}' await test_topic.send(key=msg, value=msg)

- Start worker `faust -A example worker -l info --without-web`. Worker will create kafka topic. Task will write 10 messages to topic
- Two actors will start processing messages. First actor will take message with offset 0, second actor will take message with offset 1
- After second actor will finish processing message with offset 1 (but before first actor will finish processing message with offset 0) offset 1 will be committed

## Expected behavior

Worker commits offset only after all messages with smaller offsets are processed.

## Actual behavior

Worker commits offset 1 after message with offset is processed by one actor, but message with offset 0 is still being processed by another actor.

# Versions

* Python version - 3.8.6
* Faust version - git+https://github.com/robinhood/faust@b6ebdd955c1de5ceed81f53ade76b4b04561151e
* Operating system - Ubuntu 20.04.1
* Kafka version - 2.4.1

[patkivikram]

The concurrency processes messages out of order. You wont see this behavior with concurrency=1

[ghost]

I was also curious about this behavior. I was expecting messages to be processed out of order, but offsets to be committed in order. The current design could clearly lead to data loss in the case of a failure and restart.

[zailaib]

if a service crash then restart; the tp offset was not the message witch made agent crash

[wsmlby]

this is a very important issue on faust's concurrency: even if the logic for the event process is completely build with out-of-order in mind, the out-of-order commit-offset will cause loss of data when consumer process is interrupted.