/*
 * 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.zookeeper;

import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.hadoop.hbase.Stoppable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Handles coordination of a single "leader" instance among many possible
 * candidates.  The first {@link ZKLeaderManager} to successfully create
 * the given znode becomes the leader, allowing the instance to continue
 * with whatever processing must be protected.  Other {@link ZKLeaderManager}
 * instances will wait to be notified of changes to the leader znode.
 * If the current master instance fails, the ephemeral leader znode will
 * be removed, and all waiting instances will be notified, with the race
 * to claim the leader znode beginning all over again.
 * @deprecated Not used
 */
@Deprecated
@InterfaceAudience.Private
public class ZKLeaderManager extends ZKListener {
  private static final Logger LOG = LoggerFactory.getLogger(ZKLeaderManager.class);

  private final Object lock = new Object();
  private final AtomicBoolean leaderExists = new AtomicBoolean();
  private final String leaderZNode;
  private final byte[] nodeId;
  private final Stoppable candidate;

  public ZKLeaderManager(ZKWatcher watcher, String leaderZNode,
                         byte[] identifier, Stoppable candidate) {
    super(watcher);
    this.leaderZNode = leaderZNode;
    this.nodeId = identifier;
    this.candidate = candidate;
  }

  public void start() {
    try {
      watcher.registerListener(this);
      String parent = ZKUtil.getParent(leaderZNode);
      if (ZKUtil.checkExists(watcher, parent) < 0) {
        ZKUtil.createWithParents(watcher, parent);
      }
    } catch (KeeperException ke) {
      watcher.abort("Unhandled zk exception when starting", ke);
      candidate.stop("Unhandled zk exception starting up: "+ke.getMessage());
    }
  }

  @Override
  public void nodeCreated(String path) {
    if (leaderZNode.equals(path) && !candidate.isStopped()) {
      handleLeaderChange();
    }
  }

  @Override
  public void nodeDeleted(String path) {
    if (leaderZNode.equals(path) && !candidate.isStopped()) {
      handleLeaderChange();
    }
  }

  private void handleLeaderChange() {
    try {
      synchronized(lock) {
        if (ZKUtil.watchAndCheckExists(watcher, leaderZNode)) {
          LOG.info("Found new leader for znode: {}", leaderZNode);
          leaderExists.set(true);
        } else {
          LOG.info("Leader change, but no new leader found");
          leaderExists.set(false);
          lock.notifyAll();
        }
      }
    } catch (KeeperException ke) {
      watcher.abort("ZooKeeper error checking for leader znode", ke);
      candidate.stop("ZooKeeper error checking for leader: "+ke.getMessage());
    }
  }

  /**
   * Blocks until this instance has claimed the leader ZNode in ZooKeeper
   */
  public void waitToBecomeLeader() {
    while (!candidate.isStopped()) {
      try {
        if (ZKUtil.createEphemeralNodeAndWatch(watcher, leaderZNode, nodeId)) {
          // claimed the leader znode
          leaderExists.set(true);
          if (LOG.isDebugEnabled()) {
            LOG.debug("Claimed the leader znode as '"+
                Bytes.toStringBinary(nodeId)+"'");
          }
          return;
        }

        // if claiming the node failed, there should be another existing node
        byte[] currentId = ZKUtil.getDataAndWatch(watcher, leaderZNode);
        if (currentId != null && Bytes.equals(currentId, nodeId)) {
          // claimed with our ID, but we didn't grab it, possibly restarted?
          LOG.info("Found existing leader with our ID ("+
              Bytes.toStringBinary(nodeId)+"), removing");
          ZKUtil.deleteNode(watcher, leaderZNode);
          leaderExists.set(false);
        } else {
          LOG.info("Found existing leader with ID: {}", Bytes.toStringBinary(nodeId));
          leaderExists.set(true);
        }
      } catch (KeeperException ke) {
        watcher.abort("Unexpected error from ZK, stopping candidate", ke);
        candidate.stop("Unexpected error from ZK: "+ke.getMessage());
        return;
      }

      // wait for next chance
      synchronized(lock) {
        while (leaderExists.get() && !candidate.isStopped()) {
          try {
            lock.wait();
          } catch (InterruptedException ie) {
            LOG.debug("Interrupted waiting on leader", ie);
          }
        }
      }
    }
  }

  /**
   * Removes the leader znode, if it is currently claimed by this instance.
   */
  public void stepDownAsLeader() {
    try {
      synchronized(lock) {
        if (!leaderExists.get()) {
          return;
        }
        byte[] leaderId = ZKUtil.getData(watcher, leaderZNode);
        if (leaderId != null && Bytes.equals(nodeId, leaderId)) {
          LOG.info("Stepping down as leader");
          ZKUtil.deleteNodeFailSilent(watcher, leaderZNode);
          leaderExists.set(false);
        } else {
          LOG.info("Not current leader, no need to step down");
        }
      }
    } catch (KeeperException ke) {
      watcher.abort("Unhandled zookeeper exception removing leader node", ke);
      candidate.stop("Unhandled zookeeper exception removing leader node: "
          + ke.getMessage());
    } catch (InterruptedException e) {
      watcher.abort("Unhandled zookeeper exception removing leader node", e);
      candidate.stop("Unhandled zookeeper exception removing leader node: "
          + e.getMessage());
    }
  }

  public boolean hasLeader() {
    return leaderExists.get();
  }
}