modified:   consumer/log_sink_manager.go