/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.hadoop.io;

import java.io.*;
import java.util.*;

import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.SequenceFile.Metadata;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.serializer.avro.AvroReflectSerialization;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.conf.*;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.fail;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import org.mockito.Mockito;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/** Support for flat files of binary key/value pairs. */
public class TestSequenceFile {
  private static final Logger LOG =
      LoggerFactory.getLogger(TestSequenceFile.class);

  private Configuration conf = new Configuration();

  /** Unit tests for SequenceFile. */
  @Test
  public void testZlibSequenceFile() throws Exception {
    LOG.info("Testing SequenceFile with DefaultCodec");
    compressedSeqFileTest(new DefaultCodec());
    LOG.info("Successfully tested SequenceFile with DefaultCodec");
  }

  @SuppressWarnings("deprecation")
  public void testSorterProperties() throws IOException {
    // Test to ensure that deprecated properties have no default
    // references anymore.
    Configuration config = new Configuration();
    assertNull("The deprecated sort memory property "
        + CommonConfigurationKeys.IO_SORT_MB_KEY
        + " must not exist in any core-*.xml files.",
        config.get(CommonConfigurationKeys.IO_SORT_MB_KEY));
    assertNull("The deprecated sort factor property "
        + CommonConfigurationKeys.IO_SORT_FACTOR_KEY
        + " must not exist in any core-*.xml files.",
        config.get(CommonConfigurationKeys.IO_SORT_FACTOR_KEY));

    // Test deprecated property honoring
    // Set different values for old and new property names
    // and compare which one gets loaded
    config = new Configuration();
    FileSystem fs = FileSystem.get(config);
    config.setInt(CommonConfigurationKeys.IO_SORT_MB_KEY, 10);
    config.setInt(CommonConfigurationKeys.IO_SORT_FACTOR_KEY, 10);
    config.setInt(CommonConfigurationKeys.SEQ_IO_SORT_MB_KEY, 20);
    config.setInt(CommonConfigurationKeys.SEQ_IO_SORT_FACTOR_KEY, 20);
    SequenceFile.Sorter sorter = new SequenceFile.Sorter(
        fs, Text.class, Text.class, config);
    assertEquals("Deprecated memory conf must be honored over newer property",
        10*1024*1024, sorter.getMemory());
    assertEquals("Deprecated factor conf must be honored over newer property",
        10, sorter.getFactor());

    // Test deprecated properties (graceful deprecation)
    config = new Configuration();
    fs = FileSystem.get(config);
    config.setInt(CommonConfigurationKeys.IO_SORT_MB_KEY, 10);
    config.setInt(CommonConfigurationKeys.IO_SORT_FACTOR_KEY, 10);
    sorter = new SequenceFile.Sorter(
        fs, Text.class, Text.class, config);
    assertEquals("Deprecated memory property "
        + CommonConfigurationKeys.IO_SORT_MB_KEY
        + " must get properly applied.",
        10*1024*1024, // In bytes
        sorter.getMemory());
    assertEquals("Deprecated sort factor property "
        + CommonConfigurationKeys.IO_SORT_FACTOR_KEY
        + " must get properly applied.",
        10, sorter.getFactor());

    // Test regular properties (graceful deprecation)
    config = new Configuration();
    fs = FileSystem.get(config);
    config.setInt(CommonConfigurationKeys.SEQ_IO_SORT_MB_KEY, 20);
    config.setInt(CommonConfigurationKeys.SEQ_IO_SORT_FACTOR_KEY, 20);
    sorter = new SequenceFile.Sorter(
        fs, Text.class, Text.class, config);
    assertEquals("Memory property "
        + CommonConfigurationKeys.SEQ_IO_SORT_MB_KEY
        + " must get properly applied if present.",
        20*1024*1024, // In bytes
        sorter.getMemory());
    assertEquals("Merge factor property "
        + CommonConfigurationKeys.SEQ_IO_SORT_FACTOR_KEY
        + " must get properly applied if present.",
        20, sorter.getFactor());
  }

