/**
    *
    * 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.hbase.regionserver.wal;
  
  import java.io.FilterInputStream;
  import java.io.IOException;
  import java.lang.reflect.Field;
  import java.lang.reflect.Method;
  import java.util.NavigableMap;
  
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.hadoop.classification.InterfaceAudience;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.fs.FSDataInputStream;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.hbase.regionserver.wal.HLog.Entry;
  import org.apache.hadoop.io.SequenceFile;
  import org.apache.hadoop.io.Text;
  import org.apache.hadoop.io.SequenceFile.Metadata;
  
  @InterfaceAudience.Private
  public class SequenceFileLogReader extends ReaderBase {
    private static final Log LOG = LogFactory.getLog(SequenceFileLogReader.class);
  
    // Legacy stuff from pre-PB WAL metadata.
    private static final Text WAL_VERSION_KEY = new Text("version");
    // Let the version be 1.  Let absence of a version meta tag be old, version 0.
    // Set this version '1' to be the version that introduces compression,
    // the COMPRESSION_VERSION.
    private static final int COMPRESSION_VERSION = 1;
    private static final Text WAL_COMPRESSION_TYPE_KEY = new Text("compression.type");
    private static final Text DICTIONARY_COMPRESSION_TYPE = new Text("dictionary");
  
    /**
     * Hack just to set the correct file length up in SequenceFile.Reader.
     * See HADOOP-6307.  The below is all about setting the right length on the
     * file we are reading.  fs.getFileStatus(file).getLen() is passed down to
     * a private SequenceFile.Reader constructor.  This won't work.  Need to do
     * the available on the stream.  The below is ugly.  It makes getPos, the
     * first time its called, return length of the file -- i.e. tell a lie -- just
     * so this line up in SF.Reader's constructor ends up with right answer:
     *
     *         this.end = in.getPos() + length;
     *
     */
    private static class WALReader extends SequenceFile.Reader {
  
      WALReader(final FileSystem fs, final Path p, final Configuration c)
      throws IOException {
        super(fs, p, c);
      }
  
      @Override
      protected FSDataInputStream openFile(FileSystem fs, Path file,
        int bufferSize, long length)
      throws IOException {
        return new WALReaderFSDataInputStream(super.openFile(fs, file,
          bufferSize, length), length);
      }
  
      /**
       * Override just so can intercept first call to getPos.
       */
      static class WALReaderFSDataInputStream extends FSDataInputStream {
        private boolean firstGetPosInvocation = true;
        private long length;
  
        WALReaderFSDataInputStream(final FSDataInputStream is, final long l)
        throws IOException {
          super(is);
          this.length = l;
        }
  
        // This section can be confusing.  It is specific to how HDFS works.
        // Let me try to break it down.  This is the problem:
        //
        //  1. HDFS DataNodes update the NameNode about a filename's length
        //     on block boundaries or when a file is closed. Therefore,
        //     if an RS dies, then the NN's fs.getLength() can be out of date
       //  2. this.in.available() would work, but it returns int &
       //     therefore breaks for files > 2GB (happens on big clusters)
       //  3. DFSInputStream.getFileLength() gets the actual length from the DNs
       //  4. DFSInputStream is wrapped 2 levels deep : this.in.in
       //
       // So, here we adjust getPos() using getFileLength() so the
       // SequenceFile.Reader constructor (aka: first invocation) comes out
       // with the correct end of the file:
       //         this.end = in.getPos() + length;
       @Override
       public long getPos() throws IOException {
         if (this.firstGetPosInvocation) {
           this.firstGetPosInvocation = false;
           long adjust = 0;
 
           try {
             Field fIn = FilterInputStream.class.getDeclaredField("in");
             fIn.setAccessible(true);
             Object realIn = fIn.get(this.in);
             // In hadoop 0.22, DFSInputStream is a standalone class.  Before this,
             // it was an inner class of DFSClient.
             if (realIn.getClass().getName().endsWith("DFSInputStream")) {
               Method getFileLength = realIn.getClass().
                 getDeclaredMethod("getFileLength", new Class<?> []{});
               getFileLength.setAccessible(true);
               long realLength = ((Long)getFileLength.
                 invoke(realIn, new Object []{})).longValue();
               assert(realLength >= this.length);
               adjust = realLength - this.length;
             } else {
               LOG.info("Input stream class: " + realIn.getClass().getName() +
                   ", not adjusting length");
             }
           } catch(Exception e) {
             SequenceFileLogReader.LOG.warn(
               "Error while trying to get accurate file length.  " +
               "Truncation / data loss may occur if RegionServers die.", e);
           }
 
           return adjust + super.getPos();
         }
         return super.getPos();
       }
     }
   }
 
   // Protected for tests.
   protected SequenceFile.Reader reader;
   long entryStart = 0; // needed for logging exceptions
 
   public SequenceFileLogReader() {
     super();
   }
 
   @Override
   public void close() throws IOException {
     try {
       if (reader != null) {
         this.reader.close();
         this.reader = null;
       }
     } catch (IOException ioe) {
       throw addFileInfoToException(ioe);
     }
   }
 
   @Override
   public long getPosition() throws IOException {
     return reader != null ? reader.getPosition() : 0;
   }
 
   @Override
   public void reset() throws IOException {
     // Resetting the reader lets us see newly added data if the file is being written to
     // We also keep the same compressionContext which was previously populated for this file
     reader = new WALReader(fs, path, conf);
   }
 
   @Override
   protected void initReader(FSDataInputStream stream) throws IOException {
     // We don't use the stream because we have to have the magic stream above.
     if (stream != null) {
       stream.close();
     }
     reset();
   }
   
   @Override
   protected void initAfterCompression() throws IOException {
     // Nothing to do here
   }
 
   @Override
   protected boolean hasCompression() {
     return isWALCompressionEnabled(reader.getMetadata());
   }
 
   /**
    * Call this method after init() has been executed
    * @return whether WAL compression is enabled
    */
   static boolean isWALCompressionEnabled(final Metadata metadata) {
     // Check version is >= VERSION?
     Text txt = metadata.get(WAL_VERSION_KEY);
     if (txt == null || Integer.parseInt(txt.toString()) < COMPRESSION_VERSION) {
       return false;
     }
     // Now check that compression type is present.  Currently only one value.
     txt = metadata.get(WAL_COMPRESSION_TYPE_KEY);
     return txt != null && txt.equals(DICTIONARY_COMPRESSION_TYPE);
   }
 
 
   @Override
   protected boolean readNext(Entry e) throws IOException {
     try {
       boolean hasNext = this.reader.next(e.getKey(), e.getEdit());
       if (!hasNext) return false;
       // Scopes are probably in WAL edit, move to key
       NavigableMap<byte[], Integer> scopes = e.getEdit().getAndRemoveScopes();
       if (scopes != null) {
         e.getKey().setScopes(scopes);
       }
       return true;
     } catch (IOException ioe) {
       throw addFileInfoToException(ioe);
     }
   }
 
   @Override
   protected void seekOnFs(long pos) throws IOException {
     try {
       reader.seek(pos);
     } catch (IOException ioe) {
       throw addFileInfoToException(ioe);
     }
   }
 
   protected IOException addFileInfoToException(final IOException ioe)
   throws IOException {
     long pos = -1;
     try {
       pos = getPosition();
     } catch (IOException e) {
       LOG.warn("Failed getting position to add to throw", e);
     }
 
     // See what SequenceFile.Reader thinks is the end of the file
     long end = Long.MAX_VALUE;
     try {
       Field fEnd = SequenceFile.Reader.class.getDeclaredField("end");
       fEnd.setAccessible(true);
       end = fEnd.getLong(this.reader);
     } catch(Exception e) { /* reflection fail. keep going */ }
 
     String msg = (this.path == null? "": this.path.toString()) +
       ", entryStart=" + entryStart + ", pos=" + pos +
       ((end == Long.MAX_VALUE) ? "" : ", end=" + end) +
       ", edit=" + this.edit;
 
     // Enhance via reflection so we don't change the original class type
     try {
       return (IOException) ioe.getClass()
         .getConstructor(String.class)
         .newInstance(msg)
         .initCause(ioe);
     } catch(Exception e) { /* reflection fail. keep going */ }
 
     return ioe;
   }
 }