Closed #278.

Author: j3-signalroom

CHANGELOG.md

@@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file.
 The format is base on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 
+## [0.12.03.000] - 2025-10-07
+### Changed
+- Issue [#278](https://github.com/j3-signalroom/kafka_cluster-topics-partition_count_recommender-tool/issues/278)
+
 ## [0.12.02.000] - 2025-10-06
 ### Fixed
 - Issue [#276](https://github.com/j3-signalroom/kafka_cluster-topics-partition_count_recommender-tool/issues/276)

CHANGELOG.pdf

@@ -1,5 +1,5 @@
 import csv
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, timedelta
 import time
 from typing import Dict, List
 from confluent_kafka.admin import AdminClient, ConfigResource
@@ -105,7 +105,8 @@ def analyze_all_topics(self,
                            kafka_writer_topic_name: str,
                            kafka_writer_topic_partition_count: int,
                            kafka_writer_topic_replication_factor: int,
-                           kafka_writer_topic_data_retention_in_days: int) -> bool:
+                           kafka_writer_topic_data_retention_in_days: int,
+                           utc_now: datetime) -> bool:
         """Analyze all topics in the Kafka cluster.
         
         Args:
@@ -129,6 +130,7 @@ def analyze_all_topics(self,
             kafka_writer_topic_partition_count (int): The number of partitions for the Kafka writer topic.
             kafka_writer_topic_replication_factor (int): The replication factor for the Kafka writer topic.
             kafka_writer_topic_data_retention_in_days (int): The data retention period for the Kafka writer topic in days.
+            utc_now (datetime): Current UTC datetime for consistent time references.
 
         Returns:
             bool: True if analysis was successful, False otherwise.
@@ -139,7 +141,7 @@ def analyze_all_topics(self,
             return []
 
         # Start analysis
-        analysis_start_time_epoch = time.time()
+        analysis_start_time_epoch = int(utc_now.timestamp())
         self.total_topics = len(topics_to_analyze)
 
         # Log initial analysis parameters
@@ -237,8 +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
-                    utc_now = datetime.now(timezone.utc)
-                    rolling_start = utc_now - timedelta(days=topic_info['sampling_days_based_on_retention_days'])
+                    rolling_start = topic_info['utc_now'] - 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)
 

src/thread_safe_kafka_topics_analyzer.py

@@ -1,3 +1,4 @@
+from datetime import datetime, timezone
 import logging
 import os
 from typing import Dict
@@ -199,7 +200,8 @@ def main():
         'kafka_writer_topic_name': kafka_writer_topic_name,
         'kafka_writer_topic_partition_count': kafka_writer_topic_partition_count,
         'kafka_writer_topic_replication_factor': kafka_writer_topic_replication_factor,
-        'kafka_writer_topic_data_retention_in_days': kafka_writer_topic_data_retention_in_days
+        'kafka_writer_topic_data_retention_in_days': kafka_writer_topic_data_retention_in_days,
+        'utc_now': datetime.now(timezone.utc)
     }
 
     logging.info("=" * DEFAULT_CHARACTER_REPEAT)

src/thread_safe_tool.py

@@ -1,5 +1,5 @@
 from typing import Dict, List, Tuple
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, timedelta
 import time
 import logging
 import threading
@@ -241,10 +241,9 @@ def analyze_topic_with_metrics(self, metrics_config: Dict, topic_name: str, topi
 
             while retry < max_retries:
                 # Calculate the ISO 8601 formatted start and end times within a rolling window for the last 1 day
-                utc_now = datetime.now(timezone.utc)
-                one_day_ago = utc_now - timedelta(days=topic_info['sampling_days_based_on_retention_days'])
-                iso_start_time = one_day_ago.strftime('%Y-%m-%dT%H:%M:%S')
-                iso_end_time = utc_now.strftime('%Y-%m-%dT%H:%M:%S')
+                rolling_days_start = topic_info['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')
                 query_start_time =  datetime.fromisoformat(iso_start_time.replace('Z', '+00:00'))
                 query_end_time = datetime.fromisoformat(iso_end_time.replace('Z', '+00:00'))