v0.4.2 regression breaks channels

[xgamer4]

Checklist

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

It looks like the regression was added in faust/agents/agent.py, specifically this chunk of code:

        if cur_version >= req_version:
            task = asyncio.Task(
                self._execute_actor(coro, aref),
                loop=self.loop,
                name=f"{str(aref)}-{self.channel.get_topic_name()}",
            )
        else:
            task = asyncio.Task(
                self._execute_actor(coro, aref),
                loop=self.loop,
            )

Steps to reproduce

Technically nothing - yesterday pytest testing code I was writing was running fine, today that same code errored out on every test I was writing - even previously passing tests. I pulled it out an created a minimal set of code to reproduce the error here (agent.test_context() returns a Channel which is how I noticed it):

import faust

app = faust.App('faust-bug-reproduction', broker='kafka://broker:29092')

topic = app.channel()
topic.maybe_declare()
@app.agent(topic)
async def split_sentence(data):
    async for key, value in data.items():
        print(key)
        print(value)

app.k = 1
app.v = 1

# Sends a message to the topic on a timer
@app.timer(interval=2.0)
async def example_sender(app):
    await topic.send(
        value=app.v,
        key=app.k
    )
    app.k *= 2
    app.v *= 3


# When ran standalone, start it up
if __name__ == '__main__':
    app.main()

Expected behavior

Just a bunch of repeating messages of the form:

[2021-01-21 17:38:42,774] [1] [WARNING] 1 
[2021-01-21 17:38:42,774] [1] [WARNING] 1 
[2021-01-21 17:38:44,774] [1] [WARNING] 2 
[2021-01-21 17:38:44,774] [1] [WARNING] 3 
[2021-01-21 17:38:46,774] [1] [WARNING] 4 
[2021-01-21 17:38:46,774] [1] [WARNING] 9 

Actual behavior

Crash on name=f"{str(aref)}-{self.channel.get_topic_name()}. Error:NotImplemented Error('Channels are unnamed topics')

Full traceback

