HBASE-26412 Handle sink failure in RegionReplicationSink (#3815)

[hbase.git] / hbase-server / src / main / java / org / apache / hadoop / hbase / regionserver / LeaseManager.java

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

import java.io.IOException;
import java.util.ConcurrentModificationException;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

import org.apache.hadoop.hbase.log.HBaseMarkers;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Leases
 *
 * There are several server classes in HBase that need to track external
 * clients that occasionally send heartbeats.
 *
 * <p>These external clients hold resources in the server class.
 * Those resources need to be released if the external client fails to send a
 * heartbeat after some interval of time passes.
 *
 * <p>The Leases class is a general reusable class for this kind of pattern.
 * An instance of the Leases class will create a thread to do its dirty work.
 * You should close() the instance if you want to clean up the thread properly.
 *
 * <p>
 * NOTE: This class extends Thread rather than Chore because the sleep time
 * can be interrupted when there is something to do, rather than the Chore
 * sleep time which is invariant.
 */
@InterfaceAudience.Private
public class LeaseManager extends Thread {
  private static final Logger LOG = LoggerFactory.getLogger(LeaseManager.class.getName());
  private static final int MIN_WAIT_TIME = 100;

  private final Map<String, Lease> leases = new ConcurrentHashMap<>();
  private final int leaseCheckFrequency;
  private volatile boolean stopRequested = false;

  /**
   * Creates a lease manager.
   *
   * @param leaseCheckFrequency - how often the lease should be checked (milliseconds)
   */
  public LeaseManager(final int leaseCheckFrequency) {
    super("RegionServer.LeaseManager");  // thread name
    this.leaseCheckFrequency = leaseCheckFrequency;
    setDaemon(true);
  }

  @Override
  public void run() {
    long toWait = leaseCheckFrequency;
    Lease nextLease = null;
    long nextLeaseDelay = Long.MAX_VALUE;

    while (!stopRequested || (stopRequested && !leases.isEmpty()) ) {

      try {
        if (nextLease != null) {
          toWait = nextLease.getDelay(TimeUnit.MILLISECONDS);
        }

        toWait = Math.min(leaseCheckFrequency, toWait);
        toWait = Math.max(MIN_WAIT_TIME, toWait);

        Thread.sleep(toWait);
      } catch (InterruptedException | ConcurrentModificationException e) {
        continue;
      } catch (Throwable e) {
        LOG.error(HBaseMarkers.FATAL, "Unexpected exception killed leases thread", e);
        break;
      }

      nextLease = null;
      nextLeaseDelay = Long.MAX_VALUE;
      for (Iterator<Map.Entry<String, Lease>> it = leases.entrySet().iterator(); it.hasNext();) {
        Map.Entry<String, Lease> entry = it.next();
        Lease lease = entry.getValue();
        long thisLeaseDelay = lease.getDelay(TimeUnit.MILLISECONDS);
        if ( thisLeaseDelay > 0) {
          if (nextLease == null || thisLeaseDelay < nextLeaseDelay) {
            nextLease = lease;
            nextLeaseDelay = thisLeaseDelay;
          }
        } else {
          // A lease expired.  Run the expired code before removing from map
          // since its presence in map is used to see if lease exists still.
          if (lease.getListener() == null) {
            LOG.error("lease listener is null for lease " + lease.getLeaseName());
          } else {
            lease.getListener().leaseExpired();
          }
          it.remove();
        }
      }
    }
    close();
  }

  /**
   * Shuts down this lease instance when all outstanding leases expire.
   * Like {@link #close()} but rather than violently end all leases, waits
   * first on extant leases to finish.  Use this method if the lease holders
   * could lose data, leak locks, etc.  Presumes client has shutdown
   * allocation of new leases.
   */
  public void closeAfterLeasesExpire() {
    this.stopRequested = true;
  }

  /**
   * Shut down this Leases instance.  All pending leases will be destroyed,
   * without any cancellation calls.
   */
  public void close() {
    this.stopRequested = true;
    leases.clear();
    LOG.info("Closed leases");
  }

  /**
   * Create a lease and insert it to the map of leases.
   *
   * @param leaseName name of the lease
   * @param leaseTimeoutPeriod length of the lease in milliseconds
   * @param listener listener that will process lease expirations
   * @return The lease created.
   */
  public Lease createLease(String leaseName, int leaseTimeoutPeriod, final LeaseListener listener)
      throws LeaseStillHeldException {
    Lease lease = new Lease(leaseName, leaseTimeoutPeriod, listener);
    addLease(lease);
    return lease;
  }

  /**
   * Inserts lease.  Resets expiration before insertion.
   */
  public void addLease(final Lease lease) throws LeaseStillHeldException {
    if (this.stopRequested) {
      return;
    }
    if (leases.containsKey(lease.getLeaseName())) {
      throw new LeaseStillHeldException(lease.getLeaseName());
    }
    lease.resetExpirationTime();
    leases.put(lease.getLeaseName(), lease);
  }

  /**
   * Renew a lease
   *
   * @param leaseName name of the lease
   */
  public void renewLease(final String leaseName) throws LeaseException {
    if (this.stopRequested) {
      return;
    }
    Lease lease = leases.get(leaseName);

    if (lease == null ) {
      throw new LeaseException("lease '" + leaseName +
          "' does not exist or has already expired");
    }
    lease.resetExpirationTime();
  }

  /**
   * Client explicitly cancels a lease.
   *
   * @param leaseName name of lease
   */
  public void cancelLease(final String leaseName) throws LeaseException {
    removeLease(leaseName);
  }

  /**
   * Remove named lease. Lease is removed from the map of leases.
   *
   * @param leaseName name of lease
   * @return Removed lease
   */
  Lease removeLease(final String leaseName) throws LeaseException {
    Lease lease = leases.remove(leaseName);
    if (lease == null) {
      throw new LeaseException("lease '" + leaseName + "' does not exist");
    }
    return lease;
  }

  /**
   * Thrown if we are asked to create a lease but lease on passed name already
   * exists.
   */
  @SuppressWarnings("serial")
  public static class LeaseStillHeldException extends IOException {
    private final String leaseName;

    public LeaseStillHeldException(final String name) {
      this.leaseName = name;
    }

    /** @return name of lease */
    public String getName() {
      return this.leaseName;
    }
  }

  /** This class tracks a single Lease. */
  static class Lease implements Delayed {
    private final String leaseName;
    private final LeaseListener listener;
    private int leaseTimeoutPeriod;
    private long expirationTime;

    Lease(final String leaseName, int leaseTimeoutPeriod, LeaseListener listener) {
      this.leaseName = leaseName;
      this.listener = listener;
      this.leaseTimeoutPeriod = leaseTimeoutPeriod;
      this.expirationTime = 0;
    }

    /** @return the lease name */
    public String getLeaseName() {
      return leaseName;
    }

    /** @return listener */
    public LeaseListener getListener() {
      return this.listener;
    }

    @Override
    public boolean equals(Object obj) {
      if (this == obj) {
        return true;
      }
      if (obj == null) {
        return false;
      }
      if (getClass() != obj.getClass()) {
        return false;
      }
      return this.hashCode() == obj.hashCode();
    }

    @Override
    public int hashCode() {
      return this.leaseName.hashCode();
    }

    @Override
    public long getDelay(TimeUnit unit) {
      return unit.convert(this.expirationTime - EnvironmentEdgeManager.currentTime(),
          TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed o) {
      long delta = this.getDelay(TimeUnit.MILLISECONDS) -
        o.getDelay(TimeUnit.MILLISECONDS);

      return this.equals(o) ? 0 : (delta > 0 ? 1 : -1);
    }

    /**
     * Resets the expiration time of the lease.
     */
    public void resetExpirationTime() {
      this.expirationTime = EnvironmentEdgeManager.currentTime() + this.leaseTimeoutPeriod;
    }
  }
}