class SCMContainerManager implements ContainerManager {

  @Override
  public ContainerInfo allocateContainer(final ReplicationType type,
      final ReplicationFactor replicationFactor, final String owner)
      throws IOException {
    try {
      lock.lock();
      ContainerInfo containerInfo = null;
      try {
        containerInfo =
            containerStateManager.allocateContainer(pipelineManager, type,
              replicationFactor, owner);
      } catch (IOException ex) {
        scmContainerManagerMetrics.incNumFailureCreateContainers();
        throw ex;
      }
      // Add container to DB.
      try {
        addContainerToDB(containerInfo);
      } catch (IOException ex) {
        // When adding to DB failed, we are removing from containerStateMap.
        // We should also remove from pipeline2Container Map in
        // PipelineStateManager.
        pipelineManager.removeContainerFromPipeline(
            containerInfo.getPipelineID(),
            new ContainerID(containerInfo.getContainerID()));
        throw ex;
      }
      return containerInfo;
    } finally {
      lock.unlock();
    }
  }

  public  SCMContainerManager(
      final ConfigurationSource conf,
      Table<ContainerID, ContainerInfo> containerStore,
      BatchOperationHandler batchHandler,
      PipelineManager pipelineManager);

}