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);
    }
  }
@Test
  public void testSerializationUsingWritableNameAlias() throws IOException {
    Configuration config = new Configuration();
    config.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, SimpleSerializer.class.getName());
    Path path = new Path(System.getProperty("test.build.data", "."),
        "SerializationUsingWritableNameAlias");

    // write with the original serializable class
    SequenceFile.Writer writer = SequenceFile.createWriter(
        config,
        SequenceFile.Writer.file(path),
        SequenceFile.Writer.keyClass(SimpleSerializable.class),
        SequenceFile.Writer.valueClass(SimpleSerializable.class));

    int max = 10;
    try {
      SimpleSerializable val = new SimpleSerializable();
      val.setId(-1);
      for (int i = 0; i < max; i++) {
        SimpleSerializable key = new SimpleSerializable();
        key.setId(i);
        writer.append(key, val);
      }
    } finally {
      writer.close();
    }

    // override name so it gets forced to the new serializable
    WritableName.setName(AnotherSimpleSerializable.class, SimpleSerializable.class.getName());

    // read and expect our new serializable, and all the correct values read
    SequenceFile.Reader reader = new SequenceFile.Reader(
        config,
        SequenceFile.Reader.file(path));

    AnotherSimpleSerializable key = new AnotherSimpleSerializable();
    int count = 0;
    while (true) {
      key = (AnotherSimpleSerializable) reader.next(key);
      if (key == null) {
        // make sure we exhausted all the ints we wrote
        assertEquals(count, max);
        break;
      }
      assertEquals(count++, key.getId());
    }
  }