| /** |
| * 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.hdfs.server.namenode; |
| |
| import java.io.BufferedInputStream; |
| import java.io.DataInput; |
| import java.io.DataInputStream; |
| import java.io.DataOutput; |
| import java.io.EOFException; |
| import java.io.FilterInputStream; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.util.Arrays; |
| import java.util.zip.CheckedInputStream; |
| import java.util.zip.Checksum; |
| |
| import org.apache.hadoop.fs.ChecksumException; |
| import org.apache.hadoop.fs.Options.Rename; |
| import org.apache.hadoop.fs.permission.FsPermission; |
| import org.apache.hadoop.fs.permission.PermissionStatus; |
| import org.apache.hadoop.hdfs.protocol.Block; |
| import org.apache.hadoop.hdfs.protocol.DatanodeID; |
| import org.apache.hadoop.hdfs.protocol.FSConstants; |
| import org.apache.hadoop.hdfs.protocol.HdfsFileStatus; |
| import org.apache.hadoop.hdfs.protocol.LayoutVersion; |
| import org.apache.hadoop.hdfs.protocol.LayoutVersion.Feature; |
| import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier; |
| import static org.apache.hadoop.hdfs.server.common.Util.now; |
| import org.apache.hadoop.hdfs.server.common.GenerationStamp; |
| import org.apache.hadoop.hdfs.server.common.Storage; |
| import org.apache.hadoop.hdfs.server.namenode.LeaseManager.Lease; |
| import org.apache.hadoop.io.BytesWritable; |
| import org.apache.hadoop.io.LongWritable; |
| import org.apache.hadoop.io.Writable; |
| import org.apache.hadoop.io.WritableFactories; |
| import org.apache.hadoop.io.WritableFactory; |
| import org.apache.hadoop.security.token.delegation.DelegationKey; |
| |
| public class FSEditLogLoader { |
| private final FSNamesystem fsNamesys; |
| |
| public FSEditLogLoader(FSNamesystem fsNamesys) { |
| this.fsNamesys = fsNamesys; |
| } |
| |
| /** |
| * Load an edit log, and apply the changes to the in-memory structure |
| * This is where we apply edits that we've been writing to disk all |
| * along. |
| */ |
| int loadFSEdits(EditLogInputStream edits) throws IOException { |
| long startTime = now(); |
| int numEdits = loadFSEdits(edits, true); |
| FSImage.LOG.info("Edits file " + edits.getName() |
| + " of size " + edits.length() + " edits # " + numEdits |
| + " loaded in " + (now()-startTime)/1000 + " seconds."); |
| return numEdits; |
| } |
| |
| /** |
| * Read the header of fsedit log |
| * @param in fsedit stream |
| * @return the edit log version number |
| * @throws IOException if error occurs |
| */ |
| int readLogVersion(DataInputStream in) throws IOException { |
| int logVersion = 0; |
| // Read log file version. Could be missing. |
| in.mark(4); |
| // If edits log is greater than 2G, available method will return negative |
| // numbers, so we avoid having to call available |
| boolean available = true; |
| try { |
| logVersion = in.readByte(); |
| } catch (EOFException e) { |
| available = false; |
| } |
| if (available) { |
| in.reset(); |
| logVersion = in.readInt(); |
| if (logVersion < FSConstants.LAYOUT_VERSION) // future version |
| throw new IOException( |
| "Unexpected version of the file system log file: " |
| + logVersion + ". Current version = " |
| + FSConstants.LAYOUT_VERSION + "."); |
| } |
| assert logVersion <= Storage.LAST_UPGRADABLE_LAYOUT_VERSION : |
| "Unsupported version " + logVersion; |
| return logVersion; |
| } |
| |
| int loadFSEdits(EditLogInputStream edits, boolean closeOnExit) throws IOException { |
| BufferedInputStream bin = new BufferedInputStream(edits); |
| DataInputStream in = new DataInputStream(bin); |
| |
| int numEdits = 0; |
| int logVersion = 0; |
| |
| try { |
| logVersion = readLogVersion(in); |
| Checksum checksum = null; |
| if (LayoutVersion.supports(Feature.EDITS_CHESKUM, logVersion)) { |
| checksum = FSEditLog.getChecksum(); |
| in = new DataInputStream(new CheckedInputStream(bin, checksum)); |
| } |
| |
| numEdits = loadEditRecords(logVersion, in, checksum, false); |
| } finally { |
| if(closeOnExit) |
| in.close(); |
| } |
| if (logVersion != FSConstants.LAYOUT_VERSION) // other version |
| numEdits++; // save this image asap |
| return numEdits; |
| } |
| |
| @SuppressWarnings("deprecation") |
| int loadEditRecords(int logVersion, DataInputStream in, |
| Checksum checksum, boolean closeOnExit) throws IOException { |
| FSDirectory fsDir = fsNamesys.dir; |
| int numEdits = 0; |
| String clientName = null; |
| String clientMachine = null; |
| String path = null; |
| int numOpAdd = 0, numOpClose = 0, numOpDelete = 0, |
| numOpRenameOld = 0, numOpSetRepl = 0, numOpMkDir = 0, |
| numOpSetPerm = 0, numOpSetOwner = 0, numOpSetGenStamp = 0, |
| numOpTimes = 0, numOpRename = 0, numOpConcatDelete = 0, |
| numOpSymlink = 0, numOpGetDelegationToken = 0, |
| numOpRenewDelegationToken = 0, numOpCancelDelegationToken = 0, |
| numOpUpdateMasterKey = 0, numOpReassignLease = 0, numOpOther = 0; |
| |
| // Keep track of the file offsets of the last several opcodes. |
| // This is handy when manually recovering corrupted edits files. |
| PositionTrackingInputStream tracker = new PositionTrackingInputStream(in); |
| in = new DataInputStream(tracker); |
| long recentOpcodeOffsets[] = new long[4]; |
| Arrays.fill(recentOpcodeOffsets, -1); |
| |
| try { |
| try { |
| while (true) { |
| long timestamp = 0; |
| long mtime = 0; |
| long atime = 0; |
| long blockSize = 0; |
| FSEditLogOpCodes opCode; |
| try { |
| if (checksum != null) { |
| checksum.reset(); |
| } |
| in.mark(1); |
| byte opCodeByte = in.readByte(); |
| opCode = FSEditLogOpCodes.fromByte(opCodeByte); |
| if (opCode == FSEditLogOpCodes.OP_INVALID) { |
| in.reset(); // reset back to end of file if somebody reads it again |
| break; // no more transactions |
| } |
| } catch (EOFException e) { |
| break; // no more transactions |
| } |
| recentOpcodeOffsets[numEdits % recentOpcodeOffsets.length] = |
| tracker.getPos(); |
| numEdits++; |
| switch (opCode) { |
| case OP_ADD: |
| case OP_CLOSE: { |
| // versions > 0 support per file replication |
| // get name and replication |
| int length = in.readInt(); |
| if (-7 == logVersion && length != 3|| |
| -17 < logVersion && logVersion < -7 && length != 4 || |
| logVersion <= -17 && length != 5) { |
| throw new IOException("Incorrect data format." + |
| " logVersion is " + logVersion + |
| " but writables.length is " + |
| length + ". "); |
| } |
| path = FSImageSerialization.readString(in); |
| short replication = fsNamesys.adjustReplication(readShort(in)); |
| mtime = readLong(in); |
| if (LayoutVersion.supports(Feature.FILE_ACCESS_TIME, logVersion)) { |
| atime = readLong(in); |
| } |
| if (logVersion < -7) { |
| blockSize = readLong(in); |
| } |
| // get blocks |
| boolean isFileUnderConstruction = (opCode == FSEditLogOpCodes.OP_ADD); |
| BlockInfo blocks[] = |
| readBlocks(in, logVersion, isFileUnderConstruction, replication); |
| |
| // Older versions of HDFS does not store the block size in inode. |
| // If the file has more than one block, use the size of the |
| // first block as the blocksize. Otherwise use the default |
| // block size. |
| if (-8 <= logVersion && blockSize == 0) { |
| if (blocks.length > 1) { |
| blockSize = blocks[0].getNumBytes(); |
| } else { |
| long first = ((blocks.length == 1)? blocks[0].getNumBytes(): 0); |
| blockSize = Math.max(fsNamesys.getDefaultBlockSize(), first); |
| } |
| } |
| |
| PermissionStatus permissions = fsNamesys.getUpgradePermission(); |
| if (logVersion <= -11) { |
| permissions = PermissionStatus.read(in); |
| } |
| |
| // clientname, clientMachine and block locations of last block. |
| if (opCode == FSEditLogOpCodes.OP_ADD && logVersion <= -12) { |
| clientName = FSImageSerialization.readString(in); |
| clientMachine = FSImageSerialization.readString(in); |
| if (-13 <= logVersion) { |
| readDatanodeDescriptorArray(in); |
| } |
| } else { |
| clientName = ""; |
| clientMachine = ""; |
| } |
| |
| // The open lease transaction re-creates a file if necessary. |
| // Delete the file if it already exists. |
| if (FSNamesystem.LOG.isDebugEnabled()) { |
| FSNamesystem.LOG.debug(opCode + ": " + path + |
| " numblocks : " + blocks.length + |
| " clientHolder " + clientName + |
| " clientMachine " + clientMachine); |
| } |
| |
| fsDir.unprotectedDelete(path, mtime); |
| |
| // add to the file tree |
| INodeFile node = (INodeFile)fsDir.unprotectedAddFile( |
| path, permissions, |
| blocks, replication, |
| mtime, atime, blockSize); |
| if (isFileUnderConstruction) { |
| numOpAdd++; |
| // |
| // Replace current node with a INodeUnderConstruction. |
| // Recreate in-memory lease record. |
| // |
| INodeFileUnderConstruction cons = new INodeFileUnderConstruction( |
| node.getLocalNameBytes(), |
| node.getReplication(), |
| node.getModificationTime(), |
| node.getPreferredBlockSize(), |
| node.getBlocks(), |
| node.getPermissionStatus(), |
| clientName, |
| clientMachine, |
| null); |
| fsDir.replaceNode(path, node, cons); |
| fsNamesys.leaseManager.addLease(cons.getClientName(), path); |
| } |
| break; |
| } |
| case OP_SET_REPLICATION: { |
| numOpSetRepl++; |
| path = FSImageSerialization.readString(in); |
| short replication = fsNamesys.adjustReplication(readShort(in)); |
| fsDir.unprotectedSetReplication(path, replication, null); |
| break; |
| } |
| case OP_CONCAT_DELETE: { |
| numOpConcatDelete++; |
| int length = in.readInt(); |
| if (length < 3) { // trg, srcs.., timestam |
| throw new IOException("Incorrect data format. " |
| + "Mkdir operation."); |
| } |
| String trg = FSImageSerialization.readString(in); |
| int srcSize = length - 1 - 1; //trg and timestamp |
| String [] srcs = new String [srcSize]; |
| for(int i=0; i<srcSize;i++) { |
| srcs[i]= FSImageSerialization.readString(in); |
| } |
| timestamp = readLong(in); |
| fsDir.unprotectedConcat(trg, srcs); |
| break; |
| } |
| case OP_RENAME_OLD: { |
| numOpRenameOld++; |
| int length = in.readInt(); |
| if (length != 3) { |
| throw new IOException("Incorrect data format. " |
| + "Mkdir operation."); |
| } |
| String s = FSImageSerialization.readString(in); |
| String d = FSImageSerialization.readString(in); |
| timestamp = readLong(in); |
| HdfsFileStatus dinfo = fsDir.getFileInfo(d, false); |
| fsDir.unprotectedRenameTo(s, d, timestamp); |
| fsNamesys.changeLease(s, d, dinfo); |
| break; |
| } |
| case OP_DELETE: { |
| numOpDelete++; |
| int length = in.readInt(); |
| if (length != 2) { |
| throw new IOException("Incorrect data format. " |
| + "delete operation."); |
| } |
| path = FSImageSerialization.readString(in); |
| timestamp = readLong(in); |
| fsDir.unprotectedDelete(path, timestamp); |
| break; |
| } |
| case OP_MKDIR: { |
| numOpMkDir++; |
| PermissionStatus permissions = fsNamesys.getUpgradePermission(); |
| int length = in.readInt(); |
| if (-17 < logVersion && length != 2 || |
| logVersion <= -17 && length != 3) { |
| throw new IOException("Incorrect data format. " |
| + "Mkdir operation."); |
| } |
| path = FSImageSerialization.readString(in); |
| timestamp = readLong(in); |
| |
| // The disk format stores atimes for directories as well. |
| // However, currently this is not being updated/used because of |
| // performance reasons. |
| if (LayoutVersion.supports(Feature.FILE_ACCESS_TIME, logVersion)) { |
| atime = readLong(in); |
| } |
| |
| if (logVersion <= -11) { |
| permissions = PermissionStatus.read(in); |
| } |
| fsDir.unprotectedMkdir(path, permissions, timestamp); |
| break; |
| } |
| case OP_SET_GENSTAMP: { |
| numOpSetGenStamp++; |
| long lw = in.readLong(); |
| fsNamesys.setGenerationStamp(lw); |
| break; |
| } |
| case OP_DATANODE_ADD: { |
| numOpOther++; |
| //Datanodes are not persistent any more. |
| FSImageSerialization.DatanodeImage.skipOne(in); |
| break; |
| } |
| case OP_DATANODE_REMOVE: { |
| numOpOther++; |
| DatanodeID nodeID = new DatanodeID(); |
| nodeID.readFields(in); |
| //Datanodes are not persistent any more. |
| break; |
| } |
| case OP_SET_PERMISSIONS: { |
| numOpSetPerm++; |
| fsDir.unprotectedSetPermission( |
| FSImageSerialization.readString(in), FsPermission.read(in)); |
| break; |
| } |
| case OP_SET_OWNER: { |
| numOpSetOwner++; |
| if (logVersion > -11) |
| throw new IOException("Unexpected opCode " + opCode |
| + " for version " + logVersion); |
| fsDir.unprotectedSetOwner(FSImageSerialization.readString(in), |
| FSImageSerialization.readString_EmptyAsNull(in), |
| FSImageSerialization.readString_EmptyAsNull(in)); |
| break; |
| } |
| case OP_SET_NS_QUOTA: { |
| fsDir.unprotectedSetQuota(FSImageSerialization.readString(in), |
| readLongWritable(in), |
| FSConstants.QUOTA_DONT_SET); |
| break; |
| } |
| case OP_CLEAR_NS_QUOTA: { |
| fsDir.unprotectedSetQuota(FSImageSerialization.readString(in), |
| FSConstants.QUOTA_RESET, |
| FSConstants.QUOTA_DONT_SET); |
| break; |
| } |
| |
| case OP_SET_QUOTA: |
| fsDir.unprotectedSetQuota(FSImageSerialization.readString(in), |
| readLongWritable(in), |
| readLongWritable(in)); |
| |
| break; |
| |
| case OP_TIMES: { |
| numOpTimes++; |
| int length = in.readInt(); |
| if (length != 3) { |
| throw new IOException("Incorrect data format. " |
| + "times operation."); |
| } |
| path = FSImageSerialization.readString(in); |
| mtime = readLong(in); |
| atime = readLong(in); |
| fsDir.unprotectedSetTimes(path, mtime, atime, true); |
| break; |
| } |
| case OP_SYMLINK: { |
| numOpSymlink++; |
| int length = in.readInt(); |
| if (length != 4) { |
| throw new IOException("Incorrect data format. " |
| + "symlink operation."); |
| } |
| path = FSImageSerialization.readString(in); |
| String value = FSImageSerialization.readString(in); |
| mtime = readLong(in); |
| atime = readLong(in); |
| PermissionStatus perm = PermissionStatus.read(in); |
| fsDir.unprotectedSymlink(path, value, mtime, atime, perm); |
| break; |
| } |
| case OP_RENAME: { |
| numOpRename++; |
| int length = in.readInt(); |
| if (length != 3) { |
| throw new IOException("Incorrect data format. " |
| + "Mkdir operation."); |
| } |
| String s = FSImageSerialization.readString(in); |
| String d = FSImageSerialization.readString(in); |
| timestamp = readLong(in); |
| Rename[] options = readRenameOptions(in); |
| HdfsFileStatus dinfo = fsDir.getFileInfo(d, false); |
| fsDir.unprotectedRenameTo(s, d, timestamp, options); |
| fsNamesys.changeLease(s, d, dinfo); |
| break; |
| } |
| case OP_GET_DELEGATION_TOKEN: { |
| numOpGetDelegationToken++; |
| DelegationTokenIdentifier delegationTokenId = |
| new DelegationTokenIdentifier(); |
| delegationTokenId.readFields(in); |
| long expiryTime = readLong(in); |
| fsNamesys.getDelegationTokenSecretManager() |
| .addPersistedDelegationToken(delegationTokenId, expiryTime); |
| break; |
| } |
| case OP_RENEW_DELEGATION_TOKEN: { |
| numOpRenewDelegationToken++; |
| DelegationTokenIdentifier delegationTokenId = |
| new DelegationTokenIdentifier(); |
| delegationTokenId.readFields(in); |
| long expiryTime = readLong(in); |
| fsNamesys.getDelegationTokenSecretManager() |
| .updatePersistedTokenRenewal(delegationTokenId, expiryTime); |
| break; |
| } |
| case OP_CANCEL_DELEGATION_TOKEN: { |
| numOpCancelDelegationToken++; |
| DelegationTokenIdentifier delegationTokenId = |
| new DelegationTokenIdentifier(); |
| delegationTokenId.readFields(in); |
| fsNamesys.getDelegationTokenSecretManager() |
| .updatePersistedTokenCancellation(delegationTokenId); |
| break; |
| } |
| case OP_UPDATE_MASTER_KEY: { |
| numOpUpdateMasterKey++; |
| DelegationKey delegationKey = new DelegationKey(); |
| delegationKey.readFields(in); |
| fsNamesys.getDelegationTokenSecretManager().updatePersistedMasterKey( |
| delegationKey); |
| break; |
| } |
| case OP_REASSIGN_LEASE: { |
| numOpReassignLease++; |
| String leaseHolder = FSImageSerialization.readString(in); |
| path = FSImageSerialization.readString(in); |
| String newHolder = FSImageSerialization.readString(in); |
| Lease lease = fsNamesys.leaseManager.getLease(leaseHolder); |
| INodeFileUnderConstruction pendingFile = |
| (INodeFileUnderConstruction) fsDir.getFileINode(path); |
| fsNamesys.reassignLeaseInternal(lease, path, newHolder, pendingFile); |
| break; |
| } |
| default: { |
| throw new IOException("Never seen opCode " + opCode); |
| } |
| } |
| validateChecksum(in, checksum, numEdits); |
| } |
| } catch (IOException ex) { |
| check203UpgradeFailure(logVersion, ex); |
| } finally { |
| if(closeOnExit) |
| in.close(); |
| } |
| } catch (Throwable t) { |
| // Catch Throwable because in the case of a truly corrupt edits log, any |
| // sort of error might be thrown (NumberFormat, NullPointer, EOF, etc.) |
| StringBuilder sb = new StringBuilder(); |
| sb.append("Error replaying edit log at offset " + tracker.getPos()); |
| if (recentOpcodeOffsets[0] != -1) { |
| Arrays.sort(recentOpcodeOffsets); |
| sb.append("\nRecent opcode offsets:"); |
| for (long offset : recentOpcodeOffsets) { |
| if (offset != -1) { |
| sb.append(' ').append(offset); |
| } |
| } |
| } |
| String errorMessage = sb.toString(); |
| FSImage.LOG.error(errorMessage); |
| throw new IOException(errorMessage, t); |
| } |
| if (FSImage.LOG.isDebugEnabled()) { |
| FSImage.LOG.debug("numOpAdd = " + numOpAdd + " numOpClose = " + numOpClose |
| + " numOpDelete = " + numOpDelete |
| + " numOpRenameOld = " + numOpRenameOld |
| + " numOpSetRepl = " + numOpSetRepl + " numOpMkDir = " + numOpMkDir |
| + " numOpSetPerm = " + numOpSetPerm |
| + " numOpSetOwner = " + numOpSetOwner |
| + " numOpSetGenStamp = " + numOpSetGenStamp |
| + " numOpTimes = " + numOpTimes |
| + " numOpConcatDelete = " + numOpConcatDelete |
| + " numOpRename = " + numOpRename |
| + " numOpGetDelegationToken = " + numOpGetDelegationToken |
| + " numOpRenewDelegationToken = " + numOpRenewDelegationToken |
| + " numOpCancelDelegationToken = " + numOpCancelDelegationToken |
| + " numOpUpdateMasterKey = " + numOpUpdateMasterKey |
| + " numOpReassignLease = " + numOpReassignLease |
| + " numOpOther = " + numOpOther); |
| } |
| return numEdits; |
| } |
| |
| /** |
| * Validate a transaction's checksum |
| */ |
| private static void validateChecksum( |
| DataInputStream in, Checksum checksum, int tid) |
| throws IOException { |
| if (checksum != null) { |
| int calculatedChecksum = (int)checksum.getValue(); |
| int readChecksum = in.readInt(); // read in checksum |
| if (readChecksum != calculatedChecksum) { |
| throw new ChecksumException( |
| "Transaction " + tid + " is corrupt. Calculated checksum is " + |
| calculatedChecksum + " but read checksum " + readChecksum, tid); |
| } |
| } |
| } |
| |
| /** |
| * A class to read in blocks stored in the old format. The only two |
| * fields in the block were blockid and length. |
| */ |
| static class BlockTwo implements Writable { |
| long blkid; |
| long len; |
| |
| static { // register a ctor |
| WritableFactories.setFactory |
| (BlockTwo.class, |
| new WritableFactory() { |
| public Writable newInstance() { return new BlockTwo(); } |
| }); |
| } |
| |
| |
| BlockTwo() { |
| blkid = 0; |
| len = 0; |
| } |
| ///////////////////////////////////// |
| // Writable |
| ///////////////////////////////////// |
| public void write(DataOutput out) throws IOException { |
| out.writeLong(blkid); |
| out.writeLong(len); |
| } |
| |
| public void readFields(DataInput in) throws IOException { |
| this.blkid = in.readLong(); |
| this.len = in.readLong(); |
| } |
| } |
| |
| /** This method is defined for compatibility reason. */ |
| static private DatanodeDescriptor[] readDatanodeDescriptorArray(DataInput in |
| ) throws IOException { |
| DatanodeDescriptor[] locations = new DatanodeDescriptor[in.readInt()]; |
| for (int i = 0; i < locations.length; i++) { |
| locations[i] = new DatanodeDescriptor(); |
| locations[i].readFieldsFromFSEditLog(in); |
| } |
| return locations; |
| } |
| |
| static private short readShort(DataInputStream in) throws IOException { |
| return Short.parseShort(FSImageSerialization.readString(in)); |
| } |
| |
| static private long readLong(DataInputStream in) throws IOException { |
| return Long.parseLong(FSImageSerialization.readString(in)); |
| } |
| |
| // a place holder for reading a long |
| private static final LongWritable longWritable = new LongWritable(); |
| |
| /** Read an integer from an input stream */ |
| private static long readLongWritable(DataInputStream in) throws IOException { |
| synchronized (longWritable) { |
| longWritable.readFields(in); |
| return longWritable.get(); |
| } |
| } |
| |
| static Rename[] readRenameOptions(DataInputStream in) throws IOException { |
| BytesWritable writable = new BytesWritable(); |
| writable.readFields(in); |
| |
| byte[] bytes = writable.getBytes(); |
| Rename[] options = new Rename[bytes.length]; |
| |
| for (int i = 0; i < bytes.length; i++) { |
| options[i] = Rename.valueOf(bytes[i]); |
| } |
| return options; |
| } |
| |
| static private BlockInfo[] readBlocks( |
| DataInputStream in, |
| int logVersion, |
| boolean isFileUnderConstruction, |
| short replication) throws IOException { |
| int numBlocks = in.readInt(); |
| BlockInfo[] blocks = new BlockInfo[numBlocks]; |
| Block blk = new Block(); |
| BlockTwo oldblk = new BlockTwo(); |
| for (int i = 0; i < numBlocks; i++) { |
| if (logVersion <= -14) { |
| blk.readFields(in); |
| } else { |
| oldblk.readFields(in); |
| blk.set(oldblk.blkid, oldblk.len, |
| GenerationStamp.GRANDFATHER_GENERATION_STAMP); |
| } |
| if(isFileUnderConstruction && i == numBlocks-1) |
| blocks[i] = new BlockInfoUnderConstruction(blk, replication); |
| else |
| blocks[i] = new BlockInfo(blk, replication); |
| } |
| return blocks; |
| } |
| |
| /** |
| * Throw appropriate exception during upgrade from 203, when editlog loading |
| * could fail due to opcode conflicts. |
| */ |
| private void check203UpgradeFailure(int logVersion, IOException ex) |
| throws IOException { |
| // 0.20.203 version version has conflicting opcodes with the later releases. |
| // The editlog must be emptied by restarting the namenode, before proceeding |
| // with the upgrade. |
| if (Storage.is203LayoutVersion(logVersion) |
| && logVersion != FSConstants.LAYOUT_VERSION) { |
| String msg = "During upgrade failed to load the editlog version " |
| + logVersion + " from release 0.20.203. Please go back to the old " |
| + " release and restart the namenode. This empties the editlog " |
| + " and saves the namespace. Resume the upgrade after this step."; |
| throw new IOException(msg, ex); |
| } else { |
| throw ex; |
| } |
| } |
| |
| /** |
| * Stream wrapper that keeps track of the current file position. |
| */ |
| private static class PositionTrackingInputStream extends FilterInputStream { |
| private long curPos = 0; |
| private long markPos = -1; |
| |
| public PositionTrackingInputStream(InputStream is) { |
| super(is); |
| } |
| |
| public int read() throws IOException { |
| int ret = super.read(); |
| if (ret != -1) curPos++; |
| return ret; |
| } |
| |
| public int read(byte[] data) throws IOException { |
| int ret = super.read(data); |
| if (ret > 0) curPos += ret; |
| return ret; |
| } |
| |
| public int read(byte[] data, int offset, int length) throws IOException { |
| int ret = super.read(data, offset, length); |
| if (ret > 0) curPos += ret; |
| return ret; |
| } |
| |
| public void mark(int limit) { |
| super.mark(limit); |
| markPos = curPos; |
| } |
| |
| public void reset() throws IOException { |
| if (markPos == -1) { |
| throw new IOException("Not marked!"); |
| } |
| super.reset(); |
| curPos = markPos; |
| markPos = -1; |
| } |
| |
| public long getPos() { |
| return curPos; |
| } |
| } |
| } |
