diff --git a/faust/transport/drivers/aiokafka.py b/faust/transport/drivers/aiokafka.py
index 544dbd483..18cdb30d7 100644
--- a/faust/transport/drivers/aiokafka.py
+++ b/faust/transport/drivers/aiokafka.py
@@ -572,13 +572,16 @@ def _create_worker_consumer(
         conf = self.app.conf
         if self.consumer.in_transaction:
             isolation_level = "read_committed"
-        # Table recovery depends on app.assignor state to map changelog
-        # active/standby partitions. Keep Faust assignor enabled whenever
-        # this app has changelog tables configured.
+        # Transaction IDs depend on app.assignor topic groups. Table recovery
+        # also needs assignment metadata when changelog tables are configured.
         has_changelog_tables = bool(self.app.tables.changelog_topics)
         self._assignor = (
             self.app.assignor
-            if self.app.conf.table_standby_replicas > 0 or has_changelog_tables
+            if (
+                self.consumer.in_transaction
+                or self.app.conf.table_standby_replicas > 0
+                or has_changelog_tables
+            )
             else RoundRobinPartitionAssignor
         )
         auth_settings = credentials_to_aiokafka_auth(
diff --git a/tests/unit/transport/drivers/test_aiokafka.py b/tests/unit/transport/drivers/test_aiokafka.py
index 21df98711..3c8ef02a5 100644
--- a/tests/unit/transport/drivers/test_aiokafka.py
+++ b/tests/unit/transport/drivers/test_aiokafka.py
@@ -824,6 +824,17 @@ def test__create_worker_consumer__uses_roundrobin_without_tables(
             cthread._create_worker_consumer(transport)
         assert cthread._assignor is mod.RoundRobinPartitionAssignor
 
+    def test__create_worker_consumer__uses_faust_assignor_for_transactions(
+        self, *, cthread, app
+    ):
+        app.conf.table_standby_replicas = 0
+        app.tables._changelogs.clear()
+        cthread.consumer.in_transaction = True
+        transport = cthread.transport
+        with patch("aiokafka.AIOKafkaConsumer"):
+            cthread._create_worker_consumer(transport)
+        assert cthread._assignor is app.assignor
+
     def test__create_worker_consumer__uses_faust_assignor_with_changelog_topics(
         self, *, cthread, app
     ):