name=f"{str(aref)}-{self.channel.get_topic_name()}",
RuntimeWarning: Enable tracemalloc to get the object allocation traceback      ◣[2021-01-21 17:34:54,688] [1] [ERROR] [^Worker]: Error: NotImplementedError('Channels are unnamed topics') 
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/mode/worker.py", line 276, in execute_from_commandline
    self.loop.run_until_complete(self._starting_fut)
  File "/usr/local/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
  File "/usr/local/lib/python3.8/site-packages/mode/services.py", line 759, in start
    await self._default_start()
  File "/usr/local/lib/python3.8/site-packages/mode/services.py", line 766, in _default_start
    await self._actually_start()
  File "/usr/local/lib/python3.8/site-packages/mode/services.py", line 790, in _actually_start
    await child.maybe_start()
  File "/usr/local/lib/python3.8/site-packages/mode/services.py", line 818, in maybe_start
    await self.start()
  File "/usr/local/lib/python3.8/site-packages/mode/services.py", line 759, in start
    await self._default_start()
  File "/usr/local/lib/python3.8/site-packages/mode/services.py", line 766, in _default_start
    await self._actually_start()
  File "/usr/local/lib/python3.8/site-packages/mode/services.py", line 790, in _actually_start
    await child.maybe_start()
  File "/usr/local/lib/python3.8/site-packages/mode/services.py", line 818, in maybe_start
    await self.start()
  File "/usr/local/lib/python3.8/site-packages/mode/services.py", line 759, in start
    await self._default_start()
  File "/usr/local/lib/python3.8/site-packages/mode/services.py", line 766, in _default_start
    await self._actually_start()
  File "/usr/local/lib/python3.8/site-packages/mode/services.py", line 783, in _actually_start
    await self.on_start()
  File "/usr/local/lib/python3.8/site-packages/faust/agents/manager.py", line 57, in on_start
    await agent.maybe_start()
  File "/usr/local/lib/python3.8/site-packages/mode/services.py", line 818, in maybe_start
    await self.start()
  File "/usr/local/lib/python3.8/site-packages/mode/services.py", line 759, in start
    await self._default_start()
  File "/usr/local/lib/python3.8/site-packages/mode/services.py", line 766, in _default_start
    await self._actually_start()
  File "/usr/local/lib/python3.8/site-packages/mode/services.py", line 783, in _actually_start
    await self.on_start()
  File "/usr/local/lib/python3.8/site-packages/faust/agents/agent.py", line 284, in on_start
    await self._on_start_supervisor()
  File "/usr/local/lib/python3.8/site-packages/faust/agents/agent.py", line 315, in _on_start_supervisor
    res = await self._start_one(
  File "/usr/local/lib/python3.8/site-packages/faust/agents/agent.py", line 253, in _start_one
    return await self._start_task(
  File "/usr/local/lib/python3.8/site-packages/faust/agents/agent.py", line 648, in _start_task
    return await self._prepare_actor(
  File "/usr/local/lib/python3.8/site-packages/faust/agents/agent.py", line 668, in _prepare_actor
    name=f"{str(aref)}-{self.channel.get_topic_name()}",
  File "/usr/local/lib/python3.8/site-packages/faust/channels.py", line 197, in get_topic_name
    raise NotImplementedError("Channels are unnamed topics")
NotImplementedError: Channels are unnamed topics                    

Versions

• Python version = 3.8
• Faust version = 0.4.2
• Operating system = whatever python3.8-slim docker uses
• Kafka version

[bobh66]

This shows up in Python3.8 unit tests but works fine in 3.6 and 3.7:

_________________________________________________________________________________ Test_Stream.test_aiter_tracked__CancelledError __________________________________________________________________________________
self = <tests.unit.test_streams.Test_Stream object at 0x7f541498b370>
    @pytest.mark.asyncio
    @pytest.mark.allow_lingering_tasks(count=1)
    async def test_aiter_tracked__CancelledError(self, *, stream, app):
>       event = await self._assert_stream_aiter(
            stream,
            side_effect=asyncio.CancelledError(),
            raises=asyncio.CancelledError,
        )
tests/unit/test_streams.py:143:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
tests/unit/test_streams.py:186: in _assert_stream_aiter
    async with agent as _agent:
.tox/3.8/lib/python3.8/site-packages/mode/services.py:120: in __aenter__
    await self.start()
.tox/3.8/lib/python3.8/site-packages/mode/services.py:759: in start
    await self._default_start()
.tox/3.8/lib/python3.8/site-packages/mode/services.py:766: in _default_start
    await self._actually_start()
.tox/3.8/lib/python3.8/site-packages/mode/services.py:783: in _actually_start
    await self.on_start()
faust/agents/agent.py:284: in on_start
    await self._on_start_supervisor()
faust/agents/agent.py:315: in _on_start_supervisor
    res = await self._start_one(
faust/agents/agent.py:253: in _start_one
    return await self._start_task(
faust/agents/agent.py:648: in _start_task
    return await self._prepare_actor(
faust/agents/agent.py:668: in _prepare_actor
    name=f"{str(aref)}-{self.channel.get_topic_name()}",
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <Channel: <ANON>@0x7f5414996190>
    def get_topic_name(self) -> str:
        """Get the topic name, or raise if this is not a named channel."""
>       raise NotImplementedError("Channels are unnamed topics")
E       NotImplementedError: Channels are unnamed topics
faust/channels.py:197: NotImplementedError

[bobh66]

I'm guessing that the problem is here:

--- a/faust/agents/agent.py
+++ b/faust/agents/agent.py
@@ -1,5 +1,6 @@
 """Agent implementation."""
 import asyncio
+import sys
 import typing
 from contextlib import suppress
 from contextvars import ContextVar
@@ -658,7 +659,19 @@ class Agent(AgentT, Service):
         else:
             # agent yields and is an AsyncIterator so we have to consume it.
             coro = self._slurp(aref, aiter(aref))
-        task = asyncio.Task(self._execute_actor(coro, aref), loop=self.loop)
+        req_version = (3, 8)
+        cur_version = sys.version_info
+        if cur_version >= req_version:
+            task = asyncio.Task(
+                self._execute_actor(coro, aref),
+                loop=self.loop,
+                name=f"{str(aref)}-{self.channel.get_topic_name()}",
+            )
+        else:
+            task = asyncio.Task(
+                self._execute_actor(coro, aref),
+                loop=self.loop,
+            )