diff --git a/faust/topics.py b/faust/topics.py
index b09a8d22d..630014b60 100644
--- a/faust/topics.py
+++ b/faust/topics.py
@@ -42,6 +42,7 @@
 )
 from .types.topics import ChannelT, TopicT
 from .types.transports import ProducerT
+from .types.tuples import _get_len
 
 if typing.TYPE_CHECKING:  # pragma: no cover
     from .app import App as _App
@@ -409,8 +410,12 @@ async def publish_message(
             producer,
             topic,
             message=message,
-            keysize=len(key) if key else 0,
-            valsize=len(value) if value else 0,
+            # ``key``/``value`` are not always serialized bytes here -- e.g. a
+            # raw int key produced via group_by/forward (#513) -- and this only
+            # feeds a sensor statistic, so use the bytes-safe helper rather than
+            # a bare len() that would crash the publish for non-sized payloads.
+            keysize=_get_len(key),
+            valsize=_get_len(value),
         )
         if wait:
             ret: RecordMetadata = await producer.send_and_wait(
diff --git a/tests/unit/test_topics.py b/tests/unit/test_topics.py
index 38d290fa6..e880ce000 100644
--- a/tests/unit/test_topics.py
+++ b/tests/unit/test_topics.py
@@ -7,7 +7,7 @@
 import faust
 from faust import Event, Record
 from faust.exceptions import KeyDecodeError, ValueDecodeError
-from faust.types import Message
+from faust.types import FutureMessage, Message, PendingMessage
 from tests.helpers import AsyncMock
 
 
@@ -162,6 +162,38 @@ async def test_publish_message__wait_enabled(self, *, topic, app):
         )
         callback.assert_called_once_with(fm)
 
+    @pytest.mark.asyncio
+    async def test_publish_message__non_bytes_key_does_not_crash(self, *, topic, app):
+        # A raw, unserialized non-bytes key/value can reach publish_message
+        # (e.g. an int key produced via group_by/forward). Computing the
+        # sensor keysize must not crash the publish -- see #513.
+        producer = Mock(send_and_wait=AsyncMock())
+        topic._get_producer = AsyncMock(return_value=producer)
+        topic._finalize_message = AsyncMock()
+        app.sensors.on_send_initiated = Mock(return_value={})
+        app.sensors.on_send_completed = Mock()
+
+        pending = PendingMessage(
+            channel=topic,
+            key=12345,
+            value=678,
+            partition=None,
+            timestamp=None,
+            headers=None,
+            key_serializer=None,
+            value_serializer=None,
+            callback=None,
+            topic="foo",
+        )
+        fm = FutureMessage(pending)
+
+        # Must not raise TypeError from len(int).
+        await topic.publish_message(fm, wait=True)
+
+        _, kwargs = app.sensors.on_send_initiated.call_args
+        assert kwargs["keysize"] == 0
+        assert kwargs["valsize"] == 0
+
     @pytest.mark.asyncio
     async def test_send__attachments_enabled(self, *, topic, app):
         app._attachments.enabled = True