Hadoop-2.6.0NodeManager Restart Recover实现分析（二）

      继上篇《Hadoop-2.6.0NodeManager Restart Recover实现分析（二）》。

      4、NMLeveldbStateStoreService实现分析

      在

      1）、initStorage()

      initStorage()方法中，完成了存储相关的初始化，如下：

  @Override
  protected void initStorage(Configuration conf)
      throws IOException {
    Path storeRoot = createStorageDir(conf);
    Options options = new Options();
    options.createIfMissing(false);
    options.logger(new LeveldbLogger());
    LOG.info("Using state database at " + storeRoot + " for recovery");
    File dbfile = new File(storeRoot.toString());
    try {
      db = JniDBFactory.factory.open(dbfile, options);
    } catch (NativeDB.DBException e) {
      if (e.isNotFound() || e.getMessage().contains(" does not exist ")) {
        LOG.info("Creating state database at " + dbfile);
        isNewlyCreated = true;
        options.createIfMissing(true);
        try {
          db = JniDBFactory.factory.open(dbfile, options);
          // store version
          storeVersion();
        } catch (DBException dbErr) {
          throw new IOException(dbErr.getMessage(), dbErr);
        }
      } else {
        throw e;
      }
    }
    checkVersion();
  }

      最主要的是通过JniDBFactory.factory.open(dbfile, options)打开了一个db实例。

      2）startStorage()

      为空。

      3）closeStorage()

      关闭db，如下：

  @Override
  protected void closeStorage() throws IOException {
    if (db != null) {
      db.close();
    }
  }

      下面，以Applications为例，罗列下实现方式：

      1）storeApplication()

      存储Application，如下：

  @Override
  public void storeApplication(ApplicationId appId,
      ContainerManagerApplicationProto p) throws IOException {
    String key = APPLICATIONS_KEY_PREFIX + appId;
    try {
      db.put(bytes(key), p.toByteArray());
    } catch (DBException e) {
      throw new IOException(e);
    }
  }

      key为ContainerManager/applications/再加appId，value为ContainerManagerApplicationProto，消息内容如下：

message ContainerManagerApplicationProto {
  optional ApplicationIdProto id = 1;
  optional string user = 2;
  optional bytes credentials = 3;
  repeated ApplicationACLMapProto acls = 4;
  optional LogAggregationContextProto log_aggregation_context = 5;
}

      2）storeFinishedApplication()

      存储已完成Application，如下：

  @Override
  public void storeFinishedApplication(ApplicationId appId)
      throws IOException {
    String key = FINISHED_APPS_KEY_PREFIX + appId;
    try {
      db.put(bytes(key), new byte[0]);
    } catch (DBException e) {
      throw new IOException(e);
    }
  }

      其中，key为ContainerManager/finishedApps/再加appId，value为空，即new byte[0]。
      3）removeApplication()

      删除Application，如下：

  @Override
  public void removeApplication(ApplicationId appId)
      throws IOException {
    try {
      WriteBatch batch = db.createWriteBatch();
      try {
        String key = APPLICATIONS_KEY_PREFIX + appId;
        batch.delete(bytes(key));
        key = FINISHED_APPS_KEY_PREFIX + appId;
        batch.delete(bytes(key));
        db.write(batch);
      } finally {
        batch.close();
      }
    } catch (DBException e) {
      throw new IOException(e);
    }
  }

      根据各自的key分别删除storeApplication()和storeFinishedApplication()存储的Application信息，

      4）loadApplicationsState()
      加载所有的Application状态，以便进行recovery。如下：

  @Override
  public RecoveredApplicationsState loadApplicationsState()
      throws IOException {
    RecoveredApplicationsState state = new RecoveredApplicationsState();
    state.applications = new ArrayList<ContainerManagerApplicationProto>();
    String keyPrefix = APPLICATIONS_KEY_PREFIX;
    LeveldbIterator iter = null;
    try {
      iter = new LeveldbIterator(db);
      iter.seek(bytes(keyPrefix));
      while (iter.hasNext()) {
        Entry<byte[], byte[]> entry = iter.next();
        String key = asString(entry.getKey());
        if (!key.startsWith(keyPrefix)) {
          break;
        }
        state.applications.add(
            ContainerManagerApplicationProto.parseFrom(entry.getValue()));
      }

      state.finishedApplications = new ArrayList<ApplicationId>();
      keyPrefix = FINISHED_APPS_KEY_PREFIX;
      iter.seek(bytes(keyPrefix));
      while (iter.hasNext()) {
        Entry<byte[], byte[]> entry = iter.next();
        String key = asString(entry.getKey());
        if (!key.startsWith(keyPrefix)) {
          break;
        }
        ApplicationId appId =
            ConverterUtils.toApplicationId(key.substring(keyPrefix.length()));
        state.finishedApplications.add(appId);
      }
    } catch (DBException e) {
      throw new IOException(e);
    } finally {
      if (iter != null) {
        iter.close();
      }
    }

    return state;
  }

      通过LeveldbIterator和key的前缀ContainerManager/applications/进行load。

      后续会分析哪些地方调用了上述方法，未完待续！