  public void compressedSeqFileTest(CompressionCodec codec) throws Exception {
    int count = 1024 * 10;
    int megabytes = 1;
    int factor = 5;
    Path file = new Path(GenericTestUtils.getTempPath("test.seq"));
    Path recordCompressedFile = new Path(GenericTestUtils.getTempPath(
        "test.rc.seq"));
    Path blockCompressedFile = new Path(GenericTestUtils.getTempPath(
        "test.bc.seq"));
 
    int seed = new Random().nextInt();
    LOG.info("Seed = " + seed);

    FileSystem fs = FileSystem.getLocal(conf);
    try {
      // SequenceFile.Writer
      writeTest(fs, count, seed, file, CompressionType.NONE, null);
      readTest(fs, count, seed, file);

      sortTest(fs, count, megabytes, factor, false, file);
      checkSort(fs, count, seed, file);

      sortTest(fs, count, megabytes, factor, true, file);
      checkSort(fs, count, seed, file);

      mergeTest(fs, count, seed, file, CompressionType.NONE, false, 
                factor, megabytes);
      checkSort(fs, count, seed, file);

      mergeTest(fs, count, seed, file, CompressionType.NONE, true, 
                factor, megabytes);
      checkSort(fs, count, seed, file);
        
      // SequenceFile.RecordCompressWriter
      writeTest(fs, count, seed, recordCompressedFile, CompressionType.RECORD, 
                codec);
      readTest(fs, count, seed, recordCompressedFile);

      sortTest(fs, count, megabytes, factor, false, recordCompressedFile);
      checkSort(fs, count, seed, recordCompressedFile);

      sortTest(fs, count, megabytes, factor, true, recordCompressedFile);
      checkSort(fs, count, seed, recordCompressedFile);

      mergeTest(fs, count, seed, recordCompressedFile, 
                CompressionType.RECORD, false, factor, megabytes);
      checkSort(fs, count, seed, recordCompressedFile);

      mergeTest(fs, count, seed, recordCompressedFile, 
                CompressionType.RECORD, true, factor, megabytes);
      checkSort(fs, count, seed, recordCompressedFile);
        
      // SequenceFile.BlockCompressWriter
      writeTest(fs, count, seed, blockCompressedFile, CompressionType.BLOCK,
                codec);
      readTest(fs, count, seed, blockCompressedFile);

      sortTest(fs, count, megabytes, factor, false, blockCompressedFile);
      checkSort(fs, count, seed, blockCompressedFile);

      sortTest(fs, count, megabytes, factor, true, blockCompressedFile);
      checkSort(fs, count, seed, blockCompressedFile);

      mergeTest(fs, count, seed, blockCompressedFile, CompressionType.BLOCK, 
                false, factor, megabytes);
      checkSort(fs, count, seed, blockCompressedFile);

      mergeTest(fs, count, seed, blockCompressedFile, CompressionType.BLOCK, 
                true, factor, megabytes);
      checkSort(fs, count, seed, blockCompressedFile);

    } finally {
      fs.close();
    }
  }

  @SuppressWarnings("deprecation")
  private void writeTest(FileSystem fs, int count, int seed, Path file, 
                                CompressionType compressionType, CompressionCodec codec)
    throws IOException {
    fs.delete(file, true);
    LOG.info("creating " + count + " records with " + compressionType +
             " compression");
    SequenceFile.Writer writer = 
      SequenceFile.createWriter(fs, conf, file, 
                                RandomDatum.class, RandomDatum.class, compressionType, codec);
    RandomDatum.Generator generator = new RandomDatum.Generator(seed);
    for (int i = 0; i < count; i++) {
      generator.next();
      RandomDatum key = generator.getKey();
      RandomDatum value = generator.getValue();

      writer.append(key, value);
    }
    writer.close();
  }

