Function GetSink

worker/sink_handler.go:48–56

(conf *z.SuperFlag)

)

func GetSink(conf *z.SuperFlag) (Sink, error) {
	switch {
	case conf.GetString("kafka") != "":
		return newKafkaSink(conf)
	case conf.GetPath("file") != "":
		return newFileSink(conf)
	}
	return nil, errors.New("sink config is not provided")
}

// Kafka client is not concurrency safe.
// Its the responsibility of callee to manage the concurrency.