content: /**
   * 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 com.alibaba.wasp.master.handler;
  
  import com.alibaba.wasp.EntityGroupInfo;
  import com.alibaba.wasp.FConstants;
  import com.alibaba.wasp.Server;
  import com.alibaba.wasp.ServerName;
  import com.alibaba.wasp.executor.EventHandler;
  import com.alibaba.wasp.master.AssignmentManager;
  import com.alibaba.wasp.master.DeadServer;
  import com.alibaba.wasp.master.EntityGroupState;
  import com.alibaba.wasp.master.EntityGroupStates;
  import com.alibaba.wasp.master.FMasterServices;
  import com.alibaba.wasp.meta.FMetaEditor;
  import com.alibaba.wasp.meta.FMetaReader;
  import com.alibaba.wasp.meta.FMetaScanner;
  import com.alibaba.wasp.meta.FMetaVisitor;
  import com.alibaba.wasp.zookeeper.ZKAssign;
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.hbase.client.Result;
  import org.apache.hadoop.hbase.util.Bytes;
  import org.apache.hadoop.hbase.util.PairOfSameType;
  import org.apache.zookeeper.KeeperException;
  
  import java.io.IOException;
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Map;
  import java.util.NavigableMap;
  
  /**
   * Process server shutdown. Server-to-handle must be already in the deadservers
   * lists.
   */
  public class ServerShutdownHandler extends EventHandler {
    private static final Log LOG = LogFactory.getLog(ServerShutdownHandler.class);
    private final ServerName serverName;
    private final AssignmentManager assignmentManager;
    private final DeadServer deadServers;
  
    public ServerShutdownHandler(Server server, FMasterServices services,
        DeadServer deadservers, ServerName serverName) {
      super(server, EventType.M_SERVER_SHUTDOWN);
      this.serverName = serverName;
      this.assignmentManager = services.getAssignmentManager();
      this.deadServers = deadservers;
      if (!this.deadServers.contains(this.serverName)) {
        LOG.warn(this.serverName + " is NOT in deadservers; it should be!");
      }
    }
  
    @Override
    public void process() throws IOException {
      final ServerName serverName = this.serverName;
      try {
        NavigableMap<EntityGroupInfo, Result> egis = null;
        while (!this.server.isStopped()) {
          try {
            egis = FMetaReader.getServerUserEntityGroups(
                server.getConfiguration(), this.serverName);
            break;
          } catch (IOException ioe) {
            LOG.info(
                "Received exception accessing META during server shutdown of "
                    + serverName + ", retrying META read", ioe);
          }
        }
        if (this.server.isStopped()) {
          throw new IOException("Server is stopped");
        }
  
        // Clean out anything in entityGroups in transition. Being conservative
        // and
        // doing after log splitting. Could do some states before -- OPENING?
        // OFFLINE? -- and then others after like CLOSING that depend on log
        // splitting.
        List<EntityGroupState> entityGroupsInTransition = assignmentManager
            .processServerShutdown(serverName);
        LOG.info("Reassigning " + ((egis == null) ? 0 : egis.size())
            + " entityGroup(s) that "
            + (serverName == null ? "null" : serverName)
            + " was carrying (skipping " + entityGroupsInTransition.size()
            + " entityGroup(s) that are already in transition)");
  
        // Iterate entityGroups that were on this server and assign them
        if (egis != null) {
          EntityGroupStates entityGroupStates = assignmentManager
              .getEntityGroupStates();
          List<EntityGroupInfo> toAssignEntityGroups = new ArrayList<EntityGroupInfo>();
          for (Map.Entry<EntityGroupInfo, Result> e : egis.entrySet()) {
            EntityGroupInfo egi = e.getKey();
            EntityGroupState egit = entityGroupStates
                .getEntityGroupTransitionState(egi);
            if (processDeadEntityGroup(egi, e.getValue(), assignmentManager,
                server)) {
              ServerName addressFromAM = entityGroupStates
                  .getFServerOfEntityGroup(egi);
              if (addressFromAM != null && !addressFromAM.equals(this.serverName)) {
                // If this entityGroup is in transition on the dead server, it
                // must be
                // opening or pending_open, which is covered by
                // AM#processServerShutdown
                LOG.debug("Skip assigning entityGroup "
                    + egi.getEntityGroupNameAsString()
                    + " because it has been opened in "
                    + addressFromAM.getServerName());
                continue;
              }
              if (egit != null) {
                if (!egit.isOnServer(serverName) || egit.isClosed()
                    || egit.isOpened() || egit.isSplit()) {
                  // Skip entityGroups that are in transition on other server,
                  // or in state closed/opened/split
                  LOG.info("Skip assigning entityGroup " + egit);
                  continue;
                }
                try {
                  // clean zk node
                  LOG.info("Reassigning entityGroup with eg = " + egit
                      + " and deleting zk node if exists");
                  ZKAssign.deleteNodeFailSilent(server.getZooKeeper(), egi);
                } catch (KeeperException ke) {
                  this.server.abort(
                      "Unexpected ZK exception deleting unassigned node " + egi,
                      ke);
                  return;
                }
              }
              toAssignEntityGroups.add(egi);
            } else if (egit != null) {
              if (egit.isSplitting() || egit.isSplit()) {
                // This will happen when the FServer went down and the call back
                // for
                // the SPLIITING or SPLIT
                // has not yet happened for node Deleted event. In that case if
                // the EntityGroup was actually
                // split
                // but the FServer had gone down before completing the split
                // process
                // then will not try to
                // assign the parent EntityGroup again. In that case we should
                // make the
                // EntityGroup offline and
                // also delete the EntityGroup from EGIT.
                assignmentManager.entityGroupOffline(egi);
              } else if ((egit.isClosing() || egit.isPendingClose())
                  && assignmentManager.getZKTable().isDisablingOrDisabledTable(
                      egi.getTableNameAsString())) {
                // If the table was partially disabled and the FServer went down,
                // we
                // should clear the EGIT
                // and remove the node for the EntityGroup.
                // The egit that we use may be stale in case the table was in
                // DISABLING state
                // but though we did assign we will not be clearing the znode in
                // CLOSING state.
                // Doing this will have no harm. See HBASE-5927
                assignmentManager.deleteClosingOrClosedNode(egi);
                assignmentManager.entityGroupOffline(egi);
              } else {
                LOG.warn("THIS SHOULD NOT HAPPEN: unexpected entityGroup in transition "
                    + egit + " not to be assigned by SSH of server " + serverName);
              }
            }
          }
          try {
            assignmentManager.assign(toAssignEntityGroups);
          } catch (InterruptedException ie) {
            LOG.error("Caught " + ie + " during round-robin assignment");
            throw new IOException(ie);
          }
        }
      } finally {
        this.deadServers.finish(serverName);
      }
      LOG.info("Finished processing of shutdown of " + serverName);
    }
  
    /**
     * Process a dead EntityGroup from a dead FServer. Checks if the EntityGroup
     * is disabled or disabling or if the EntityGroup has a partially completed
     * split.
     * 
     * @param egi
     * @param result
     * @param assignmentManager
     * @param server
     * @return Returns true if specified EntityGroup should be assigned, false if
     *         not.
     * @throws java.io.IOException
     */
    public boolean processDeadEntityGroup(EntityGroupInfo egi, Result result,
        AssignmentManager assignmentManager, Server server) throws IOException {
      boolean tablePresent = assignmentManager.getZKTable().isTablePresent(
          egi.getTableNameAsString());
      if (!tablePresent) {
        LOG.info("The table " + egi.getTableNameAsString()
            + " was deleted.  Hence not proceeding.");
        return false;
      }
      // If table is not disabled but the EntityGroup is offlined,
      boolean disabled = assignmentManager.getZKTable().isDisabledTable(
          egi.getTableNameAsString());
      if (disabled) {
        LOG.info("The table " + egi.getTableNameAsString()
            + " was disabled.  Hence not proceeding.");
        return false;
      }
      if (egi.isOffline() && egi.isSplit()) {
        LOG.debug("Offlined and split entityGroup "
            + egi.getEntityGroupNameAsString() + "; checking daughter presence");
        if (FMetaReader.getEntityGroupAndLocation(server.getConfiguration(),
            egi.getEntityGroupName()) == null) {
          return false;
        }
        fixupDaughters(result, assignmentManager, server.getConfiguration());
        return false;
      }
      boolean disabling = assignmentManager.getZKTable().isDisablingTable(
          egi.getTableNameAsString());
      if (disabling) {
        LOG.info("The table " + egi.getTableNameAsString()
            + " is disabled.  Hence not assigning entityGroup"
            + egi.getEncodedName());
        return false;
      }
      return true;
    }
  
    /**
     * Check that daughter entityGroups are up in .FMETA. and if not, add them.
     *
     * @param result
     *          The contents of the parent row in .META.
     * @return the number of daughters missing and fixed
     * @throws java.io.IOException
     */
    public static int fixupDaughters(final Result result,
        final AssignmentManager assignmentManager, final Configuration conf)
        throws IOException {
      PairOfSameType<EntityGroupInfo> daughters = EntityGroupInfo
          .getDaughterEntityGroups(result);
      int fixedA = fixupDaughter(result, daughters.getFirst(), assignmentManager,
          conf);
      int fixedB = fixupDaughter(result, daughters.getSecond(),
          assignmentManager, conf);
      return fixedA + fixedB;
    }
  
    /**
     * Check individual daughter is up in .FMETA.; fixup if its not.
     *
     * @param result
     *          The contents of the parent row in .FMETA.
     * @return 1 if the daughter is missing and fixed. Otherwise 0
     * @throws java.io.IOException
     */
    static int fixupDaughter(final Result result, EntityGroupInfo daughter,
        final AssignmentManager assignmentManager, final Configuration conf)
        throws IOException {
      if (daughter == null)
        return 0;
      if (isDaughterMissing(conf, daughter)) {
        LOG.info("Fixup; missing daughter "
            + daughter.getEntityGroupNameAsString());
        FMetaEditor.addDaughter(conf, daughter, null);
  
        // And assign it.
        assignmentManager.assign(daughter, true, true);
        return 1;
      } else {
        LOG.debug("Daughter " + daughter.getEntityGroupNameAsString()
            + " present");
      }
      return 0;
    }
  
    /**
     * Look for presence of the daughter OR of a split of the daughter in .FMETA.
     * Daughter could have been split over on fserver before a run of the
     * catalogJanitor had chance to clear reference from parent.
     *
     * @param daughter
     *          Daughter entityGroup to search for.
     * @throws java.io.IOException
     */
    private static boolean isDaughterMissing(final Configuration conf,
        final EntityGroupInfo daughter)
        throws IOException {
      FindDaughterVisitor visitor = new FindDaughterVisitor(daughter);
      // Start the scan at what should be the daughter's row in the .FMETA.
      // We will either 1., find the daughter or some derivative split of the
      // daughter (will have same table name and start row at least but will sort
      // after because has larger entityGroupid -- the entityGroupid is timestamp
      // of entityGroup
      // creation), OR, we will not find anything with same table name and start
      // row. If the latter, then assume daughter missing and do fixup.
      byte[] startrow = daughter.getTableName();
      FMetaScanner.fullScan(conf, visitor, startrow, null);
      return !visitor.foundDaughter();
    }
  
    /**
     * Looks for daughter. Sets a flag if daughter or some progeny of daughter is
     * found up in <code>.FMETA.</code>.
     */
    static class FindDaughterVisitor implements FMetaVisitor {
      private final EntityGroupInfo daughter;
      private boolean found = false;
  
      FindDaughterVisitor(final EntityGroupInfo daughter) {
        this.daughter = daughter;
      }
  
      /**
       * @return True if we found a daughter entityGroup during our visiting.
       */
      boolean foundDaughter() {
        return this.found;
      }
  
      @Override
      public boolean visit(Result r) throws IOException {
        EntityGroupInfo egi = EntityGroupInfo.getEntityGroupInfo(r);
        if (egi == null) {
          LOG.warn("No serialized EntityGroupInfo in " + r);
          return true;
        }
        byte[] value = r.getValue(FConstants.CATALOG_FAMILY,
            FConstants.EGLOCATION);
        // See if daughter is assigned to some server
        if (value == null)
          return false;
  
        // Now see if we have gone beyond the daughter's startrow.
        if (!Bytes.equals(daughter.getTableName(), egi.getTableName())) {
          // We fell into another table. Stop scanning.
          return false;
        }
        // If our start rows do not compare, move on.
        if (!Bytes.equals(daughter.getStartKey(), egi.getStartKey())) {
          return false;
        }
        // Else, table name and start rows compare. It means that the daughter
        // or some derivative split of the daughter is up in .FMETA. Daughter
        // exists.
        this.found = true;
        return false;
      }
    }
  
    @Override
    public String toString() {
      String name = "UnknownServerName";
      if (server != null && server.getServerName() != null) {
        name = server.getServerName().toString();
      }
      return getClass().getSimpleName() + "-" + name + "-" + getSeqid();
    }
  }
  