  @SuppressWarnings("deprecation")
  private void readTest(FileSystem fs, int count, int seed, Path file)
    throws IOException {
    LOG.debug("reading " + count + " records");
    SequenceFile.Reader reader = new SequenceFile.Reader(fs, file, conf);
    RandomDatum.Generator generator = new RandomDatum.Generator(seed);

    RandomDatum k = new RandomDatum();
    RandomDatum v = new RandomDatum();
    DataOutputBuffer rawKey = new DataOutputBuffer();
    SequenceFile.ValueBytes rawValue = reader.createValueBytes();
    
    for (int i = 0; i < count; i++) {
      generator.next();
      RandomDatum key = generator.getKey();
      RandomDatum value = generator.getValue();

      try {
        if ((i%5) == 0) {
          // Testing 'raw' apis
          rawKey.reset();
          reader.nextRaw(rawKey, rawValue);
        } else {
          // Testing 'non-raw' apis 
          if ((i%2) == 0) {
            reader.next(k);
            reader.getCurrentValue(v);
          } else {
            reader.next(k, v);
          }
          
          // Check
          if (!k.equals(key))
            throw new RuntimeException("wrong key at " + i);
          if (!v.equals(value))
            throw new RuntimeException("wrong value at " + i);
        }
      } catch (IOException ioe) {
        LOG.info("Problem on row " + i);
        LOG.info("Expected key = " + key);
        LOG.info("Expected len = " + key.getLength());
        LOG.info("Actual key = " + k);
        LOG.info("Actual len = " + k.getLength());
        LOG.info("Expected value = " + value);
        LOG.info("Expected len = " + value.getLength());
        LOG.info("Actual value = " + v);
        LOG.info("Actual len = " + v.getLength());
        LOG.info("Key equals: " + k.equals(key));
        LOG.info("value equals: " + v.equals(value));
        throw ioe;
      }

    }
    reader.close();
  }


  private void sortTest(FileSystem fs, int count, int megabytes, 
                               int factor, boolean fast, Path file)
    throws IOException {
    fs.delete(new Path(file+".sorted"), true);
    SequenceFile.Sorter sorter = newSorter(fs, fast, megabytes, factor);
    LOG.debug("sorting " + count + " records");
    sorter.sort(file, file.suffix(".sorted"));
    LOG.info("done sorting " + count + " debug");
  }

  @SuppressWarnings("deprecation")
  private void checkSort(FileSystem fs, int count, int seed, Path file)
    throws IOException {
    LOG.info("sorting " + count + " records in memory for debug");
    RandomDatum.Generator generator = new RandomDatum.Generator(seed);
    SortedMap<RandomDatum, RandomDatum> map =
      new TreeMap<RandomDatum, RandomDatum>();
    for (int i = 0; i < count; i++) {
      generator.next();
      RandomDatum key = generator.getKey();
      RandomDatum value = generator.getValue();
      map.put(key, value);
    }

    LOG.debug("checking order of " + count + " records");
    RandomDatum k = new RandomDatum();
    RandomDatum v = new RandomDatum();
    Iterator<Map.Entry<RandomDatum, RandomDatum>> iterator =
      map.entrySet().iterator();
    SequenceFile.Reader reader =
      new SequenceFile.Reader(fs, file.suffix(".sorted"), conf);
    for (int i = 0; i < count; i++) {
      Map.Entry<RandomDatum, RandomDatum> entry = iterator.next();
      RandomDatum key = entry.getKey();
      RandomDatum value = entry.getValue();

      reader.next(k, v);

      if (!k.equals(key))
        throw new RuntimeException("wrong key at " + i);
      if (!v.equals(value))
        throw new RuntimeException("wrong value at " + i);
    }

    reader.close();
    LOG.debug("sucessfully checked " + count + " records");
  }

