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 testSerializationAvailability() throws IOException {
    Configuration conf = new Configuration();
    Path path = new Path(GenericTestUtils.getTempPath(
        "serializationAvailability"));
    // Check if any serializers aren't found.
    try {
      SequenceFile.createWriter(
          conf,
          SequenceFile.Writer.file(path),
          SequenceFile.Writer.keyClass(String.class),
          SequenceFile.Writer.valueClass(NullWritable.class));
      // Note: This may also fail someday if JavaSerialization
      // is activated by default.
      fail("Must throw IOException for missing serializer for the Key class");
    } catch (IOException e) {
      assertTrue(e.getMessage().startsWith(
        "Could not find a serializer for the Key class: '" +
            String.class.getName() + "'."));
    }
    try {
      SequenceFile.createWriter(
          conf,
          SequenceFile.Writer.file(path),
          SequenceFile.Writer.keyClass(NullWritable.class),
          SequenceFile.Writer.valueClass(String.class));
      // Note: This may also fail someday if JavaSerialization
      // is activated by default.
      fail("Must throw IOException for missing serializer for the Value class");
    } catch (IOException e) {
      assertTrue(e.getMessage().startsWith(
        "Could not find a serializer for the Value class: '" +
            String.class.getName() + "'."));
    }

    // Write a simple file to test deserialization failures with
    writeTest(FileSystem.get(conf), 1, 1, path, CompressionType.NONE, null);

    // Remove Writable serializations, to enforce error.
    conf.setStrings(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY,
        AvroReflectSerialization.class.getName());

    // Now check if any deserializers aren't found.
    try {
      new SequenceFile.Reader(
          conf,
          SequenceFile.Reader.file(path));
      fail("Must throw IOException for missing deserializer for the Key class");
    } catch (IOException e) {
      assertTrue(e.getMessage().startsWith(
        "Could not find a deserializer for the Key class: '" +
            RandomDatum.class.getName() + "'."));
    }
  }