add all partitions of global tables as active in order to let them re…

…cover in the beginning (faust-streaming#213) Co-authored-by: Alexander Oberegger <alexander.oberegger@smaxtec.com>
kiniamogh · Nov 3, 2021 · 60a8696 · 60a8696
1 parent 41200e4
commit 60a8696
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/faust/assignor/partition_assignor.py b/faust/assignor/partition_assignor.py
@@ -329,6 +329,9 @@ def _global_table_standby_assignments(
                     # Only add those partitions as standby which aren't active
                     standby_partitions = all_partitions - active_partitions
                     assignment.standbys[changelog_topic_name] = list(standby_partitions)
+                    # We add all_partitions as active so they are recovered
+                    # in the beginning.
+                    assignment.actives[changelog_topic_name] = list(all_partitions)
         return assignments
 
     def _protocol_assignments(