  @SuppressWarnings("deprecation")
  private void mergeTest(FileSystem fs, int count, int seed, Path file, 
                                CompressionType compressionType,
                                boolean fast, int factor, int megabytes)
    throws IOException {

    LOG.debug("creating "+factor+" files with "+count/factor+" records");

    SequenceFile.Writer[] writers = new SequenceFile.Writer[factor];
    Path[] names = new Path[factor];
    Path[] sortedNames = new Path[factor];
    
    for (int i = 0; i < factor; i++) {
      names[i] = file.suffix("."+i);
      sortedNames[i] = names[i].suffix(".sorted");
      fs.delete(names[i], true);
      fs.delete(sortedNames[i], true);
      writers[i] = SequenceFile.createWriter(fs, conf, names[i], 
                                             RandomDatum.class, RandomDatum.class, compressionType);
    }

    RandomDatum.Generator generator = new RandomDatum.Generator(seed);

    for (int i = 0; i < count; i++) {
      generator.next();
      RandomDatum key = generator.getKey();
      RandomDatum value = generator.getValue();

      writers[i%factor].append(key, value);
    }

    for (int i = 0; i < factor; i++)
      writers[i].close();

    for (int i = 0; i < factor; i++) {
      LOG.debug("sorting file " + i + " with " + count/factor + " records");
      newSorter(fs, fast, megabytes, factor).sort(names[i], sortedNames[i]);
    }

    LOG.info("merging " + factor + " files with " + count/factor + " debug");
    fs.delete(new Path(file+".sorted"), true);
    newSorter(fs, fast, megabytes, factor)
      .merge(sortedNames, file.suffix(".sorted"));
  }

  private SequenceFile.Sorter newSorter(FileSystem fs, 
                                               boolean fast,
                                               int megabytes, int factor) {
    SequenceFile.Sorter sorter = 
      fast
      ? new SequenceFile.Sorter(fs, new RandomDatum.Comparator(),
                                RandomDatum.class, RandomDatum.class, conf)
      : new SequenceFile.Sorter(fs, RandomDatum.class, RandomDatum.class, conf);
    sorter.setMemory(megabytes * 1024*1024);
    sorter.setFactor(factor);
    return sorter;
  }

  /** Unit tests for SequenceFile metadata. */
  @Test
  public void testSequenceFileMetadata() throws Exception {
    LOG.info("Testing SequenceFile with metadata");
    int count = 1024 * 10;
    CompressionCodec codec = new DefaultCodec();
    Path file = new Path(GenericTestUtils.getTempPath("test.seq.metadata"));
    Path sortedFile = new Path(GenericTestUtils.getTempPath(
        "test.sorted.seq.metadata"));
    Path recordCompressedFile = new Path(GenericTestUtils.getTempPath(
        "test.rc.seq.metadata"));
    Path blockCompressedFile = new Path(GenericTestUtils.getTempPath(
        "test.bc.seq.metadata"));
 
    FileSystem fs = FileSystem.getLocal(conf);
    SequenceFile.Metadata theMetadata = new SequenceFile.Metadata();
    theMetadata.set(new Text("name_1"), new Text("value_1"));
    theMetadata.set(new Text("name_2"), new Text("value_2"));
    theMetadata.set(new Text("name_3"), new Text("value_3"));
    theMetadata.set(new Text("name_4"), new Text("value_4"));
    
    int seed = new Random().nextInt();
    
    try {
      // SequenceFile.Writer
      writeMetadataTest(fs, count, seed, file, CompressionType.NONE, null, theMetadata);
      SequenceFile.Metadata aMetadata = readMetadata(fs, file);
      if (!theMetadata.equals(aMetadata)) {
        LOG.info("The original metadata:\n" + theMetadata.toString());
        LOG.info("The retrieved metadata:\n" + aMetadata.toString());
        throw new RuntimeException("metadata not match:  " + 1);
      }
      // SequenceFile.RecordCompressWriter
      writeMetadataTest(fs, count, seed, recordCompressedFile, CompressionType.RECORD, 
                        codec, theMetadata);
      aMetadata = readMetadata(fs, recordCompressedFile);
      if (!theMetadata.equals(aMetadata)) {
        LOG.info("The original metadata:\n" + theMetadata.toString());
        LOG.info("The retrieved metadata:\n" + aMetadata.toString());
        throw new RuntimeException("metadata not match:  " + 2);
      }
      // SequenceFile.BlockCompressWriter
      writeMetadataTest(fs, count, seed, blockCompressedFile, CompressionType.BLOCK,
                        codec, theMetadata);
      aMetadata =readMetadata(fs, blockCompressedFile);
      if (!theMetadata.equals(aMetadata)) {
        LOG.info("The original metadata:\n" + theMetadata.toString());
        LOG.info("The retrieved metadata:\n" + aMetadata.toString());
        throw new RuntimeException("metadata not match:  " + 3);
      }
      // SequenceFile.Sorter
      sortMetadataTest(fs, file, sortedFile, theMetadata);
      aMetadata = readMetadata(fs, recordCompressedFile);
      if (!theMetadata.equals(aMetadata)) {
        LOG.info("The original metadata:\n" + theMetadata.toString());
        LOG.info("The retrieved metadata:\n" + aMetadata.toString());
        throw new RuntimeException("metadata not match:  " + 4);
      }
    } finally {
      fs.close();
    }
    LOG.info("Successfully tested SequenceFile with metadata");
  }
  
  
  @SuppressWarnings("deprecation")
  private SequenceFile.Metadata readMetadata(FileSystem fs, Path file)
    throws IOException {
    LOG.info("reading file: " + file.toString());
    SequenceFile.Reader reader = new SequenceFile.Reader(fs, file, conf);
    SequenceFile.Metadata meta = reader.getMetadata(); 
    reader.close();
    return meta;
  }

