Resolved #278.

Author: j3-signalroom

src/thread_safe_kafka_topics_analyzer.py

@@ -214,7 +214,7 @@ def update_progress() -> None:
                 progress = (self.completed_topics / self.total_topics) * 100
                 logging.info("Progress: %d of %d (%.1f%%) topics completed", self.completed_topics, self.total_topics, progress)
 
-        def analyze_topic_worker(topic_name: str, topic_info: Dict) -> Dict:
+        def analyze_topic_worker(topic_name: str, topic_info: Dict, start_time_epoch: int) -> Dict:
             """Worker function to analyze a single topic.
             
             Args:
@@ -239,7 +239,7 @@ def analyze_topic_worker(topic_name: str, topic_info: Dict) -> Dict:
                     # Use sample records approach
 
                     # Calculate the ISO 8601 formatted start timestamp of the rolling window
-                    rolling_start = topic_info['utc_now'] - timedelta(days=topic_info['sampling_days_based_on_retention_days'])
+                    rolling_start = start_time_epoch - timedelta(days=topic_info['sampling_days_based_on_retention_days'])
                     iso_start_time = datetime.fromisoformat(rolling_start.strftime('%Y-%m-%dT%H:%M:%S+00:00'))
                     start_time_epoch_ms = int(rolling_start.timestamp() * 1000)
 
@@ -260,7 +260,7 @@ def analyze_topic_worker(topic_name: str, topic_info: Dict) -> Dict:
 
                 else:
                     # Use Metrics API approach
-                    result = thread_analyzer.analyze_topic_with_metrics(metrics_config, topic_name, topic_info)
+                    result = thread_analyzer.analyze_topic_with_metrics(metrics_config, topic_name, topic_info, start_time_epoch)
 
                 return result
 
@@ -282,7 +282,7 @@ def analyze_topic_worker(topic_name: str, topic_info: Dict) -> Dict:
         with ThreadPoolExecutor(max_workers=max_workers_per_cluster) as executor:
             # Submit all tasks
             future_to_topic = {
-                executor.submit(analyze_topic_worker, topic_name, topic_info): topic_name
+                executor.submit(analyze_topic_worker, topic_name, topic_info, analysis_start_time_epoch): topic_name
                 for topic_name, topic_info in topics_to_analyze.items()
             }
 
@@ -308,7 +308,7 @@ def analyze_topic_worker(topic_name: str, topic_info: Dict) -> Dict:
                     update_progress()
 
                 except Exception as e:
-                    logging.error("Error processing topic %s: %s", topic_name, e)
+                    logging.warning("Failed processing topic %s, because of %s", topic_name, e)
                     update_progress()
 
         # Calculate summary statistics

src/thread_safe_topic_analyzer.py

@@ -121,13 +121,14 @@ def analyze_topic(self,
             'is_internal': topic_name.startswith('_')
         }
 
-    def analyze_topic_with_metrics(self, metrics_config: Dict, topic_name: str, topic_info: Dict) -> Dict:
+    def analyze_topic_with_metrics(self, metrics_config: Dict, topic_name: str, topic_info: Dict, start_time_epoch: int) -> Dict:
         """Analyze a single topic using Metrics API.
 
         Args:
             metrics_config (Dict): Configuration dictionary for Metrics API client.
             topic_name (str): The name of the topic to analyze.
             topic_info (Dict): Metadata and retention info about the topic.
+            start_time_epoch (int): The start time in epoch milliseconds for sampling.
 
         Returns:
             Dict: Analysis results including partition count, compaction status,
@@ -240,10 +241,11 @@ def analyze_topic_with_metrics(self, metrics_config: Dict, topic_name: str, topi
             retry = 0
 
             while retry < max_retries:
-                # Calculate the ISO 8601 formatted start and end times within a rolling window for the last 1 day
-                rolling_days_start = topic_info['utc_now'] - timedelta(days=topic_info['sampling_days_based_on_retention_days'])
+                # Calculate the ISO 8601 formatted start timestamp of the rolling window
+                utc_now = datetime.fromtimestamp(start_time_epoch)
+                rolling_days_start = utc_now - timedelta(days=topic_info['sampling_days_based_on_retention_days'])
                 iso_start_time = rolling_days_start.strftime('%Y-%m-%dT%H:%M:%S')
-                iso_end_time = topic_info['utc_now'].strftime('%Y-%m-%dT%H:%M:%S')
+                iso_end_time = utc_now.strftime('%Y-%m-%dT%H:%M:%S')
                 query_start_time =  datetime.fromisoformat(iso_start_time.replace('Z', '+00:00'))
                 query_end_time = datetime.fromisoformat(iso_end_time.replace('Z', '+00:00'))
 

tests/test_metrics_client.py

@@ -47,7 +47,7 @@ class TestMetricsClient:
     def test_get_topic_received_total_bytes(self, metrics_client, kafka_cluster_id, kafka_topic_name):
         """Test the get_topic_total() function for getting the total bytes."""
 
-        # Calculate the ISO 8601 formatted start and end times within a rolling window for the last 1 day
+        # Calculate the ISO 8601 formatted start timestamp of the rolling window
         utc_now = datetime.now(timezone.utc)
         seven_days_ago = utc_now - timedelta(days=7)
         iso_start_time = seven_days_ago.strftime('%Y-%m-%dT%H:%M:%S')
@@ -71,7 +71,7 @@ def test_get_topic_received_total_bytes(self, metrics_client, kafka_cluster_id,
     def test_get_topic_received_total_records(self, metrics_client, kafka_cluster_id, kafka_topic_name):
         """Test the get_topic_total() function for getting the total records."""
 
-        # Calculate the ISO 8601 formatted start and end times within a rolling window for the last 1 day
+        # Calculate the ISO 8601 formatted start timestamp of the rolling window
         utc_now = datetime.now(timezone.utc)
         seven_days_ago = utc_now - timedelta(days=7)
         iso_start_time = seven_days_ago.strftime('%Y-%m-%dT%H:%M:%S')
@@ -174,7 +174,7 @@ def test_is_topic_partition_hot_by_ingress_throughput(self, metrics_client, kafk
         """Test the is_topic_partition_hot() function for checking if a topic partition is hot
         by ingress throughput."""
 
-        # Calculate the ISO 8601 formatted start and end times within a rolling window for the last 1 day
+        # Calculate the ISO 8601 formatted start timestamp of the rolling window
         utc_now = datetime.now(timezone.utc)
         seven_days_ago = utc_now - timedelta(days=7)
         iso_start_time = seven_days_ago.strftime('%Y-%m-%dT%H:%M:%S')
@@ -199,7 +199,7 @@ def test_is_topic_partition_hot_by_egress_throughput(self, metrics_client, kafka
         """Test the is_topic_partition_hot() function for checking if a topic partition is hot
         by egress throughput."""
 
-        # Calculate the ISO 8601 formatted start and end times within a rolling window for the last 1 day
+        # Calculate the ISO 8601 formatted start timestamp of the rolling window
         utc_now = datetime.now(timezone.utc)
         seven_days_ago = utc_now - timedelta(days=7)
         iso_start_time = seven_days_ago.strftime('%Y-%m-%dT%H:%M:%S')