class HDFSFileSink extends Sink<T> {

  public static <T> HDFSFileSink<T, AvroKey<T>, NullWritable> toAvro(String path,
                                                                     final AvroCoder<T> coder,
                                                                     Configuration conf) {
    SerializableFunction<T, KV<AvroKey<T>, NullWritable>> outputConverter =
        new SerializableFunction<T, KV<AvroKey<T>, NullWritable>>() {
          @Override
          public KV<AvroKey<T>, NullWritable> apply(T input) {
            return KV.of(new AvroKey<>(input), NullWritable.get());
          }
        };
    conf.set("avro.schema.output.key", coder.getSchema().toString());
    return to(
        path,
        AvroKeyOutputFormat.class,
        (Class<AvroKey<T>>) (Class<?>) AvroKey.class,
        NullWritable.class,
        outputConverter).withConfiguration(conf);
  }

}