Exception thrown on faust-streaming/faust but not on robinhood/faust

[alxdb]

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

I have the same single module app that works correctly with the robinhood version of faust, but does not work with this fork. Namely, the exception quoted in the title on_rebalance() takes 5 positional arguments but 6 were given is thrown on app startup in when faust-streaming but not faust. relevant versions used are 0.4.8 466dbf2 (git master at time of writing) and 1.10.4. The single module app, and exception are as follows:

from typing import Any

import faust


class UserUpdate(faust.Record):
    user_id: int
    event_type: str
    event_data: Any


app = faust.App("faust-demo", broker="kafka://kafka-1:9092")
user_updates = app.topic("user_updates", value_type=UserUpdate, partitions=8)
users = app.Table("user", partitions=8)


@app.agent(user_updates)
async def process_user_updates(updates: faust.StreamT[UserUpdate]):
    async for update in updates:
        user = users.get(update.user_id)
        if user:
            user[update.event_type] = update.event_data
            users[update.user_id] = user
        else:
            users[update.user_id] = {update.event_type: update.event_data}


@app.page("/users")
async def get_users(self, request):
    return self.json(users)

Expected behavior

Exception is not thrown

Actual behavior

Exception is thrown

Full traceback

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/faust/app/base.py", line 1720, in _on_partitions_assigned
    await T(self.tables.on_rebalance)(
  File "/usr/local/lib/python3.9/site-packages/faust/utils/tracing.py", line 133, in corowrapped
    await_ret = await ret
  File "/usr/local/lib/python3.9/site-packages/faust/tables/manager.py", line 194, in on_rebalance
    await T(table.on_rebalance)(
  File "/usr/local/lib/python3.9/site-packages/faust/utils/tracing.py", line 133, in corowrapped
    await_ret = await ret
  File "/usr/local/lib/python3.9/site-packages/faust/tables/base.py", line 571, in on_rebalance
    await self.data.on_rebalance(
TypeError: on_rebalance() takes 5 positional arguments but 6 were given

Versions

• python: 3.9.1
• faust: 0.4.8 466dbf2 (master)
• os: debian (python:3.9 docker image)
• kafka: 2.7.0

[patkivikram]

I am unable to reproduce this on the latest master. Can you double check?

[alxdb]

I have double checked, at first i thought it might be due to using multiple brokers, but that appears not to be the case. I've created a repo that should be able to recreate the issue: faust-demo/faust-streaming. if you have docker/docker-compose installed could you try it out and see if it reproduces the issue?