  @SuppressWarnings("deprecation")
  private void writeMetadataTest(FileSystem fs, int count, int seed, Path file, 
                                        CompressionType compressionType, CompressionCodec codec, SequenceFile.Metadata metadata)
    throws IOException {
    fs.delete(file, true);
    LOG.info("creating " + count + " records with metadata and with " + compressionType +
             " compression");
    SequenceFile.Writer writer = 
      SequenceFile.createWriter(fs, conf, file, 
                                RandomDatum.class, RandomDatum.class, compressionType, codec, null, metadata);
    RandomDatum.Generator generator = new RandomDatum.Generator(seed);
    for (int i = 0; i < count; i++) {
      generator.next();
      RandomDatum key = generator.getKey();
      RandomDatum value = generator.getValue();

      writer.append(key, value);
    }
    writer.close();
  }

  private void sortMetadataTest(FileSystem fs, Path unsortedFile, Path sortedFile, SequenceFile.Metadata metadata)
    throws IOException {
    fs.delete(sortedFile, true);
    LOG.info("sorting: " + unsortedFile + " to: " + sortedFile);
    final WritableComparator comparator = WritableComparator.get(RandomDatum.class);
    SequenceFile.Sorter sorter = new SequenceFile.Sorter(fs, comparator, RandomDatum.class, RandomDatum.class, conf, metadata);
    sorter.sort(new Path[] { unsortedFile }, sortedFile, false);
  }

  @SuppressWarnings("deprecation")
  @Test
  public void testClose() throws IOException {
    Configuration conf = new Configuration();
    LocalFileSystem fs = FileSystem.getLocal(conf);
  
    // create a sequence file 1
    Path path1 = new Path(GenericTestUtils.getTempPath("test1.seq"));
    SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, path1,
        Text.class, NullWritable.class, CompressionType.BLOCK);
    writer.append(new Text("file1-1"), NullWritable.get());
    writer.append(new Text("file1-2"), NullWritable.get());
    writer.close();
  
    Path path2 = new Path(GenericTestUtils.getTempPath("test2.seq"));
    writer = SequenceFile.createWriter(fs, conf, path2, Text.class,
        NullWritable.class, CompressionType.BLOCK);
    writer.append(new Text("file2-1"), NullWritable.get());
    writer.append(new Text("file2-2"), NullWritable.get());
    writer.close();
  
    // Create a reader which uses 4 BuiltInZLibInflater instances
    SequenceFile.Reader reader = new SequenceFile.Reader(fs, path1, conf);
    // Returns the 4 BuiltInZLibInflater instances to the CodecPool
    reader.close();
    // The second close _could_ erroneously returns the same 
    // 4 BuiltInZLibInflater instances to the CodecPool again
    reader.close();
  
    // The first reader gets 4 BuiltInZLibInflater instances from the CodecPool
    SequenceFile.Reader reader1 = new SequenceFile.Reader(fs, path1, conf);
    // read first value from reader1
    Text text = new Text();
    reader1.next(text);
    assertEquals("file1-1", text.toString());
    
