.

getsentry · Nov 29, 2023 · e379aba · e379aba
1 parent 4eb66b1
commit e379aba
Showing 1 changed file with 13 additions and 10 deletions.
diff --git a/arroyo/dlq.py b/arroyo/dlq.py
@@ -329,16 +329,19 @@ def __init__(
     ) -> None:
         self.MAX_PENDING_FUTURES = 1000  # This is a per partition max
         self.__dlq_policy = policy
-        if policy:
-            self.__futures: MutableMapping[
-                Partition,
-                Deque[
-                    Tuple[
-                        BrokerValue[TStrategyPayload],
-                        Future[BrokerValue[TStrategyPayload]],
-                    ]
-                ],
-            ] = defaultdict(deque)
+        if policy is None:
+            return
+
+        self.__futures: MutableMapping[
+            Partition,
+            Deque[
+                Tuple[
+                    BrokerValue[TStrategyPayload],
+                    Future[BrokerValue[TStrategyPayload]],
+                ]
+            ],
+        ] = defaultdict(deque)
+        self.reset_offsets({})
 
     def reset_offsets(self, assignment: Mapping[Partition, int]) -> None:
         """