refactored out of logstash.broker state