SAM0123-231: Remove Kafka integration from Ingestor service.

ingestor/ingestor.py

@@ -19,7 +19,7 @@ class Ingestor(SientiaMonitoring):
     - Managing slot leases for load balancing across multiple instances
     - Connecting to and monitoring OPC UA servers
     - Subscribing to OPC tags and collecting real-time data
-    - Distributing data to Kafka and MongoDB
+    - Distributing data to MongoDB
     - Providing health monitoring and metrics collection
 
     The ingestor uses a slot-based architecture where each slot represents
@@ -27,8 +27,6 @@ class Ingestor(SientiaMonitoring):
     This allows for horizontal scaling and load distribution.
 
     Environment Variables:
-        KAFKA_SERVERS: Comma-separated list of Kafka server addresses (default: "localhost:9092")
-        EXPORT_TO_KAFKA: Enable/disable Kafka export (default: "false")
         REDIS_HOST: Redis server hostname (default: "localhost")
         REDIS_PORT: Redis server port (default: 6379)
         REDIS_USERNAME: Redis username (optional)
@@ -43,7 +41,6 @@ class Ingestor(SientiaMonitoring):
         MONGODB_DATABASE: MongoDB database name (default: "sientia")
 
     Attributes:
-        export_to_kafka (bool): Whether to export data to Kafka
         redis_host (str): Redis server hostname
         redis_port (int): Redis server port
         redis_username (str): Redis username (optional)
@@ -54,7 +51,6 @@ class Ingestor(SientiaMonitoring):
         poll_interval (int): Interval in seconds for polling operations
         mongo_database (str): MongoDB database name
         mongo_connection_string (str): Complete MongoDB connection string
-        kafka_servers (list): List of Kafka server addresses
         logger: Logger instance for application logging
         notification_handler: Handler for sending notifications
         metadata (dict): Application metadata for notifications and tracking
@@ -66,17 +62,12 @@ class Ingestor(SientiaMonitoring):
         Initializes the ingestor with configuration values retrieved from environment variables.
 
         Sets up all necessary connections and configurations for:
