Commit e33ba8b8 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Telemetry backend component:

- Added call to close() method
- Fixed exception log messages in method close()
parent bd059d3d
Loading
Loading
Loading
Loading
+9 −5
Original line number Diff line number Diff line
@@ -151,7 +151,7 @@ class AnalyticsBackendService(GenericGrpcService):
            LOGGER.error("Failed to stop Dask Streamer. ERROR: {:}".format(e))
            return False

    def close(self):        # TODO: Is this function needed?
    def close(self):
        """
        Close the producer and cluster cleanly.
        """
@@ -159,11 +159,15 @@ class AnalyticsBackendService(GenericGrpcService):
            try:
                self.central_producer.flush()
                LOGGER.info("Kafka producer flushed and closed.")
            except Exception as e:
                LOGGER.error(f"Error closing Kafka producer: {e}")
            except:
                LOGGER.exception("Error closing Kafka producer")
        if self.cluster:
            try:
                self.cluster.close()
                LOGGER.info("Dask cluster closed.")
            except Exception as e:
                LOGGER.error(f"Error closing Dask cluster: {e}")
            except:
                LOGGER.exception("Error closing Dask cluster")

    def stop(self):
        self.close()
        return super().stop()