Random "failed to retrieve" error in streaming scenario

Hello together,

I currently working on a lokal setup where solace software acts as the message broker in docker setup. I try to receive events from a existing queue, which is not a problem in general, but in my case I emit them in a flink job (wirtten with pyflink). I receive around 1000-3000 events/sec from this queue. I can receive around 150k events and randomly i get the following error and my setup stops working (see below).

The error varies from "Failed to retrieved" to "payload of message cannot be read" or "message id not present" all this results that my flink job stops working.

I want to mentioned that the data that is inserted in solace broker is generated by myself. Those messages are 100% bulletproof and have a message body etc. I also tried to implement some message acknowledgement in general but this doesnt work either.

Any hint in any way I should look for is really appreciated! 😊 As for now I am kinda lost about any solution, because the retrieving works for so many messages, that are generated by my application and validated against a schema etc.

Would like to hear from you guys!

If something is missing just write me :-)

Error Message:
13:11:25,779 [WARNING] solace.messaging.core: [_message.py:497] Failed to retrieve2024-11-23 14:11:25 Caller Description: Message->process_rest_data. Error Info Sub code: [2]. Error: [Bad msg_p pointer '(nil)' in _solClient_msgHeaderMap_getString]. Sub code: [SOLCLIENT_SUBCODE_PARAM_NULL_PTR]. Return code: [Fail]2024-11-23 14:11:25 2024-11-23 13:11:25,780 [ERROR] solace.messaging.receiver: [_persistent_message_receiver.py:348] [[SERVICE: 0xffff80e3beb0] [RECEIVER: 0xffff81f88310]] Failed to retrieve2024-11-23 14:11:25 Caller Description: Message->process_rest_data. Error Info Sub code: [2]. Error: [Bad msg_p pointer '(nil)' in _solClient_msgHeaderMap_getString]. Sub code: [SOLCLIENT_SUBCODE_PARAM_NULL_PTR]. Return code: [Fail]2024-11-23 14:11:25 Exception ignored on calling ctypes callback function: <bound method _PersistentMessageReceiver._flow_message_receive_callback_routine of <solace.messaging.receiver._impl._persistent_message_receiver._PersistentMessageReceiver object at 0xffff81f88310>>2024-11-23 14:11:25 Traceback (most recent call last):2024-11-23 14:11:25 File "/usr/local/lib/python3.10/dist-packages/solace/messaging/receiver/_impl/_persistent_message_receiver.py", line 349, in _flow_message_receive_callback_routine2024-11-23 14:11:25 raise PubSubPlusClientError(message=exception) from exception2024-11-23 13:11:25,781 [WARNING] solace.messaging.core.api: [_inbound_message_utility.py:192] Unable to get message id2024-11-23 14:11:25 Caller Description: _InboundMessage->get_message_id. Error Info Sub code: [2]. Error: [Bad msg_p pointer '(nil)' in solClient_msg_getMsgId]. Sub code: [SOLCLIENT_SUBCODE_PARAM_NULL_PTR]. Return code: [Fail]2024-11-23 14:11:25 Error during message settlement: Unable to get message id2024-11-23 14:11:25 Caller Description: _InboundMessage->get_message_id. Error Info Sub code: [2]. Error: [Bad msg_p pointer '(nil)' in solClient_msg_getMsgId]. Sub code: [SOLCLIENT_SUBCODE_PARAM_NULL_PTR]. Return code: [Fail]

Custom MessageHandlerImpl Class:
class MessageHandlerImpl(MessageHandler): def __init__(self, flink_source, persistent_receiver: PersistentMessageReceiver): self.flink_source = flink_source self.receiver: PersistentMessageReceiver = persistent_receiver def on_message(self, message: InboundMessage): try: payload = message.get_payload_as_string() if message.get_payload_as_string() is not None else message.get_payload_as_bytes() if isinstance(payload, bytearray): print(f"Received a message of type: {type(payload)}. Decoding to string") payload = payload.decode() topic = message.get_destination_name() print(f"Received message on: {topic}") #print(f"Message payload test: {payload}") try: # Attempt to emit the payload to Flink self.flink_source.emit(payload) except Exception as e: print(f"Error emitting message to Flink: {e}") # Nack the message so it gets sent to the DLQ self.receiver.ack(message) return # If everything succeeded, acknowledge the message self.receiver.ack(message) except PubSubPlusClientError as e: print(f"Error during message settlement: {e}") except Exception as e: print(f"Unexpected error during settlement: {e}")