hanging

[rerpha]

after a while(?) the forwarder seems to hang and build up a huge queue in memory of messages it's attempted to send to kafka.

[rerpha]

on HIFI this has happened twice. I have put the log in \\isis\shares\ISIS_Experiment_Controls\data for tickets\forwarder_5 that shows it happening

The first time, at 2025-03-08 10:43:26.000, stopped forwarding values after a config change where BSKAFKA sent a message but it was never picked up by the forwarder (or at least the forwarder never printed a config change)

The second time, at 2025-03-19 09:43:32 HIFI changed config (presumably) and it never looked to pick up the values from the new monitors created.

We think we have recreated the second (and possibly what caused the first) one by:

• Start forwarder
• send forwarder config update and watch it send updates
• Kill off GWBLOCK
• Send forwarder new pair of subscribe messages (unsubscribe then subscribe to the same PVs as before)
• wait a while (120 seconds)
• Restart GWBLOCK
• See epics connection status (ep01) message, but no f144 value updates for any subsequent updates

Monitor updates do now not come in - this is a bug.

Fix is #8

To review:

1. Set up a redpanda instance following new docs on Add docs on running own kafka instance ibex_developers_manual#46
2. create a topic called forwarder_config by going to localhost:8080/topics
3. clone this repo and set up a venv etc. to run it
4. run the forwarder with forwarder-launch --config-topic localhost:19092/forwarder_config --status-topic localhost:19092/forwarder_status -v Trace --output-broker localhost:19092
5. Follow the instructions above to reproduce, first on main to make sure you can reproduce it, then afterwards on the branch in the PR to make sure it doesn't happen.

A script to send the forwarder config updates (note put your usual prefix and a block that exists, i have used p3 pointing at SIMPLE:P3:

from confluent_kafka import Producer
from streaming_data_types.forwarder_config_update_fc00 import (Protocol,
                                                               StreamInfo,
                                                               serialise_fc00)
from streaming_data_types.fbschemas.forwarder_config_update_fc00.UpdateType import UpdateType
from time import sleep


CONFIG_BROKER = "localhost:19092"

CONFIG_TOPIC = "forwarder_config"

STREAMS = [
    StreamInfo("XX:YY:CS:SB:p3", "f144", "NDW2932_sampleEnv", Protocol.Protocol.CA, 0),
]

producer = Producer({"bootstrap.servers": CONFIG_BROKER})

producer.produce(CONFIG_TOPIC, serialise_fc00(UpdateType.REMOVEALL, []))
producer.flush()

sleep(5)
# Add new streams
producer.produce(CONFIG_TOPIC, serialise_fc00(UpdateType.ADD, STREAMS))
producer.flush()

You can use https://github.com/ISISComputingGroup/saluki to deserialise the updates in machinename_sampleEnv.