RollingFileSystemSink implements MetricsSink, Closeable { @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); if (UserGroupInformation.isSecurityEnabled()) { checkIfPropertyExists(KEYTAB_PROPERTY_KEY); checkIfPropertyExists(USERNAME_PROPERTY_KEY); try { 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); } } } RollingFileSystemSink(); @VisibleForTesting protected  RollingFileSystemSink(long flushIntervalMillis,
      long flushOffsetIntervalMillis); @Override void init(SubsetConfiguration metrics2Properties); @Override void putMetrics(MetricsRecord record); @Override void flush(); @Override void close();  }
@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"); }