-        - Kafka connectivity (if enabled)
         - Redis for slot management and coordination
         - MongoDB for data persistence and notifications
         - OPC UA server management
         - Metrics collection and monitoring
         """
 
-        kafka_servers = getenv('KAFKA_SERVERS', 'localhost:9092')
-        export_to_kafka: bool = getenv('EXPORT_TO_KAFKA', 'false') == 'true'
-
-        self.export_to_kafka = export_to_kafka
         self.redis_host = getenv('REDIS_HOST', 'localhost')
         self.redis_port = int(getenv('REDIS_PORT', '6379'))
         self.redis_username = getenv('REDIS_USERNAME', None)
@@ -91,7 +82,6 @@ class Ingestor(SientiaMonitoring):
         self.mongo_database = getenv('MONGODB_DATABASE', 'sientia')
         self.mongo_connection_string = f'mongodb://{mongo_username}:{mongo_password}@{mongo_url}'
 
-        self.kafka_servers = kafka_servers.split(',')
         self.logger = get_logger(__name__)
         self.notification_handler = NotificationHandler(
             connection_string=self.mongo_connection_string,
@@ -179,7 +169,6 @@ class Ingestor(SientiaMonitoring):
         """
 
         self.ingestor_manager = IngestorManager(
-            kafka_servers=','.join(self.kafka_servers),
             redis_data={
                 'host': self.redis_host,
                 'port': self.redis_port,
@@ -195,7 +184,6 @@ class Ingestor(SientiaMonitoring):
             logger=self.logger,
             notification_handler=self.notification_handler,
             metrics_controller=self.metrics_controller,
-            export_to_kafka=self.export_to_kafka,
         )
         assert self.ingestor_manager is not None
 

ingestor/managers/data_manager.py

@@ -1,10 +1,6 @@
-import json
 import os
 import traceback
-from time import sleep
 
-from kafka import KafkaProducer
-from kafka.errors import NoBrokersAvailable
 from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
 from sientia_do.notifications.models import NotificationLevel
 from sientia_do.observability.logger import Logger
@@ -18,31 +14,22 @@ import ingestor.metrics as metrics
 
 class DataManager(SientiaMonitoring):
     """
-    Manages data persistence and export operations for the OPC Ingestor.
+    Manages data persistence operations for the OPC Ingestor.
 
     The DataManager is responsible for:
     - Storing OPC data in MongoDB for historical analysis and persistence
-    - Exporting data to Kafka for real-time streaming and downstream processing
     - Managing database connections and ensuring data integrity
     - Providing data access interfaces for other components
 
-    The manager supports both MongoDB and Kafka operations, with Kafka export
-    being optional and configurable. It implements retry logic for connection
-    failures and provides comprehensive error handling and notification.
-
     Args:
-        kafka_servers (str): Comma-separated string of Kafka server addresses
         mongo_connection_string (str): MongoDB connection string
         mongo_database (str): MongoDB database name
-        export_to_kafka (bool): Whether to enable Kafka export functionality
         metadata (dict): Application metadata for notifications and tracking
         logger (Logger): Logger instance for application logging
         notification_handler (NotificationHandler): Handler for sending notifications
 
     Attributes:
         pod_id (str): Pod identifier for metrics labeling
-        kafka_producer (KafkaProducer): Kafka producer instance for data export
-        export_to_kafka (bool): Whether Kafka export is enabled
         connection_string (str): MongoDB connection string
         database (str): MongoDB database name
         mongo_client (MongoClient): MongoDB client instance
@@ -51,47 +38,25 @@ class DataManager(SientiaMonitoring):
 
     def __init__(
         self,
-        kafka_servers: str,
         mongo_connection_string: str,
         mongo_database: str,
-        export_to_kafka: bool,
         metadata: dict,
         logger: Logger,
         notification_handler: NotificationHandler,
         metrics_controller: MetricsController,
     ) -> None:
         """
-        Initializes the DataManager instance with Kafka and MongoDB connections.
+        Initializes the DataManager instance with a MongoDB connection.
-
-        This constructor attempts to establish connections to the specified services:
-        1. Kafka: Initializes producer with retry logic (up to 3 attempts)
-        2. MongoDB: Establishes connection and verifies server availability
-
-        The initialization process includes:
-        - Kafka producer setup with JSON serialization
-        - MongoDB client initialization and connection testing
-        - Metrics recording for connection status
-        - Error handling with notifications
 
         Args:
-            kafka_servers (str): Comma-separated string of Kafka server addresses
             mongo_connection_string (str): MongoDB connection string
             mongo_database (str): MongoDB database name
-            export_to_kafka (bool): Whether to enable Kafka export
             metadata (dict): Application metadata
             logger (Logger): Logger instance
             notification_handler (NotificationHandler): Notification handler
-
-        Raises:
-            NoBrokersAvailable: If the connection to Kafka servers fails after 3 attempts.
-
-        Metrics:
-            - KAFKA_CONNECTION_STATUS: Set to 1 on successful connection, 0 on failure
         """
 
         self.pod_id = os.getenv('HOSTNAME', 'localhost')
-        self.kafka_producer = None
-        self.export_to_kafka = export_to_kafka
 
         SientiaMonitoring.__init__(
             self,
@@ -100,37 +65,6 @@ class DataManager(SientiaMonitoring):
             notification_handler=notification_handler,
         )
 
-        if self.export_to_kafka:
-            for i in range(0, 3):
-                logger.info(
-                    f'Trying ({i}) to initializing DataManager with Kafka servers: {kafka_servers}'
-                )
-                try:
-                    self.kafka_producer = KafkaProducer(
-                        bootstrap_servers=kafka_servers,
-                        value_serializer=lambda v: json.dumps(v).encode(
-                            'utf-8'
-                        ),  # Serialize JSON messages
-                        key_serializer=lambda k: str(k).encode('utf-8') if k else None,
-                    )
-                    # Kafka connected
-                    metrics.KAFKA_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(1)
-                    break
-                except NoBrokersAvailable:
-                    logger.error(f'Kafka servers {kafka_servers} are not available. Retrying...')
-                    sleep(5)
-            else:
-                # Kafka not connected
-                metrics.KAFKA_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(0)
-                logger.error(
-                    f'Failed to connect to Kafka servers {kafka_servers} after 3 attempts.'
-                )
-                raise NoBrokersAvailable(
-                    f'Failed to connect to Kafka servers {kafka_servers} after 3 attempts.'
-                )
-
-            logger.info(f'DataManager initialized with Kafka servers: {kafka_servers}')
-
         logger.info(
             f'Trying to initializing DataManager with MongoDB servers: {mongo_connection_string}'
         )
@@ -153,26 +87,7 @@ class DataManager(SientiaMonitoring):
     def shutdown(self):
         """
         Gracefully shuts down the DataManager and closes all connections.
-
-        This method ensures proper cleanup of:
-        - Kafka producer connection with message flushing
-        - MongoDB client connection
-        - Metrics recording for connection status
-
-        The method handles connection closure gracefully, logging any errors
-        that occur during shutdown while ensuring all resources are properly released.
         """
-        if self.kafka_producer:
-            try:
-                self.kafka_producer.flush(timeout=10)
-                self.kafka_producer.close()
-                # Mark as disconnected
-                metrics.KAFKA_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(0)
-            except Exception as e:
-                self.logger.error(f'Error closing Kafka producer: {e}')
-        else:
-            self.logger.warning('Kafka producer is already closed or not initialized.')
-
         try:
             self.mongo_repository.close()
         except Exception as e:
@@ -181,83 +96,18 @@ class DataManager(SientiaMonitoring):
     def __del__(self):
         self.shutdown()
 
-    def delivery_report(self, msg):
-        """
-        Callback for successful Kafka message delivery reports.
-
-        This method is called by the Kafka producer when a message is successfully
-        delivered to a topic. It logs the delivery details including topic, partition,
-        and offset information for debugging and monitoring purposes.
-
-        Args:
-            msg: Kafka message object containing delivery details
-        """
-        self.logger.debug(
-            f'Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}'
-        )
-
-    def delivery_error(self, err):
-        """
-        Callback for Kafka message delivery error reports.
-
-        This method is called by the Kafka producer when a message delivery fails.
-        It logs the error details for debugging and monitoring purposes.
-
-        Args:
-            err: Error information from the failed delivery attempt
-        """
-        self.logger.error(f'Delivery failed for record : {err}')
-
     async def publish(self, topic: str, data: dict) -> None:
         """
-        Publishes a message to a specified Kafka topic.
+        Persists a message to MongoDB.
 
         Args:
-            topic (str): The name of the Kafka topic to which the message will be published.
+            topic (str): The name of the MongoDB collection to which the message will be written.
-            data (dict): The message data to be sent to the Kafka topic.
+            data (dict): The message data to be stored.
 
         Returns:
             None
-
-        Raises:
-            Exception: If there is an error during message delivery, it will be handled by the `delivery_error` callback.
         """
 
-        if self.export_to_kafka and self.kafka_producer:
-            try:
-                self.logger.debug(f'Publishing message to topic {topic}: {data}')
-                self.kafka_producer.send(topic=topic, value=data).add_callback(
-                    self.delivery_report
-                ).add_errback(self.delivery_error)
-
-                self.kafka_producer.flush(timeout=10)
-                await self.emit_metric(
-                    metric_object=metrics.KAFKA_MESSAGES_SENT,
-                    tags={
-                        'pod_id': self.pod_id,
-                        'topic': topic,
-                    },
-                )
-
-            except Exception as e:
-                await self.emit_metric(
-                    metric_object=metrics.KAFKA_MESSAGES_ERRORS,
-                    tags={
-                        'pod_id': self.pod_id,
-                        'topic': topic,
-                    },
-                )
-                trace = traceback.format_exc()
-                await self.send_notification_async(
-                    metadata=self.metadata,
-                    notification_id=f'KAFKA_PRODUCER_ERROR_{topic}',
-                    message=f'Error publishing message to topic {topic}: {e}',
-                    block='kafka_producer',
-                    level=NotificationLevel.ERROR,
-                    attachment_content=trace,
-                )
-                self.logger.error(trace)
-
         try:
             await self.mongo_repository.insert(
                 collection_name=topic,

ingestor/managers/ingestor_manager.py

@@ -19,7 +19,7 @@ class IngestorManager(SientiaMonitoring):
     Central coordinator for managing OPC data ingestion operations.
 
     The IngestorManager orchestrates the interaction between different components:
-    - DataManager: Handles data persistence and Kafka export
+    - DataManager: Handles data persistence
     - OPC Managers: Manage individual OPC UA server connections
     - ResourceManager: Coordinates slot leasing and load balancing
 
@@ -36,7 +36,6 @@ class IngestorManager(SientiaMonitoring):
     - Load balancing across multiple ingestor instances
 
     Args:
-        kafka_servers (str): Comma-separated list of Kafka server addresses
         redis_data (dict): Redis connection parameters (host, port, username, password)
         lease_ttl (int): Time-to-live for slot leases in seconds
         heartbeat_ttl (int): Time-to-live for heartbeat signals in seconds
@@ -46,10 +45,9 @@ class IngestorManager(SientiaMonitoring):
         metadata (dict): Application metadata for notifications and tracking
         logger (Logger): Logger instance for application logging
         notification_handler (NotificationHandler): Handler for sending notifications
-        export_to_kafka (bool): Whether to export data to Kafka
 
     Attributes:
-        data_manager (DataManager): Manages data persistence and Kafka export
+        data_manager (DataManager): Manages data persistence
         opc_managers (dict): Dictionary of OPC managers keyed by server name
         resource_manager (ResourceManager): Manages Redis-based resource coordination
         number_of_slots (int): Total number of slots configured in the system
@@ -61,7 +59,6 @@ class IngestorManager(SientiaMonitoring):
 
     def __init__(
         self,
-        kafka_servers: str,
         redis_data: dict,
         lease_ttl: int,
         heartbeat_ttl: int,
@@ -72,7 +69,6 @@ class IngestorManager(SientiaMonitoring):
         logger: Logger,
         notification_handler: NotificationHandler,
         metrics_controller: MetricsController,
-        export_to_kafka: bool = False,
     ):
         redis_host: str = redis_data['host']
         redis_port: int = int(redis_data['port'])
@@ -87,10 +83,8 @@ class IngestorManager(SientiaMonitoring):
         )
 
         self.data_manager = DataManager(
-            kafka_servers=kafka_servers,
             mongo_connection_string=mongo_connection_string,
             mongo_database=mongo_database,
-            export_to_kafka=export_to_kafka,
             metadata=metadata,
             logger=logger,
             notification_handler=notification_handler,

ingestor/metrics.py

@@ -19,7 +19,6 @@ from prometheus_client import Counter, Gauge, Histogram
 # Metric label definitions for consistent labeling across all metrics
 POD_ID_LABEL = ['pod_id']
 SERVER_LABELS = ['pod_id', 'server_name', 'server_url']
-KAFKA_LABELS = ['pod_id', 'topic']
 REDIS_LABELS = ['pod_id', 'operation']
 NOTIFICATION_LABELS = ['pod_id', 'level', 'block']
 
@@ -134,22 +133,6 @@ OPC_RECONNECTIONS_TOTAL = Counter(
     ['pod_id', 'server_name'],
 )
 
-# --- Data Manager (Kafka) Metrics ---
-KAFKA_MESSAGES_SENT = Counter(
-    'kafka_messages_sent_total', 'Total messages sent to Kafka', KAFKA_LABELS
-)
-KAFKA_MESSAGES_ERRORS = Counter(
-    'kafka_messages_errors_total',
-    'Total errors sending messages to Kafka',
-    KAFKA_LABELS,
-)
-KAFKA_CONNECTION_STATUS = Gauge(
-    'kafka_connection_status',
-    'Connection status with Kafka (1=connected, 0=disconnected)',
-    POD_ID_LABEL,
-)
-
-
 # --- Notification Metrics ---
 NOTIFICATIONS_SENT = Counter(
     'notifications_sent_total',

tests/unit/managers/test_data_manager.py

@@ -1,6 +1,5 @@
 from unittest.mock import ANY, AsyncMock, MagicMock, patch
 
-from kafka.errors import NoBrokersAvailable
 from pytest import fixture, mark
 from sientia_do.notifications.models import NotificationLevel
 
@@ -18,14 +17,11 @@ metadata = {
 
 
 @fixture
-@patch('ingestor.managers.data_manager.KafkaProducer')
 @patch('ingestor.managers.data_manager.MongoDBRepository')
-def data_manager(mongodb_repository, kafka):
+def data_manager(mongodb_repository):
     data_manager = DataManager(
-        kafka_servers='localhost:9092',
         mongo_connection_string='mongodb://localhost:27017',
         mongo_database='sientia',
-        export_to_kafka=True,
         logger=MagicMock(),
         notification_handler=MagicMock(),
         metadata=metadata['metadata'],
@@ -38,141 +34,34 @@ def data_manager(mongodb_repository, kafka):
     return data_manager
 
 
-@patch('ingestor.managers.data_manager.KafkaProducer')
 @patch('ingestor.managers.data_manager.MongoDBRepository')
-def test___init___success(mongodb_repository, kafka):
+def test___init___success(mongodb_repository):
     logger_mock = MagicMock()
 
     data_manager = DataManager(
         metadata=metadata['metadata'],
-        kafka_servers='localhost:9092',
         mongo_connection_string='mongodb://localhost:27017',
         mongo_database='sientia',
-        export_to_kafka=True,
         logger=logger_mock,
         notification_handler=MagicMock(),
         metrics_controller=MagicMock(),
     )
 
-    kafka.assert_called_once_with(
-        bootstrap_servers='localhost:9092', value_serializer=ANY, key_serializer=ANY
-    )
-    assert data_manager.kafka_producer is not None
     logger_mock.info.assert_any_call(
-        'Trying (0) to initializing DataManager with Kafka servers: localhost:9092'
+        'DataManager initialized with MongoDB servers: mongodb://localhost:27017'
     )
-    logger_mock.info.assert_any_call('DataManager initialized with Kafka servers: localhost:9092')
     logger_mock.error.assert_not_called()
-    assert logger_mock.info.call_count == 4
+    assert data_manager.connection_string == 'mongodb://localhost:27017'
 
 
-@patch('ingestor.managers.data_manager.KafkaProducer')
+def test_shutdown(data_manager):
-@patch('ingestor.managers.data_manager.MongoDBRepository')
-def test___init___second_attempt(mongodb_repository, kafka):
-    kafka.side_effect = [NoBrokersAvailable, MagicMock()]
-    logger_mock = MagicMock()
-
-    data_manager = DataManager(
-        kafka_servers='localhost:9092',
-        mongo_connection_string='mongodb://localhost:27017',
-        mongo_database='sientia',
-        export_to_kafka=True,
-        logger=logger_mock,
-        notification_handler=MagicMock(),
-        metadata=metadata['metadata'],
-        metrics_controller=MagicMock(),
-    )
-
-    kafka.assert_any_call(
-        bootstrap_servers='localhost:9092', value_serializer=ANY, key_serializer=ANY
-    )
-    assert kafka.call_count == 2
-    assert data_manager.kafka_producer is not None
-    logger_mock.info.assert_any_call(
-        'Trying (0) to initializing DataManager with Kafka servers: localhost:9092'
-    )
-    logger_mock.info.assert_any_call(
-        'Trying (1) to initializing DataManager with Kafka servers: localhost:9092'
-    )
-    logger_mock.info.assert_any_call('DataManager initialized with Kafka servers: localhost:9092')
-    logger_mock.error.assert_called_once_with(
-        'Kafka servers localhost:9092 are not available. Retrying...'
-    )
-    assert logger_mock.info.call_count == 5
-
-
-@patch('ingestor.managers.data_manager.KafkaProducer')
-@patch('ingestor.managers.data_manager.MongoDBRepository')
-def test___init___failure_max_attempts(mongodb_repository, kafka):
-    kafka.side_effect = NoBrokersAvailable
-    logger_mock = MagicMock()
-
-    try:
-        DataManager(
-            kafka_servers='localhost:9092',
-            mongo_connection_string='mongodb://localhost:27017',
-            mongo_database='sientia',
-            export_to_kafka=True,
-            logger=logger_mock,
-            notification_handler=MagicMock(),
-            metadata=metadata['metadata'],
-            metrics_controller=MagicMock(),
-        )
-    except NoBrokersAvailable as e:
-        assert (
-            str(e)
-            == 'NoBrokersAvailable: Failed to connect to Kafka servers localhost:9092 after 3 attempts.'
-        )
-
-        assert kafka.call_count == 3
-        logger_mock.info.assert_any_call(
-            'Trying (0) to initializing DataManager with Kafka servers: localhost:9092'
-        )
-        logger_mock.info.assert_any_call(
-            'Trying (1) to initializing DataManager with Kafka servers: localhost:9092'
-        )
-        logger_mock.info.assert_any_call(
-            'Trying (2) to initializing DataManager with Kafka servers: localhost:9092'
-        )
-        logger_mock.error.assert_called_with(
-            'Failed to connect to Kafka servers localhost:9092 after 3 attempts.'
-        )
-        assert logger_mock.info.call_count == 3
-
-    else:
-        raise AssertionError('Expected NoBrokersAvailable exception was not raised.')
-
-
-def test_shutdown_has_producer(data_manager):
-    flush_mock = MagicMock()
     close_mock = MagicMock()
-
+    data_manager.mongo_repository.close = close_mock
-    data_manager.kafka_producer.flush = flush_mock
-    data_manager.kafka_producer.close = close_mock
 
     data_manager.shutdown()
-    flush_mock.assert_called_once()
     close_mock.assert_called_once()
 
 
-def test_shutdown_no_producer(data_manager):
-    data_manager.kafka_producer = None
-
-    data_manager.shutdown()
-
-    data_manager.logger.warning.assert_any_call(
-        'Kafka producer is already closed or not initialized.'
-    )
-
-
-def test_shutdown_exception(data_manager):
-    data_manager.kafka_producer.flush = MagicMock(side_effect=Exception('Test error'))
-    data_manager.kafka_producer.close = MagicMock()
-
-    data_manager.shutdown()
-    data_manager.logger.error.assert_called_once_with('Error closing Kafka producer: Test error')
-
-
 def test_shutdown_exception_mongo(data_manager):
     data_manager.mongo_repository.close = MagicMock(side_effect=Exception('Test error'))
 
@@ -187,88 +76,24 @@ def test___del__(data_manager):
     data_manager.shutdown.assert_called_once()
 
 
-def test_delivery_report(data_manager):
-    msg = MagicMock()
-    msg.topic = 'test_topic'
-    msg.partition = 0
-    msg.offset = 1
-
-    data_manager.delivery_report(msg)
-
-    data_manager.logger.debug.assert_called_once_with(
-        f'Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}'
-    )
-
-
-def test_delivery_error(data_manager):
-    err = 'Test error'
-    data_manager.delivery_error(err)
-
-    data_manager.logger.error.assert_called_once_with(f'Delivery failed for record : {err}')
-
-
 @mark.asyncio
 async def test_publish(data_manager):
-    topic = 'test_topic'
-    data = {'key': 'value'}
-
-    # Mock the send method of the Kafka producer
-    send_mock = MagicMock()
-    data_manager.kafka_producer.send = send_mock
-
-    # Call the publish method
-    await data_manager.publish(topic, data)
-
-    # Check if the send method was called with the correct arguments
-    send_mock.assert_called_once_with(topic=topic, value=data)
-
-    send_mock.return_value.add_callback.assert_called_once()
-
-    data_manager.kafka_producer.flush.assert_called_once()
-
-
-def test_publish_no_kafka(data_manager):
-    data_manager.export_to_kafka = False
-    topic = 'test_topic'
-    data = {'key': 'value'}
-
-    data_manager.publish(topic, data)
-
-    data_manager.kafka_producer.send.assert_not_called()
-
-
-@patch('ingestor.managers.data_manager.traceback')
-@mark.asyncio
-async def test_publish_error(traceback, data_manager):
     topic = 'test_topic'
     data = {'key': 'value', 'name': 'test_tag'}
 
-    # Mock the send method of the Kafka producer to raise an exception
+    data_manager.mongo_repository.insert = AsyncMock()
-    send_mock = MagicMock(side_effect=Exception('Test error'))
-    data_manager.kafka_producer.send = send_mock
 
-    data_manager.mongo_repository = AsyncMock()
-
-    # Call the publish method
     await data_manager.publish(topic, data)
 
-    # Check if the send method was called with the correct arguments
+    data_manager.mongo_repository.insert.assert_called_once_with(
-    send_mock.assert_called_once_with(topic=topic, value=data)
+        collection_name=topic,
-
+        document={**data, 'inserted_at': ANY},
-    # Check if the error was logged
-    data_manager.send_notification_async.assert_called_once_with(
-        notification_id=f'KAFKA_PRODUCER_ERROR_{topic}',
-        message=f'Error publishing message to topic {topic}: Test error',
-        block='kafka_producer',
-        level=NotificationLevel.ERROR,
-        attachment_content=traceback.format_exc.return_value,
         metadata=metadata['metadata'],
     )
 
 
 @mark.asyncio
 async def test_publish_error_mongo(data_manager):
-    data_manager.export_to_kafka = False
     data_manager.mongo_repository.insert = AsyncMock(side_effect=Exception('Test error'))
 
     await data_manager.publish('test_topic', {'key': 'value'})

tests/unit/managers/test_ingestor_manager.py

@@ -20,14 +20,12 @@ metadata = {
 @patch('ingestor.managers.ingestor_manager.ResourceManager')
 def ingestor_manager(data_manager_mock, resource_manager_mock):
     ingestor = IngestorManager(
-        kafka_servers='localhost:9092',
         redis_data={'host': 'localhost', 'port': 6379},
         lease_ttl=60,
         heartbeat_ttl=60,
         poll_interval=5,
         mongo_connection_string='mongodb://localhost:27017',
         mongo_database='sientia',
-        export_to_kafka=False,
         logger=MagicMock(),
         notification_handler=MagicMock(),
         metadata=metadata['metadata'],
@@ -49,14 +47,12 @@ def test___init__(
     notification_handler_mock, resource_manager_mock, data_manager_mock, opc_manager_mock
 ):
     ingestor = IngestorManager(
-        kafka_servers='localhost:9092',
         redis_data={'host': 'localhost', 'port': 6379},
         lease_ttl=60,
         heartbeat_ttl=60,
         poll_interval=5,
         mongo_connection_string='mongodb://localhost:27017',
         mongo_database='sientia',
-        export_to_kafka=False,
         logger=MagicMock(),
         notification_handler=MagicMock(),
         metadata=metadata['metadata'],
@@ -65,10 +61,8 @@ def test___init__(
 
     opc_manager_mock.assert_not_called()
     data_manager_mock.assert_called_once_with(
-        kafka_servers='localhost:9092',
         mongo_connection_string='mongodb://localhost:27017',
         mongo_database='sientia',
-        export_to_kafka=False,
         metadata=metadata['metadata'],
         logger=ingestor.logger,
         notification_handler=ingestor.notification_handler,

tests/unit/test_ingestor.py

@@ -9,8 +9,6 @@ from ingestor.ingestor import Ingestor
 @patch('ingestor.ingestor.NotificationHandler')
 def test___init__(notification_handler, getenv):
     getenv.side_effect = [
-        'localhost:9092,localhost:35',  # KAFKA_SERVERS
-        'true',  # EXPORT_TO_KAFKA
         'localhost',  # REDIS_HOST
         '63790',  # REDIS_PORT
         'user',  # REDIS_USERNAME
@@ -27,7 +25,6 @@ def test___init__(notification_handler, getenv):
 
     ingestor = Ingestor()
 
-    getenv.assert_any_call('KAFKA_SERVERS', 'localhost:9092')
     getenv.assert_any_call('REDIS_HOST', 'localhost')
     getenv.assert_any_call('REDIS_PORT', '6379')
     getenv.assert_any_call('REDIS_USERNAME', None)
@@ -37,7 +34,6 @@ def test___init__(notification_handler, getenv):
     getenv.assert_any_call('HOSTNAME', 'localhost')
     getenv.assert_any_call('POLL_INTERVAL', '5')
 
-    assert ingestor.kafka_servers == ['localhost:9092', 'localhost:35']
     assert ingestor.redis_host == 'localhost'
     assert ingestor.redis_port == 63790
     assert ingestor.redis_username == 'user'
@@ -124,7 +120,6 @@ async def test_prepare_ingestor(ingestor_manager_mock, ingestor):
     await ingestor.prepare_ingestor()
 
     ingestor_manager_mock.assert_called_once_with(
-        kafka_servers=','.join(ingestor.kafka_servers),
         redis_data={
             'host': ingestor.redis_host,
             'port': ingestor.redis_port,
@@ -139,7 +134,6 @@ async def test_prepare_ingestor(ingestor_manager_mock, ingestor):
         metadata=ingestor.metadata,
         logger=ingestor.logger,
         notification_handler=ingestor.notification_handler,
-        export_to_kafka=ingestor.export_to_kafka,
         metrics_controller=ingestor.metrics_controller,
     )
     ingestor_manager.declare_active.assert_called_once()

tests/unit/test_metrics.py

@@ -183,30 +183,6 @@ def test_opc_reconnections_total():
     assert set(metrics.OPC_RECONNECTIONS_TOTAL._labelnames) == {'pod_id', 'server_name'}
 
 
-def test_kafka_messages_sent():
-    """Verify the definition of KAFKA_MESSAGES_SENT."""
-    assert metrics.KAFKA_MESSAGES_SENT is not None
-    assert isinstance(metrics.KAFKA_MESSAGES_SENT, Counter)
-    assert metrics.KAFKA_MESSAGES_SENT._name == 'kafka_messages_sent'  # REMOVED _total
-    assert set(metrics.KAFKA_MESSAGES_SENT._labelnames) == {'pod_id', 'topic'}
-
-
-def test_kafka_messages_errors():
-    """Verify the definition of KAFKA_MESSAGES_ERRORS."""
-    assert metrics.KAFKA_MESSAGES_ERRORS is not None
-    assert isinstance(metrics.KAFKA_MESSAGES_ERRORS, Counter)
-    assert metrics.KAFKA_MESSAGES_ERRORS._name == 'kafka_messages_errors'  # REMOVED _total
-    assert set(metrics.KAFKA_MESSAGES_ERRORS._labelnames) == {'pod_id', 'topic'}
-
-
-def test_kafka_connection_status():
-    """Verify the definition of KAFKA_CONNECTION_STATUS."""
-    assert metrics.KAFKA_CONNECTION_STATUS is not None
-    assert isinstance(metrics.KAFKA_CONNECTION_STATUS, Gauge)
-    assert metrics.KAFKA_CONNECTION_STATUS._name == 'kafka_connection_status'
-    assert set(metrics.KAFKA_CONNECTION_STATUS._labelnames) == {'pod_id'}
-
-
 def test_notifications_sent():
     """Verify the definition of NOTIFICATIONS_SENT."""
     assert metrics.NOTIFICATIONS_SENT is not None