From 9c438c2e1caf535e125803d79e2c0f4f18cf84bb Mon Sep 17 00:00:00 2001
From: William Barnhart <williambbarnhart@gmail.com>
Date: Wed, 21 Oct 2026 22:53:47 -0400
Subject: [PATCH] Avoid restarting transactions on retained partitions

---
 faust/transport/consumer.py           | 18 ++++++++++--------
 tests/unit/transport/test_consumer.py |  2 +-
 2 files changed, 11 insertions(+), 9 deletions(-)

diff --git a/faust/transport/consumer.py b/faust/transport/consumer.py
index 9e49a9a8f..ebac316cd 100644
--- a/faust/transport/consumer.py
+++ b/faust/transport/consumer.py
@@ -237,17 +237,19 @@ async def on_rebalance(
             )
             await T(self._stop_transactions, tids=revoked_tids)(revoked_tids)
 
-        # Start produers for assigned partitions
-        assigned_tids = sorted(self._tps_to_transactional_ids(assigned))
-        if assigned_tids:
+        # Start producers only for partitions newly assigned in this rebalance.
+        newly_assigned_tids = sorted(self._tps_to_transactional_ids(newly_assigned))
+        if newly_assigned_tids:
             self.log.info(
                 "Starting %r transactional %s for %r assigned %s...",
-                len(assigned_tids),
-                pluralize(len(assigned_tids), "producer"),
-                len(assigned),
-                pluralize(len(assigned), "partition"),
+                len(newly_assigned_tids),
+                pluralize(len(newly_assigned_tids), "producer"),
+                len(newly_assigned),
+                pluralize(len(newly_assigned), "partition"),
+            )
+            await T(self._start_transactions, tids=newly_assigned_tids)(
+                newly_assigned_tids
             )
-            await T(self._start_transactions, tids=assigned_tids)(assigned_tids)
 
     async def _stop_transactions(self, tids: Iterable[str]) -> None:
         T = traced_from_parent_span()
diff --git a/tests/unit/transport/test_consumer.py b/tests/unit/transport/test_consumer.py
index d0261179a..5350b22ed 100644
--- a/tests/unit/transport/test_consumer.py
+++ b/tests/unit/transport/test_consumer.py
@@ -182,7 +182,7 @@ async def test_on_rebalance(self, *, manager):
         manager._stop_transactions = AsyncMock()
         manager._start_transactions = AsyncMock()
 
-        assigned = {TP2}
+        assigned = {TP1, TP2}
         revoked = {TP3}
         newly_assigned = {TP2}
         await manager.on_rebalance(assigned, revoked, newly_assigned)