001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018
019package org.apache.hadoop.fs;
020
021import java.io.EOFException;
022import java.io.FileNotFoundException;
023import java.io.IOException;
024import java.io.InputStream;
025import java.nio.channels.ClosedChannelException;
026import java.util.Arrays;
027import java.util.List;
028
029import org.apache.hadoop.classification.InterfaceAudience;
030import org.apache.hadoop.classification.InterfaceStability;
031import org.apache.hadoop.conf.Configuration;
032import org.apache.hadoop.fs.permission.AclEntry;
033import org.apache.hadoop.fs.permission.FsPermission;
034import org.apache.hadoop.util.DataChecksum;
035import org.apache.hadoop.util.Progressable;
036
037/****************************************************************
038 * Abstract Checksumed FileSystem.
039 * It provide a basic implementation of a Checksumed FileSystem,
040 * which creates a checksum file for each raw file.
041 * It generates & verifies checksums at the client side.
042 *
043 *****************************************************************/
044@InterfaceAudience.Public
045@InterfaceStability.Stable
046public abstract class ChecksumFileSystem extends FilterFileSystem {
047  private static final byte[] CHECKSUM_VERSION = new byte[] {'c', 'r', 'c', 0};
048  private int bytesPerChecksum = 512;
049  private boolean verifyChecksum = true;
050  private boolean writeChecksum = true;
051
052  public static double getApproxChkSumLength(long size) {
053    return ChecksumFSOutputSummer.CHKSUM_AS_FRACTION * size;
054  }
055  
056  public ChecksumFileSystem(FileSystem fs) {
057    super(fs);
058  }
059
060  @Override
061  public void setConf(Configuration conf) {
062    super.setConf(conf);
063    if (conf != null) {
064      bytesPerChecksum = conf.getInt(LocalFileSystemConfigKeys.LOCAL_FS_BYTES_PER_CHECKSUM_KEY,
065                                     LocalFileSystemConfigKeys.LOCAL_FS_BYTES_PER_CHECKSUM_DEFAULT);
066    }
067  }
068  
069  /**
070   * Set whether to verify checksum.
071   */
072  @Override
073  public void setVerifyChecksum(boolean verifyChecksum) {
074    this.verifyChecksum = verifyChecksum;
075  }
076
077  @Override
078  public void setWriteChecksum(boolean writeChecksum) {
079    this.writeChecksum = writeChecksum;
080  }
081  
082  /** get the raw file system */
083  @Override
084  public FileSystem getRawFileSystem() {
085    return fs;
086  }
087
088  /** Return the name of the checksum file associated with a file.*/
089  public Path getChecksumFile(Path file) {
090    return new Path(file.getParent(), "." + file.getName() + ".crc");
091  }
092
093  /** Return true iff file is a checksum file name.*/
094  public static boolean isChecksumFile(Path file) {
095    String name = file.getName();
096    return name.startsWith(".") && name.endsWith(".crc");
097  }
098
099  /** Return the length of the checksum file given the size of the 
100   * actual file.
101   **/
102  public long getChecksumFileLength(Path file, long fileSize) {
103    return getChecksumLength(fileSize, getBytesPerSum());
104  }
105
106  /** Return the bytes Per Checksum */
107  public int getBytesPerSum() {
108    return bytesPerChecksum;
109  }
110
111  private int getSumBufferSize(int bytesPerSum, int bufferSize) {
112    int defaultBufferSize = getConf().getInt(
113                       LocalFileSystemConfigKeys.LOCAL_FS_STREAM_BUFFER_SIZE_KEY,
114                       LocalFileSystemConfigKeys.LOCAL_FS_STREAM_BUFFER_SIZE_DEFAULT);
115    int proportionalBufferSize = bufferSize / bytesPerSum;
116    return Math.max(bytesPerSum,
117                    Math.max(proportionalBufferSize, defaultBufferSize));
118  }
119
120  /*******************************************************
121   * For open()'s FSInputStream
122   * It verifies that data matches checksums.
123   *******************************************************/
124  private static class ChecksumFSInputChecker extends FSInputChecker {
125    private ChecksumFileSystem fs;
126    private FSDataInputStream datas;
127    private FSDataInputStream sums;
128    
129    private static final int HEADER_LENGTH = 8;
130    
131    private int bytesPerSum = 1;
132    
133    public ChecksumFSInputChecker(ChecksumFileSystem fs, Path file)
134      throws IOException {
135      this(fs, file, fs.getConf().getInt(
136                       LocalFileSystemConfigKeys.LOCAL_FS_STREAM_BUFFER_SIZE_KEY, 
137                       LocalFileSystemConfigKeys.LOCAL_FS_STREAM_BUFFER_SIZE_DEFAULT));
138    }
139    
140    public ChecksumFSInputChecker(ChecksumFileSystem fs, Path file, int bufferSize)
141      throws IOException {
142      super( file, fs.getFileStatus(file).getReplication() );
143      this.datas = fs.getRawFileSystem().open(file, bufferSize);
144      this.fs = fs;
145      Path sumFile = fs.getChecksumFile(file);
146      try {
147        int sumBufferSize = fs.getSumBufferSize(fs.getBytesPerSum(), bufferSize);
148        sums = fs.getRawFileSystem().open(sumFile, sumBufferSize);
149
150        byte[] version = new byte[CHECKSUM_VERSION.length];
151        sums.readFully(version);
152        if (!Arrays.equals(version, CHECKSUM_VERSION))
153          throw new IOException("Not a checksum file: "+sumFile);
154        this.bytesPerSum = sums.readInt();
155        set(fs.verifyChecksum, DataChecksum.newCrc32(), bytesPerSum, 4);
156      } catch (IOException e) {
157        // mincing the message is terrible, but java throws permission
158        // exceptions as FNF because that's all the method signatures allow!
159        if (!(e instanceof FileNotFoundException) ||
160            e.getMessage().endsWith(" (Permission denied)")) {
161          LOG.warn("Problem opening checksum file: "+ file +
162              ".  Ignoring exception: " , e);
163        }
164        set(fs.verifyChecksum, null, 1, 0);
165      }
166    }
167    
168    private long getChecksumFilePos( long dataPos ) {
169      return HEADER_LENGTH + 4*(dataPos/bytesPerSum);
170    }
171    
172    @Override
173    protected long getChunkPosition( long dataPos ) {
174      return dataPos/bytesPerSum*bytesPerSum;
175    }
176    
177    @Override
178    public int available() throws IOException {
179      return datas.available() + super.available();
180    }
181    
182    @Override
183    public int read(long position, byte[] b, int off, int len)
184      throws IOException {
185      // parameter check
186      if ((off | len | (off + len) | (b.length - (off + len))) < 0) {
187        throw new IndexOutOfBoundsException();
188      } else if (len == 0) {
189        return 0;
190      }
191      if( position<0 ) {
192        throw new IllegalArgumentException(
193            "Parameter position can not to be negative");
194      }
195
196      ChecksumFSInputChecker checker = new ChecksumFSInputChecker(fs, file);
197      checker.seek(position);
198      int nread = checker.read(b, off, len);
199      checker.close();
200      return nread;
201    }
202    
203    @Override
204    public void close() throws IOException {
205      datas.close();
206      if( sums != null ) {
207        sums.close();
208      }
209      set(fs.verifyChecksum, null, 1, 0);
210    }
211    
212
213    @Override
214    public boolean seekToNewSource(long targetPos) throws IOException {
215      long sumsPos = getChecksumFilePos(targetPos);
216      fs.reportChecksumFailure(file, datas, targetPos, sums, sumsPos);
217      boolean newDataSource = datas.seekToNewSource(targetPos);
218      return sums.seekToNewSource(sumsPos) || newDataSource;
219    }
220
221    @Override
222    protected int readChunk(long pos, byte[] buf, int offset, int len,
223        byte[] checksum) throws IOException {
224
225      boolean eof = false;
226      if (needChecksum()) {
227        assert checksum != null; // we have a checksum buffer
228        assert checksum.length % CHECKSUM_SIZE == 0; // it is sane length
229        assert len >= bytesPerSum; // we must read at least one chunk
230
231        final int checksumsToRead = Math.min(
232          len/bytesPerSum, // number of checksums based on len to read
233          checksum.length / CHECKSUM_SIZE); // size of checksum buffer
234        long checksumPos = getChecksumFilePos(pos); 
235        if(checksumPos != sums.getPos()) {
236          sums.seek(checksumPos);
237        }
238
239        int sumLenRead = sums.read(checksum, 0, CHECKSUM_SIZE * checksumsToRead);
240        if (sumLenRead >= 0 && sumLenRead % CHECKSUM_SIZE != 0) {
241          throw new ChecksumException(
242            "Checksum file not a length multiple of checksum size " +
243            "in " + file + " at " + pos + " checksumpos: " + checksumPos +
244            " sumLenread: " + sumLenRead,
245            pos);
246        }
247        if (sumLenRead <= 0) { // we're at the end of the file
248          eof = true;
249        } else {
250          // Adjust amount of data to read based on how many checksum chunks we read
251          len = Math.min(len, bytesPerSum * (sumLenRead / CHECKSUM_SIZE));
252        }
253      }
254      if(pos != datas.getPos()) {
255        datas.seek(pos);
256      }
257      int nread = readFully(datas, buf, offset, len);
258      if (eof && nread > 0) {
259        throw new ChecksumException("Checksum error: "+file+" at "+pos, pos);
260      }
261      return nread;
262    }
263  }
264  
265  private static class FSDataBoundedInputStream extends FSDataInputStream {
266    private FileSystem fs;
267    private Path file;
268    private long fileLen = -1L;
269
270    FSDataBoundedInputStream(FileSystem fs, Path file, InputStream in) {
271      super(in);
272      this.fs = fs;
273      this.file = file;
274    }
275    
276    @Override
277    public boolean markSupported() {
278      return false;
279    }
280    
281    /* Return the file length */
282    private long getFileLength() throws IOException {
283      if( fileLen==-1L ) {
284        fileLen = fs.getContentSummary(file).getLength();
285      }
286      return fileLen;
287    }
288    
289    /**
290     * Skips over and discards <code>n</code> bytes of data from the
291     * input stream.
292     *
293     *The <code>skip</code> method skips over some smaller number of bytes
294     * when reaching end of file before <code>n</code> bytes have been skipped.
295     * The actual number of bytes skipped is returned.  If <code>n</code> is
296     * negative, no bytes are skipped.
297     *
298     * @param      n   the number of bytes to be skipped.
299     * @return     the actual number of bytes skipped.
300     * @exception  IOException  if an I/O error occurs.
301     *             ChecksumException if the chunk to skip to is corrupted
302     */
303    @Override
304    public synchronized long skip(long n) throws IOException {
305      long curPos = getPos();
306      long fileLength = getFileLength();
307      if( n+curPos > fileLength ) {
308        n = fileLength - curPos;
309      }
310      return super.skip(n);
311    }
312    
313    /**
314     * Seek to the given position in the stream.
315     * The next read() will be from that position.
316     * 
317     * <p>This method does not allow seek past the end of the file.
318     * This produces IOException.
319     *
320     * @param      pos   the postion to seek to.
321     * @exception  IOException  if an I/O error occurs or seeks after EOF
322     *             ChecksumException if the chunk to seek to is corrupted
323     */
324
325    @Override
326    public synchronized void seek(long pos) throws IOException {
327      if (pos > getFileLength()) {
328        throw new EOFException("Cannot seek after EOF");
329      }
330      super.seek(pos);
331    }
332
333  }
334
335  /**
336   * Opens an FSDataInputStream at the indicated Path.
337   * @param f the file name to open
338   * @param bufferSize the size of the buffer to be used.
339   */
340  @Override
341  public FSDataInputStream open(Path f, int bufferSize) throws IOException {
342    FileSystem fs;
343    InputStream in;
344    if (verifyChecksum) {
345      fs = this;
346      in = new ChecksumFSInputChecker(this, f, bufferSize);
347    } else {
348      fs = getRawFileSystem();
349      in = fs.open(f, bufferSize);
350    }
351    return new FSDataBoundedInputStream(fs, f, in);
352  }
353
354  @Override
355  public FSDataOutputStream append(Path f, int bufferSize,
356      Progressable progress) throws IOException {
357    throw new IOException("Not supported");
358  }
359
360  /**
361   * Calculated the length of the checksum file in bytes.
362   * @param size the length of the data file in bytes
363   * @param bytesPerSum the number of bytes in a checksum block
364   * @return the number of bytes in the checksum file
365   */
366  public static long getChecksumLength(long size, int bytesPerSum) {
367    //the checksum length is equal to size passed divided by bytesPerSum +
368    //bytes written in the beginning of the checksum file.  
369    return ((size + bytesPerSum - 1) / bytesPerSum) * 4 +
370             CHECKSUM_VERSION.length + 4;  
371  }
372
373  /** This class provides an output stream for a checksummed file.
374   * It generates checksums for data. */
375  private static class ChecksumFSOutputSummer extends FSOutputSummer {
376    private FSDataOutputStream datas;    
377    private FSDataOutputStream sums;
378    private static final float CHKSUM_AS_FRACTION = 0.01f;
379    private boolean isClosed = false;
380    
381    public ChecksumFSOutputSummer(ChecksumFileSystem fs, 
382                          Path file, 
383                          boolean overwrite,
384                          int bufferSize,
385                          short replication,
386                          long blockSize,
387                          Progressable progress)
388      throws IOException {
389      super(DataChecksum.newDataChecksum(DataChecksum.Type.CRC32,
390          fs.getBytesPerSum()));
391      int bytesPerSum = fs.getBytesPerSum();
392      this.datas = fs.getRawFileSystem().create(file, overwrite, bufferSize, 
393                                         replication, blockSize, progress);
394      int sumBufferSize = fs.getSumBufferSize(bytesPerSum, bufferSize);
395      this.sums = fs.getRawFileSystem().create(fs.getChecksumFile(file), true, 
396                                               sumBufferSize, replication,
397                                               blockSize);
398      sums.write(CHECKSUM_VERSION, 0, CHECKSUM_VERSION.length);
399      sums.writeInt(bytesPerSum);
400    }
401    
402    @Override
403    public void close() throws IOException {
404      try {
405        flushBuffer();
406        sums.close();
407        datas.close();
408      } finally {
409        isClosed = true;
410      }
411    }
412    
413    @Override
414    protected void writeChunk(byte[] b, int offset, int len, byte[] checksum,
415        int ckoff, int cklen)
416    throws IOException {
417      datas.write(b, offset, len);
418      sums.write(checksum, ckoff, cklen);
419    }
420
421    @Override
422    protected void checkClosed() throws IOException {
423      if (isClosed) {
424        throw new ClosedChannelException();
425      }
426    }
427  }
428
429  @Override
430  public FSDataOutputStream create(Path f, FsPermission permission,
431      boolean overwrite, int bufferSize, short replication, long blockSize,
432      Progressable progress) throws IOException {
433    return create(f, permission, overwrite, true, bufferSize,
434        replication, blockSize, progress);
435  }
436
437  private FSDataOutputStream create(Path f, FsPermission permission,
438      boolean overwrite, boolean createParent, int bufferSize,
439      short replication, long blockSize,
440      Progressable progress) throws IOException {
441    Path parent = f.getParent();
442    if (parent != null) {
443      if (!createParent && !exists(parent)) {
444        throw new FileNotFoundException("Parent directory doesn't exist: "
445            + parent);
446      } else if (!mkdirs(parent)) {
447        throw new IOException("Mkdirs failed to create " + parent
448            + " (exists=" + exists(parent) + ", cwd=" + getWorkingDirectory()
449            + ")");
450      }
451    }
452    final FSDataOutputStream out;
453    if (writeChecksum) {
454      out = new FSDataOutputStream(
455          new ChecksumFSOutputSummer(this, f, overwrite, bufferSize, replication,
456              blockSize, progress), null);
457    } else {
458      out = fs.create(f, permission, overwrite, bufferSize, replication,
459          blockSize, progress);
460      // remove the checksum file since we aren't writing one
461      Path checkFile = getChecksumFile(f);
462      if (fs.exists(checkFile)) {
463        fs.delete(checkFile, true);
464      }
465    }
466    if (permission != null) {
467      setPermission(f, permission);
468    }
469    return out;
470  }
471
472  @Override
473  public FSDataOutputStream createNonRecursive(Path f, FsPermission permission,
474      boolean overwrite, int bufferSize, short replication, long blockSize,
475      Progressable progress) throws IOException {
476    return create(f, permission, overwrite, false, bufferSize, replication,
477        blockSize, progress);
478  }
479
480  abstract class FsOperation {
481    boolean run(Path p) throws IOException {
482      boolean status = apply(p);
483      if (status) {
484        Path checkFile = getChecksumFile(p);
485        if (fs.exists(checkFile)) {
486          apply(checkFile);
487        }
488      }
489      return status;
490    }
491    abstract boolean apply(Path p) throws IOException;
492  }
493
494
495  @Override
496  public void setPermission(Path src, final FsPermission permission)
497      throws IOException {
498    new FsOperation(){
499      @Override
500      boolean apply(Path p) throws IOException {
501        fs.setPermission(p, permission);
502        return true;
503      }
504    }.run(src);
505  }
506
507  @Override
508  public void setOwner(Path src, final String username, final String groupname)
509      throws IOException {
510    new FsOperation(){
511      @Override
512      boolean apply(Path p) throws IOException {
513        fs.setOwner(p, username, groupname);
514        return true;
515      }
516    }.run(src);
517  }
518
519  @Override
520  public void setAcl(Path src, final List<AclEntry> aclSpec)
521      throws IOException {
522    new FsOperation(){
523      @Override
524      boolean apply(Path p) throws IOException {
525        fs.setAcl(p, aclSpec);
526        return true;
527      }
528    }.run(src);
529  }
530
531  @Override
532  public void modifyAclEntries(Path src, final List<AclEntry> aclSpec)
533      throws IOException {
534    new FsOperation(){
535      @Override
536      boolean apply(Path p) throws IOException {
537        fs.modifyAclEntries(p, aclSpec);
538        return true;
539      }
540    }.run(src);
541  }
542
543  @Override
544  public void removeAcl(Path src) throws IOException {
545    new FsOperation(){
546      @Override
547      boolean apply(Path p) throws IOException {
548        fs.removeAcl(p);
549        return true;
550      }
551    }.run(src);
552  }
553
554  @Override
555  public void removeAclEntries(Path src, final List<AclEntry> aclSpec)
556      throws IOException {
557    new FsOperation(){
558      @Override
559      boolean apply(Path p) throws IOException {
560        fs.removeAclEntries(p, aclSpec);
561        return true;
562      }
563    }.run(src);
564  }
565
566  @Override
567  public void removeDefaultAcl(Path src) throws IOException {
568    new FsOperation(){
569      @Override
570      boolean apply(Path p) throws IOException {
571        fs.removeDefaultAcl(p);
572        return true;
573      }
574    }.run(src);
575  }
576
577  /**
578   * Set replication for an existing file.
579   * Implement the abstract <tt>setReplication</tt> of <tt>FileSystem</tt>
580   * @param src file name
581   * @param replication new replication
582   * @throws IOException
583   * @return true if successful;
584   *         false if file does not exist or is a directory
585   */
586  @Override
587  public boolean setReplication(Path src, final short replication)
588      throws IOException {
589    return new FsOperation(){
590      @Override
591      boolean apply(Path p) throws IOException {
592        return fs.setReplication(p, replication);
593      }
594    }.run(src);
595  }
596
597  /**
598   * Rename files/dirs
599   */
600  @Override
601  public boolean rename(Path src, Path dst) throws IOException {
602    if (fs.isDirectory(src)) {
603      return fs.rename(src, dst);
604    } else {
605      if (fs.isDirectory(dst)) {
606        dst = new Path(dst, src.getName());
607      }
608
609      boolean value = fs.rename(src, dst);
610      if (!value)
611        return false;
612
613      Path srcCheckFile = getChecksumFile(src);
614      Path dstCheckFile = getChecksumFile(dst);
615      if (fs.exists(srcCheckFile)) { //try to rename checksum
616        value = fs.rename(srcCheckFile, dstCheckFile);
617      } else if (fs.exists(dstCheckFile)) {
618        // no src checksum, so remove dst checksum
619        value = fs.delete(dstCheckFile, true); 
620      }
621
622      return value;
623    }
624  }
625
626  /**
627   * Implement the delete(Path, boolean) in checksum
628   * file system.
629   */
630  @Override
631  public boolean delete(Path f, boolean recursive) throws IOException{
632    FileStatus fstatus = null;
633    try {
634      fstatus = fs.getFileStatus(f);
635    } catch(FileNotFoundException e) {
636      return false;
637    }
638    if (fstatus.isDirectory()) {
639      //this works since the crcs are in the same
640      //directories and the files. so we just delete
641      //everything in the underlying filesystem
642      return fs.delete(f, recursive);
643    } else {
644      Path checkFile = getChecksumFile(f);
645      if (fs.exists(checkFile)) {
646        fs.delete(checkFile, true);
647      }
648      return fs.delete(f, true);
649    }
650  }
651    
652  final private static PathFilter DEFAULT_FILTER = new PathFilter() {
653    @Override
654    public boolean accept(Path file) {
655      return !isChecksumFile(file);
656    }
657  };
658
659  /**
660   * List the statuses of the files/directories in the given path if the path is
661   * a directory.
662   * 
663   * @param f
664   *          given path
665   * @return the statuses of the files/directories in the given path
666   * @throws IOException
667   */
668  @Override
669  public FileStatus[] listStatus(Path f) throws IOException {
670    return fs.listStatus(f, DEFAULT_FILTER);
671  }
672  
673  /**
674   * List the statuses of the files/directories in the given path if the path is
675   * a directory.
676   * 
677   * @param f
678   *          given path
679   * @return the statuses of the files/directories in the given patch
680   * @throws IOException
681   */
682  @Override
683  public RemoteIterator<LocatedFileStatus> listLocatedStatus(Path f)
684  throws IOException {
685    return fs.listLocatedStatus(f, DEFAULT_FILTER);
686  }
687  
688  @Override
689  public boolean mkdirs(Path f) throws IOException {
690    return fs.mkdirs(f);
691  }
692
693  @Override
694  public void copyFromLocalFile(boolean delSrc, Path src, Path dst)
695    throws IOException {
696    Configuration conf = getConf();
697    FileUtil.copy(getLocal(conf), src, this, dst, delSrc, conf);
698  }
699
700  /**
701   * The src file is under FS, and the dst is on the local disk.
702   * Copy it from FS control to the local dst name.
703   */
704  @Override
705  public void copyToLocalFile(boolean delSrc, Path src, Path dst)
706    throws IOException {
707    Configuration conf = getConf();
708    FileUtil.copy(this, src, getLocal(conf), dst, delSrc, conf);
709  }
710
711  /**
712   * The src file is under FS, and the dst is on the local disk.
713   * Copy it from FS control to the local dst name.
714   * If src and dst are directories, the copyCrc parameter
715   * determines whether to copy CRC files.
716   */
717  public void copyToLocalFile(Path src, Path dst, boolean copyCrc)
718    throws IOException {
719    if (!fs.isDirectory(src)) { // source is a file
720      fs.copyToLocalFile(src, dst);
721      FileSystem localFs = getLocal(getConf()).getRawFileSystem();
722      if (localFs.isDirectory(dst)) {
723        dst = new Path(dst, src.getName());
724      }
725      dst = getChecksumFile(dst);
726      if (localFs.exists(dst)) { //remove old local checksum file
727        localFs.delete(dst, true);
728      }
729      Path checksumFile = getChecksumFile(src);
730      if (copyCrc && fs.exists(checksumFile)) { //copy checksum file
731        fs.copyToLocalFile(checksumFile, dst);
732      }
733    } else {
734      FileStatus[] srcs = listStatus(src);
735      for (FileStatus srcFile : srcs) {
736        copyToLocalFile(srcFile.getPath(), 
737                        new Path(dst, srcFile.getPath().getName()), copyCrc);
738      }
739    }
740  }
741
742  @Override
743  public Path startLocalOutput(Path fsOutputFile, Path tmpLocalFile)
744    throws IOException {
745    return tmpLocalFile;
746  }
747
748  @Override
749  public void completeLocalOutput(Path fsOutputFile, Path tmpLocalFile)
750    throws IOException {
751    moveFromLocalFile(tmpLocalFile, fsOutputFile);
752  }
753
754  /**
755   * Report a checksum error to the file system.
756   * @param f the file name containing the error
757   * @param in the stream open on the file
758   * @param inPos the position of the beginning of the bad data in the file
759   * @param sums the stream open on the checksum file
760   * @param sumsPos the position of the beginning of the bad data in the checksum file
761   * @return if retry is neccessary
762   */
763  public boolean reportChecksumFailure(Path f, FSDataInputStream in,
764                                       long inPos, FSDataInputStream sums, long sumsPos) {
765    return false;
766  }
767}