synchronized MetricsSink registerSink(String name, String desc, MetricsSink sink) {
        Contracts.checkNotNull(config, "config");
        MetricsSinkAdapter sa = sinks.get(name);
        if (sa != null) {
            LOG.warn("Sink name " + name + " already exists!");
            return sa.sink();
        }
        MetricsConfig conf = sinkConfigs.get(name);
        sa = conf != null ? newSink(name, desc, sink, conf) : newSink(name, desc, sink,
                                                                      config.subset(MetricsConfig.SINK_KEY));
        sinks.put(name, sa);
        sa.start();
        LOG.debug("Registered sink " + name);
        return sa.sink();
    }