Generalised some subscription kafka stuff so it can be used outside of the subscription stuff