    // The second reader _could_ get the same 4 BuiltInZLibInflater 
    // instances from the CodePool as reader1
    SequenceFile.Reader reader2 = new SequenceFile.Reader(fs, path2, conf);
    
    // read first value from reader2
    reader2.next(text);
    assertEquals("file2-1", text.toString());
    // read second value from reader1
    reader1.next(text);
    assertEquals("file1-2", text.toString());
    // read second value from reader2 (this throws an exception)
    reader2.next(text);
    assertEquals("file2-2", text.toString());
  
    assertFalse(reader1.next(text));
    assertFalse(reader2.next(text));
  }

  /**
   * Test that makes sure the FileSystem passed to createWriter
   * @throws Exception
   */
  @SuppressWarnings("deprecation")
  @Test
  public void testCreateUsesFsArg() throws Exception {
    FileSystem fs = FileSystem.getLocal(conf);
    FileSystem spyFs = Mockito.spy(fs);
    Path p = new Path(GenericTestUtils.getTempPath("testCreateUsesFSArg.seq"));
    SequenceFile.Writer writer = SequenceFile.createWriter(
        spyFs, conf, p, NullWritable.class, NullWritable.class);
    writer.close();
    Mockito.verify(spyFs).getDefaultReplication(p);
  }

  private static class TestFSDataInputStream extends FSDataInputStream {
    private boolean closed = false;

    private TestFSDataInputStream(InputStream in) throws IOException {
      super(in);
    }

    @Override
    public void close() throws IOException {
      closed = true;
      super.close();
    }

    public boolean isClosed() {
      return closed;
    }
  }

  @SuppressWarnings("deprecation")
  @Test
  public void testCloseForErroneousSequenceFile()
    throws IOException {
    Configuration conf = new Configuration();
    LocalFileSystem fs = FileSystem.getLocal(conf);

    // create an empty file (which is not a valid sequence file)
    Path path = new Path(GenericTestUtils.getTempPath("broken.seq"));
    fs.create(path).close();

    // try to create SequenceFile.Reader
    final TestFSDataInputStream[] openedFile = new TestFSDataInputStream[1];
    try {
      new SequenceFile.Reader(fs, path, conf) {
        // this method is called by the SequenceFile.Reader constructor, overwritten, so we can access the opened file
        @Override
        protected FSDataInputStream openFile(FileSystem fs, Path file, int bufferSize, long length) throws IOException {
          final InputStream in = super.openFile(fs, file, bufferSize, length);
          openedFile[0] = new TestFSDataInputStream(in);
          return openedFile[0];
        }
      };
      fail("IOException expected.");
    } catch (IOException expected) {}

    assertNotNull(path + " should have been opened.", openedFile[0]);
    assertTrue("InputStream for " + path + " should have been closed.", openedFile[0].isClosed());
  }

  /**
   * Test to makes sure zero length sequence file is handled properly while
   * initializing.
   */
  @Test
  public void testInitZeroLengthSequenceFile() throws IOException {
    Configuration conf = new Configuration();
    LocalFileSystem fs = FileSystem.getLocal(conf);

    // create an empty file (which is not a valid sequence file)
    Path path = new Path(GenericTestUtils.getTempPath("zerolength.seq"));
    fs.create(path).close();

    try {
      new SequenceFile.Reader(conf, SequenceFile.Reader.file(path));
      fail("IOException expected.");
    } catch (IOException expected) {
      assertTrue(expected instanceof EOFException);
    }
  }

   /**
   * Test that makes sure createWriter succeeds on a file that was 
   * already created
   * @throws IOException
   */
  @SuppressWarnings("deprecation")
  @Test
  public void testCreateWriterOnExistingFile() throws IOException {
    Configuration conf = new Configuration();
    FileSystem fs = FileSystem.getLocal(conf);
    Path name = new Path(new Path(GenericTestUtils.getTempPath(
        "createWriterOnExistingFile")), "file");

    fs.create(name);
    SequenceFile.createWriter(fs, conf, name, RandomDatum.class,
        RandomDatum.class, 512, (short) 1, 4096, false,
        CompressionType.NONE, null, new Metadata());
  }

  @SuppressWarnings("deprecation")
  @Test
  public void testRecursiveSeqFileCreate() throws IOException {
    FileSystem fs = FileSystem.getLocal(conf);
    Path name = new Path(new Path(GenericTestUtils.getTempPath(
        "recursiveCreateDir")), "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
    }

    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
  }

  @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() + "'."));
    }
  }

  /** For debugging and testing. */
  public static void main(String[] args) throws Exception {
    int count = 1024 * 1024;
    int megabytes = 1;
    int factor = 10;
    boolean create = true;
    boolean rwonly = false;
    boolean check = false;
    boolean fast = false;
    boolean merge = false;
    String compressType = "NONE";
    String compressionCodec = "org.apache.hadoop.io.compress.DefaultCodec";
    Path file = null;
    int seed = new Random().nextInt();

    String usage = "Usage: testsequencefile " +
      "[-count N] " + 
      "[-seed #] [-check] [-compressType <NONE|RECORD|BLOCK>] " + 
      "-codec <compressionCodec> " + 
      "[[-rwonly] | {[-megabytes M] [-factor F] [-nocreate] [-fast] [-merge]}] " +
      " file";
    if (args.length == 0) {
      System.err.println(usage);
      System.exit(-1);
    }
    
    FileSystem fs = null;
    try {
      for (int i=0; i < args.length; ++i) {       // parse command line
        if (args[i] == null) {
          continue;
        } else if (args[i].equals("-count")) {
          count = Integer.parseInt(args[++i]);
        } else if (args[i].equals("-megabytes")) {
          megabytes = Integer.parseInt(args[++i]);
        } else if (args[i].equals("-factor")) {
          factor = Integer.parseInt(args[++i]);
        } else if (args[i].equals("-seed")) {
          seed = Integer.parseInt(args[++i]);
        } else if (args[i].equals("-rwonly")) {
          rwonly = true;
        } else if (args[i].equals("-nocreate")) {
          create = false;
        } else if (args[i].equals("-check")) {
          check = true;
        } else if (args[i].equals("-fast")) {
          fast = true;
        } else if (args[i].equals("-merge")) {
          merge = true;
        } else if (args[i].equals("-compressType")) {
          compressType = args[++i];
        } else if (args[i].equals("-codec")) {
          compressionCodec = args[++i];
        } else {
          // file is required parameter
          file = new Path(args[i]);
        }
      }
        
      TestSequenceFile test = new TestSequenceFile();
      
      fs = file.getFileSystem(test.conf);

      LOG.info("count = " + count);
      LOG.info("megabytes = " + megabytes);
      LOG.info("factor = " + factor);
      LOG.info("create = " + create);
      LOG.info("seed = " + seed);
      LOG.info("rwonly = " + rwonly);
      LOG.info("check = " + check);
      LOG.info("fast = " + fast);
      LOG.info("merge = " + merge);
      LOG.info("compressType = " + compressType);
      LOG.info("compressionCodec = " + compressionCodec);
      LOG.info("file = " + file);

      if (rwonly && (!create || merge || fast)) {
        System.err.println(usage);
        System.exit(-1);
      }

      CompressionType compressionType = 
        CompressionType.valueOf(compressType);
      CompressionCodec codec = (CompressionCodec)ReflectionUtils.newInstance(
                                                                             test.conf.getClassByName(compressionCodec), 
                                                                             test.conf);

      if (rwonly || (create && !merge)) {
        test.writeTest(fs, count, seed, file, compressionType, codec);
        test.readTest(fs, count, seed, file);
      }

      if (!rwonly) {
        if (merge) {
          test.mergeTest(fs, count, seed, file, compressionType, 
                    fast, factor, megabytes);
        } else {
          test.sortTest(fs, count, megabytes, factor, fast, file);
        }
      }
    
      if (check) {
        test.checkSort(fs, count, seed, file);
      }
    } finally {
      if (fs != null) {
        fs.close();
      }
    }
  }
}