Merge pull request #55 from areeda/Kafka-exception-catch

Stop Kafka errors propagating and allow the consumer to retry indefinitely
astronomy-commons · Apr 1, 2022 · 0d38c10 · 0d38c10
2 parents 8dcbe7b + a33abdd
commit 0d38c10
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/adc/consumer.py b/adc/consumer.py
@@ -119,8 +119,9 @@ def _stream_forever(self,
                         batch_timeout: timedelta = timedelta(seconds=1.0),
                         ) -> Iterator[confluent_kafka.Message]:
         while True:
-            messages = self._consumer.consume(batch_size, batch_timeout.total_seconds())
             try:
+                messages = self._consumer.consume(batch_size,
+                                                  batch_timeout.total_seconds())
                 for m in messages:
                     err = m.error()
                     if err is None: