public static Writer createWriter(Configuration conf, Writer.Option... opts
                                    ) throws IOException {
    Writer.CompressionOption compressionOption = 
      Options.getOption(Writer.CompressionOption.class, opts);
    CompressionType kind;
    if (compressionOption != null) {
      kind = compressionOption.getValue();
    } else {
      kind = getDefaultCompressionType(conf);
      opts = Options.prependOptions(opts, Writer.compression(kind));
    }
    switch (kind) {
      default:
      case NONE:
        return new Writer(conf, opts);
      case RECORD:
        return new RecordCompressWriter(conf, opts);
      case BLOCK:
        return new BlockCompressWriter(conf, opts);
    }
  }
@SuppressWarnings("deprecation")
  @Test
  public void testRecursiveSeqFileCreate() throws IOException {
    FileSystem fs = FileSystem.getLocal(conf);
    Path parentDir = new Path(GenericTestUtils.getTempPath(
            "recursiveCreateDir"));
    Path name = new Path(parentDir, "file");
    boolean createParent = false;

    try {
      SequenceFile.createWriter(fs, conf, name, RandomDatum.class,
          RandomDatum.class, 512, (short) 1, 4096, createParent,
          CompressionType.NONE, null, new Metadata());
      fail("Expected an IOException due to missing parent");
    } catch (IOException ioe) {
      // Expected
    }

    try {
      createParent = true;
      SequenceFile.createWriter(fs, conf, name, RandomDatum.class,
          RandomDatum.class, 512, (short) 1, 4096, createParent,
          CompressionType.NONE, null, new Metadata());
      // should succeed, fails if exception thrown
    } finally {
      fs.deleteOnExit(parentDir);
      fs.close();
    }
  }