CustomDfsSink extends EventSink.Base { public static CompressionCodec getCodec(Configuration conf, String codecName) { List<Class<? extends CompressionCodec>> codecs = CompressionCodecFactory .getCodecClasses(FlumeConfiguration.get()); CompressionCodec codec = null; ArrayList<String> codecStrs = new ArrayList<String>(); codecStrs.add("None"); for (Class<? extends CompressionCodec> cls : codecs) { codecStrs.add(cls.getSimpleName()); if (codecMatches(cls, codecName)) { try { codec = cls.newInstance(); } catch (InstantiationException e) { LOG.error("Unable to instantiate " + cls + " class"); } catch (IllegalAccessException e) { LOG.error("Unable to access " + cls + " class"); } } } if (codec == null) { if (!codecName.equalsIgnoreCase("None")) { throw new IllegalArgumentException("Unsupported compression codec " + codecName + ". Please choose from: " + codecStrs); } } else if (codec instanceof Configurable) { ((Configurable) codec).setConf(conf); } return codec; } CustomDfsSink(String path, OutputFormat format); @Override void append(Event e); @Override void close(); @Override void open(); static CompressionCodec getCodec(Configuration conf, String codecName); static SinkBuilder builder(); @Override String getName(); @Override ReportEvent getMetrics(); @Deprecated @Override ReportEvent getReport();  }
@Test public void testGetCodec() { checkCodec(GzipCodec.class, "gzip"); checkCodec(GzipCodec.class, "Gzip"); checkCodec(GzipCodec.class, "gzipcodec"); checkCodec(GzipCodec.class, "GzipCodec"); checkCodec(GzipCodec.class, "org.apache.hadoop.io.compress.GzipCodec"); }