/**
    * 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.coordination;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.hadoop.hbase.classification.InterfaceAudience;
  import org.apache.hadoop.hbase.CoordinatedStateManager;
  import org.apache.hadoop.hbase.HRegionInfo;
  import org.apache.hadoop.hbase.ServerName;
  import org.apache.hadoop.hbase.executor.EventType;
  import org.apache.hadoop.hbase.master.AssignmentManager;
  import org.apache.hadoop.hbase.master.RegionState;
  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos;
  import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos;
  import org.apache.hadoop.hbase.regionserver.HRegion;
  import org.apache.hadoop.hbase.regionserver.RegionServerServices;
  import org.apache.hadoop.hbase.zookeeper.ZKAssign;
  import org.apache.hadoop.hbase.zookeeper.ZKUtil;
  import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
  import org.apache.zookeeper.KeeperException;
  
  import java.io.IOException;
  
  /**
   * ZK-based implementation of {@link OpenRegionCoordination}.
   */
  @InterfaceAudience.Private
  public class ZkOpenRegionCoordination implements OpenRegionCoordination {
    private static final Log LOG = LogFactory.getLog(ZkOpenRegionCoordination.class);
  
    private CoordinatedStateManager coordination;
    private final ZooKeeperWatcher watcher;
  
    public ZkOpenRegionCoordination(CoordinatedStateManager coordination,
                                    ZooKeeperWatcher watcher) {
      this.coordination = coordination;
      this.watcher = watcher;
    }
  
    //-------------------------------
    // Region Server-side operations
    //-------------------------------
  
    /**
     * @param r Region we're working on.
     * @return whether znode is successfully transitioned to OPENED state.
     * @throws java.io.IOException
     */
    @Override
    public boolean transitionToOpened(final HRegion r, OpenRegionDetails ord) throws IOException {
      ZkOpenRegionDetails zkOrd = (ZkOpenRegionDetails) ord;
  
      boolean result = false;
      HRegionInfo hri = r.getRegionInfo();
      final String name = hri.getRegionNameAsString();
      // Finally, Transition ZK node to OPENED
      try {
        if (ZKAssign.transitionNodeOpened(watcher, hri,
          zkOrd.getServerName(), zkOrd.getVersion()) == -1) {
          String warnMsg = "Completed the OPEN of region " + name +
            " but when transitioning from " + " OPENING to OPENED ";
          try {
            String node = ZKAssign.getNodeName(watcher, hri.getEncodedName());
            if (ZKUtil.checkExists(watcher, node) < 0) {
              // if the znode
              coordination.getServer().abort(warnMsg + "the znode disappeared", null);
            } else {
              LOG.warn(warnMsg + "got a version mismatch, someone else clashed; " +
                "so now unassigning -- closing region on server: " + zkOrd.getServerName());
            }
          } catch (KeeperException ke) {
            coordination.getServer().abort(warnMsg, ke);
          }
        } else {
          LOG.debug("Transitioned " + r.getRegionInfo().getEncodedName() +
            " to OPENED in zk on " + zkOrd.getServerName());
          result = true;
        }
      } catch (KeeperException e) {
        LOG.error("Failed transitioning node " + name +
          " from OPENING to OPENED -- closing region", e);
      }
      return result;
   }
 
   /**
    * Transition ZK node from OFFLINE to OPENING.
    * @param regionInfo region info instance
    * @param ord - instance of open region details, for ZK implementation
    *   will include version Of OfflineNode that needs to be compared
    *   before changing the node's state from OFFLINE
    * @return True if successful transition.
    */
   @Override
   public boolean transitionFromOfflineToOpening(HRegionInfo regionInfo,
                                                 OpenRegionDetails ord) {
     ZkOpenRegionDetails zkOrd = (ZkOpenRegionDetails) ord;
 
     // encoded name is used as znode encoded name in ZK
     final String encodedName = regionInfo.getEncodedName();
 
     // TODO: should also handle transition from CLOSED?
     try {
       // Initialize the znode version.
       zkOrd.setVersion(ZKAssign.transitionNode(watcher, regionInfo,
         zkOrd.getServerName(), EventType.M_ZK_REGION_OFFLINE,
         EventType.RS_ZK_REGION_OPENING, zkOrd.getVersionOfOfflineNode()));
     } catch (KeeperException e) {
       LOG.error("Error transition from OFFLINE to OPENING for region=" +
         encodedName, e);
       zkOrd.setVersion(-1);
       return false;
     }
     boolean b = isGoodVersion(zkOrd);
     if (!b) {
       LOG.warn("Failed transition from OFFLINE to OPENING for region=" +
         encodedName);
     }
     return b;
   }
 
   /**
    * Update our OPENING state in zookeeper.
    * Do this so master doesn't timeout this region-in-transition.
    * We may lose the znode ownership during the open.  Currently its
    * too hard interrupting ongoing region open.  Just let it complete
    * and check we still have the znode after region open.
    *
    * @param context Some context to add to logs if failure
    * @return True if successful transition.
    */
   @Override
   public boolean tickleOpening(OpenRegionDetails ord, HRegionInfo regionInfo,
                                RegionServerServices rsServices, final String context) {
     ZkOpenRegionDetails zkOrd = (ZkOpenRegionDetails) ord;
     if (!isRegionStillOpening(regionInfo, rsServices)) {
       LOG.warn("Open region aborted since it isn't opening any more");
       return false;
     }
     // If previous checks failed... do not try again.
     if (!isGoodVersion(zkOrd)) return false;
     String encodedName = regionInfo.getEncodedName();
     try {
       zkOrd.setVersion(ZKAssign.confirmNodeOpening(watcher,
           regionInfo, zkOrd.getServerName(), zkOrd.getVersion()));
     } catch (KeeperException e) {
       coordination.getServer().abort("Exception refreshing OPENING; region=" + encodedName +
         ", context=" + context, e);
       zkOrd.setVersion(-1);
       return false;
     }
     boolean b = isGoodVersion(zkOrd);
     if (!b) {
       LOG.warn("Failed refreshing OPENING; region=" + encodedName +
         ", context=" + context);
     }
     return b;
   }
 
   /**
    * Try to transition to open.
    *
    * This is not guaranteed to succeed, we just do our best.
    *
    * @param rsServices
    * @param hri Region we're working on.
    * @param ord Details about region open task
    * @return whether znode is successfully transitioned to FAILED_OPEN state.
    */
   @Override
   public boolean tryTransitionFromOfflineToFailedOpen(RegionServerServices rsServices,
                                                       final HRegionInfo hri,
                                                       OpenRegionDetails ord) {
     ZkOpenRegionDetails zkOrd = (ZkOpenRegionDetails) ord;
     boolean result = false;
     final String name = hri.getRegionNameAsString();
     try {
       LOG.info("Opening of region " + hri + " failed, transitioning" +
         " from OFFLINE to FAILED_OPEN in ZK, expecting version " +
         zkOrd.getVersionOfOfflineNode());
       if (ZKAssign.transitionNode(
         rsServices.getZooKeeper(), hri,
         rsServices.getServerName(),
         EventType.M_ZK_REGION_OFFLINE,
         EventType.RS_ZK_REGION_FAILED_OPEN,
         zkOrd.getVersionOfOfflineNode()) == -1) {
         LOG.warn("Unable to mark region " + hri + " as FAILED_OPEN. " +
           "It's likely that the master already timed out this open " +
           "attempt, and thus another RS already has the region.");
       } else {
         result = true;
       }
     } catch (KeeperException e) {
       LOG.error("Failed transitioning node " + name + " from OFFLINE to FAILED_OPEN", e);
     }
     return result;
   }
 
   private boolean isGoodVersion(ZkOpenRegionDetails zkOrd) {
     return zkOrd.getVersion() != -1;
   }
 
   /**
    * This is not guaranteed to succeed, we just do our best.
    * @param hri Region we're working on.
    * @return whether znode is successfully transitioned to FAILED_OPEN state.
    */
   @Override
   public boolean tryTransitionFromOpeningToFailedOpen(final HRegionInfo hri,
                                                       OpenRegionDetails ord) {
     ZkOpenRegionDetails zkOrd = (ZkOpenRegionDetails) ord;
     boolean result = false;
     final String name = hri.getRegionNameAsString();
     try {
       LOG.info("Opening of region " + hri + " failed, transitioning" +
         " from OPENING to FAILED_OPEN in ZK, expecting version " + zkOrd.getVersion());
       if (ZKAssign.transitionNode(
         watcher, hri,
         zkOrd.getServerName(),
         EventType.RS_ZK_REGION_OPENING,
         EventType.RS_ZK_REGION_FAILED_OPEN,
         zkOrd.getVersion()) == -1) {
         LOG.warn("Unable to mark region " + hri + " as FAILED_OPEN. " +
           "It's likely that the master already timed out this open " +
           "attempt, and thus another RS already has the region.");
       } else {
         result = true;
       }
     } catch (KeeperException e) {
       LOG.error("Failed transitioning node " + name +
         " from OPENING to FAILED_OPEN", e);
     }
     return result;
   }
 
   /**
    * Parse ZK-related fields from request.
    */
   @Override
   public OpenRegionCoordination.OpenRegionDetails parseFromProtoRequest(
       AdminProtos.OpenRegionRequest.RegionOpenInfo regionOpenInfo) {
     ZkOpenRegionCoordination.ZkOpenRegionDetails zkCrd =
       new ZkOpenRegionCoordination.ZkOpenRegionDetails();
 
     int versionOfOfflineNode = -1;
     if (regionOpenInfo.hasVersionOfOfflineNode()) {
       versionOfOfflineNode = regionOpenInfo.getVersionOfOfflineNode();
     }
     zkCrd.setVersionOfOfflineNode(versionOfOfflineNode);
     zkCrd.setServerName(coordination.getServer().getServerName());
 
     return zkCrd;
   }
 
   /**
    * No ZK tracking will be performed for that case.
    * This method should be used when we want to construct CloseRegionDetails,
    * but don't want any coordination on that (when it's initiated by regionserver),
    * so no znode state transitions will be performed.
    */
   @Override
   public OpenRegionCoordination.OpenRegionDetails getDetailsForNonCoordinatedOpening() {
     ZkOpenRegionCoordination.ZkOpenRegionDetails zkCrd =
       new ZkOpenRegionCoordination.ZkOpenRegionDetails();
     zkCrd.setVersionOfOfflineNode(-1);
     zkCrd.setServerName(coordination.getServer().getServerName());
 
     return zkCrd;
   }
 
   //--------------------------
   // HMaster-side operations
   //--------------------------
   @Override
   public boolean commitOpenOnMasterSide(AssignmentManager assignmentManager,
                                         HRegionInfo regionInfo,
                                         OpenRegionDetails ord) {
     boolean committedSuccessfully = true;
 
     // Code to defend against case where we get SPLIT before region open
     // processing completes; temporary till we make SPLITs go via zk -- 0.92.
     RegionState regionState = assignmentManager.getRegionStates()
       .getRegionTransitionState(regionInfo.getEncodedName());
     boolean openedNodeDeleted = false;
     if (regionState != null && regionState.isOpened()) {
       openedNodeDeleted = deleteOpenedNode(regionInfo, ord);
       if (!openedNodeDeleted) {
         LOG.error("Znode of region " + regionInfo.getShortNameToLog() + " could not be deleted.");
       }
     } else {
       LOG.warn("Skipping the onlining of " + regionInfo.getShortNameToLog() +
         " because regions is NOT in RIT -- presuming this is because it SPLIT");
     }
     if (!openedNodeDeleted) {
       if (assignmentManager.getTableStateManager().isTableState(regionInfo.getTable(),
           ZooKeeperProtos.Table.State.DISABLED, ZooKeeperProtos.Table.State.DISABLING)) {
         debugLog(regionInfo, "Opened region "
           + regionInfo.getShortNameToLog() + " but "
           + "this table is disabled, triggering close of region");
         committedSuccessfully = false;
       }
     }
 
     return committedSuccessfully;
   }
 
   private boolean deleteOpenedNode(HRegionInfo regionInfo, OpenRegionDetails ord) {
     ZkOpenRegionDetails zkOrd = (ZkOpenRegionDetails) ord;
     int expectedVersion = zkOrd.getVersion();
 
     debugLog(regionInfo, "Handling OPENED of " +
       regionInfo.getShortNameToLog() + " from " + zkOrd.getServerName().toString() +
       "; deleting unassigned node");
     try {
       // delete the opened znode only if the version matches.
       return ZKAssign.deleteNode(this.coordination.getServer().getZooKeeper(),
         regionInfo.getEncodedName(), EventType.RS_ZK_REGION_OPENED, expectedVersion);
     } catch(KeeperException.NoNodeException e){
       // Getting no node exception here means that already the region has been opened.
       LOG.warn("The znode of the region " + regionInfo.getShortNameToLog() +
         " would have already been deleted");
       return false;
     } catch (KeeperException e) {
       this.coordination.getServer().abort("Error deleting OPENED node in ZK (" +
         regionInfo.getRegionNameAsString() + ")", e);
     }
     return false;
   }
 
   private void debugLog(HRegionInfo region, String string) {
     if (region.isMetaTable()) {
       LOG.info(string);
     } else {
       LOG.debug(string);
     }
   }
 
   // Additional classes and helper methods
 
   /**
    * ZK-based implementation. Has details about whether the state transition should be
    * reflected in ZK, as well as expected version of znode.
    */
   public static class ZkOpenRegionDetails implements OpenRegionCoordination.OpenRegionDetails {
 
     // We get version of our znode at start of open process and monitor it across
     // the total open. We'll fail the open if someone hijacks our znode; we can
     // tell this has happened if version is not as expected.
     private volatile int version = -1;
 
     //version of the offline node that was set by the master
     private volatile int versionOfOfflineNode = -1;
 
     /**
      * Server name the handler is running on.
      */
     private ServerName serverName;
 
     public ZkOpenRegionDetails() {
     }
 
     public ZkOpenRegionDetails(int versionOfOfflineNode) {
       this.versionOfOfflineNode = versionOfOfflineNode;
     }
 
     public int getVersionOfOfflineNode() {
       return versionOfOfflineNode;
     }
 
     public void setVersionOfOfflineNode(int versionOfOfflineNode) {
       this.versionOfOfflineNode = versionOfOfflineNode;
     }
 
     public int getVersion() {
       return version;
     }
 
     public void setVersion(int version) {
       this.version = version;
     }
 
     @Override
     public ServerName getServerName() {
       return serverName;
     }
 
     @Override
     public void setServerName(ServerName serverName) {
       this.serverName = serverName;
     }
   }
 
   private boolean isRegionStillOpening(HRegionInfo regionInfo, RegionServerServices rsServices) {
     byte[] encodedName = regionInfo.getEncodedNameAsBytes();
     Boolean action = rsServices.getRegionsInTransitionInRS().get(encodedName);
     return Boolean.TRUE.equals(action); // true means opening for RIT
   }
 }