Use default timeout for consume (#126)

esque/clients/consumer.py

@@ -121,7 +121,7 @@ def create_internal_consumer(self):
 
     def consume(self, offset: int = 0, partition: int = 0) -> Message:
         self.assign_specific_partitions(self._topic_name, partitions=[partition], offset=offset)
-        return self.consume_single_message(timeout=1)
+        return self.consume_single_message(30)
 
 
 class PingConsumer(AbstractConsumer):

esque/controller/topic_controller.py

@@ -1,9 +1,11 @@
 import datetime
+import logging
 import re
 import time
 from datetime import timezone
 from enum import Enum
 from itertools import islice
+from logging import Logger
 from typing import TYPE_CHECKING, Dict, List, Union
 
 import pendulum
@@ -19,6 +21,8 @@
 from esque.helpers import ensure_kafka_future_done, invalidate_cache_after
 from esque.resources.topic import Partition, PartitionInfo, Topic, TopicDiff
 
+logger: Logger = logging.getLogger(__name__)
+
 if TYPE_CHECKING:
     from esque.cluster import Cluster
 
@@ -180,10 +184,14 @@ def _get_partition_data(
         for partition_id, meta in confluent_topic.partitions.items():
             low = int(low_watermarks[partition_id].offset[0])
             high = int(high_watermarks[partition_id].offset[0])
+            latest_timestamp = None
             if high > low:
-                latest_timestamp = float(consumer.consume(high - 1, partition_id).timestamp()[1]) / 1000
-            else:
-                latest_timestamp = None
+                try:
+                    latest_timestamp = float(consumer.consume(high - 1, partition_id).timestamp()[1]) / 1000
+                except MessageEmptyException:
+                    logger.warning(
+                        f"Due to timeout latest timestamp for topic `{topic.name}` and partition `{partition_id}` is missing."
+                    )
             partition = Partition(partition_id, low, high, meta.isrs, meta.leader, meta.replicas, latest_timestamp)
             partitions.append(partition)
 

esque/resources/topic.py

@@ -30,9 +30,8 @@ def __init__(
         self.partition_isrs = partition_isrs
         self.partition_leader = partition_leader
         self.partition_replicas = partition_replicas
-        if latest_message_timestamp is None:
-            self.latest_message_timestamp = None
-        else:
+        self.latest_message_timestamp = None
+        if latest_message_timestamp is not None:
             self.latest_message_timestamp = pendulum.from_timestamp(latest_message_timestamp).to_datetime_string()
 
     def as_dict(self):