Publisher failure to send data

[atsag]

While running the Exponential smoothing predictor, the following exception was raised:

Exception in thread Thread-545519 (calculate_and_publish_predictions):
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/threading.py", line 1045, in _bootstrap_inner
    self.run()
  File "/usr/local/lib/python3.11/threading.py", line 982, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/local/lib/python3.11/site-packages/runtime/Predictor.py", line 268, in calculate_and_publish_predictions
    publisher.send(prediction_message_body,application_name)
  File "/usr/local/lib/python3.11/site-packages/runtime/utilities/PredictionPublisher.py", line 12, in send
    super(PredictionPublisher, self).send(body, application)
  File "/usr/local/lib/python3.11/site-packages/exn/core/publisher.py", line 17, in send
    _logger.info(f"[{self.key}] sending to {self._link.target.address} for application={application} - {body} ")
                                            ^^^^^^^^^^^^^^^^^
AttributeError: 'NoneType' object has no attribute 'target'

The publisher should have had data to send at the time it was called, so apparently something changed in the internal state of the object preventing it from correctly sending the data. Running the component again, does not yield this exception again - I would propose to add more verbosity to the logs of any exception raised in /exn/core/publisher.py, to allow more details to be retrieved later.

[fotisp]

@atsag it seems this publisher has no topic, can you provide a small sample or code?

[atsag]

Yes, for sure. The relevant code in which the publisher is called is the following (the exception is triggered in the last line in runtime/utilities/PredictionPublisher.py of the Exponential Smoothing Forecaster):

class PredictionPublisher(core.publisher.Publisher):
    metric_name = ""
    def __init__(self,application_name,metric_name):
        super().__init__('publisher_'+application_name+'-'+metric_name, 'eu.nebulouscloud.preliminary_predicted.'+EsPredictorState.forecaster_name+"."+metric_name, True,True)
        self.metric_name = metric_name

    def send(self, body={}, application=""):
        super(PredictionPublisher, self).send(body, application) #This is the line triggering the exception

As we can see, a topic is provided when the PredictionPublisher is created

[atsag]

Also, I think I replicated the error - if the connection to the broker drops (it dropped I think due to k8s port forwarding failing), then future attempts to publish will give this error.