This should be modelled on the existing Kafka integration.
This should be modelled on the existing Kafka integration.