@Override
  public void init(SubsetConfiguration metrics2Properties) {
    properties = metrics2Properties;
    basePath = new Path(properties.getString(BASEPATH_KEY, BASEPATH_DEFAULT));
    source = properties.getString(SOURCE_KEY, SOURCE_DEFAULT);
    ignoreError = properties.getBoolean(IGNORE_ERROR_KEY, DEFAULT_IGNORE_ERROR);
    allowAppend = properties.getBoolean(ALLOW_APPEND_KEY, DEFAULT_ALLOW_APPEND);
    rollOffsetIntervalMillis =
        getNonNegative(ROLL_OFFSET_INTERVAL_MILLIS_KEY,
          DEFAULT_ROLL_OFFSET_INTERVAL_MILLIS);
    rollIntervalMillis = getRollInterval();

    conf = loadConf();
    UserGroupInformation.setConfiguration(conf);

    // Don't do secure setup if it's not needed.
    if (UserGroupInformation.isSecurityEnabled()) {
      // Validate config so that we don't get an NPE
      checkIfPropertyExists(KEYTAB_PROPERTY_KEY);
      checkIfPropertyExists(USERNAME_PROPERTY_KEY);


      try {
        // Login as whoever we're supposed to be and let the hostname be pulled
        // from localhost. If security isn't enabled, this does nothing.
        SecurityUtil.login(conf, properties.getString(KEYTAB_PROPERTY_KEY),
            properties.getString(USERNAME_PROPERTY_KEY));
      } catch (IOException ex) {
        throw new MetricsException("Error logging in securely: ["
            + ex.toString() + "]", ex);
      }
    }
  }
@Test
  public void testInit() {
    ConfigBuilder builder = new ConfigBuilder();
    SubsetConfiguration conf =
        builder.add("sink.roll-interval", "10m")
            .add("sink.roll-offset-interval-millis", "1")
            .add("sink.basepath", "path")
            .add("sink.ignore-error", "true")
            .add("sink.allow-append", "true")
            .add("sink.source", "src")
            .subset("sink");

    RollingFileSystemSink sink = new RollingFileSystemSink();

    sink.init(conf);

    assertEquals("The roll interval was not set correctly",
        sink.rollIntervalMillis, 600000);
    assertEquals("The roll offset interval was not set correctly",
        sink.rollOffsetIntervalMillis, 1);
    assertEquals("The base path was not set correctly",
        sink.basePath, new Path("path"));
    assertEquals("ignore-error was not set correctly",
        sink.ignoreError, true);
    assertEquals("allow-append was not set correctly",
        sink.allowAppend, true);
    assertEquals("The source was not set correctly",
        sink.source, "src");
  }