001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.replication.regionserver;
019
020import java.util.Map;
021import java.util.Queue;
022import java.util.concurrent.ConcurrentHashMap;
023import java.util.concurrent.PriorityBlockingQueue;
024import org.apache.hadoop.conf.Configuration;
025import org.apache.hadoop.fs.Path;
026import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
027import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
028import org.apache.yetus.audience.InterfaceAudience;
029import org.apache.yetus.audience.InterfaceStability;
030import org.slf4j.Logger;
031import org.slf4j.LoggerFactory;
032
033/*
034  Class that does enqueueing/dequeuing of wal at one place so that we can update the metrics
035  just at one place.
036 */
037@InterfaceAudience.Private
038@InterfaceStability.Evolving
039public class ReplicationSourceLogQueue {
040  private static final Logger LOG = LoggerFactory.getLogger(ReplicationSource.class);
041  // Queues of logs to process, entry in format of walGroupId->queue,
042  // each presents a queue for one wal group
043  private Map<String, PriorityBlockingQueue<Path>> queues = new ConcurrentHashMap<>();
044  private MetricsSource metrics;
045  private Configuration conf;
046  // per group queue size, keep no more than this number of logs in each wal group
047  private int queueSizePerGroup;
048  // WARN threshold for the number of queued logs, defaults to 2
049  private int logQueueWarnThreshold;
050  private ReplicationSource source;
051
052  public ReplicationSourceLogQueue(Configuration conf, MetricsSource metrics,
053    ReplicationSource source) {
054    this.conf = conf;
055    this.metrics = metrics;
056    this.source = source;
057    this.queueSizePerGroup = this.conf.getInt("hbase.regionserver.maxlogs", 32);
058    this.logQueueWarnThreshold = this.conf.getInt("replication.source.log.queue.warn", 2);
059  }
060
061  /**
062   * Enqueue the wal
063   * @param wal        wal to be enqueued
064   * @param walGroupId Key for the wal in @queues map
065   * @return boolean whether this is the first time we are seeing this walGroupId.
066   */
067  public boolean enqueueLog(Path wal, String walGroupId) {
068    boolean exists = false;
069    PriorityBlockingQueue<Path> queue = queues.get(walGroupId);
070    if (queue == null) {
071      queue = new PriorityBlockingQueue<>(queueSizePerGroup,
072        new AbstractFSWALProvider.WALStartTimeComparator());
073      // make sure that we do not use an empty queue when setting up a ReplicationSource, otherwise
074      // the shipper may quit immediately
075      queue.put(wal);
076      queues.put(walGroupId, queue);
077    } else {
078      exists = true;
079      queue.put(wal);
080    }
081    // Increment size of logQueue
082    this.metrics.incrSizeOfLogQueue();
083    // Compute oldest wal age
084    this.metrics.setOldestWalAge(getOldestWalAge());
085    // This will wal a warning for each new wal that gets created above the warn threshold
086    int queueSize = queue.size();
087    if (queueSize > this.logQueueWarnThreshold) {
088      LOG.warn(
089        "{} WAL group {} queue size: {} exceeds value of " + "replication.source.log.queue.warn {}",
090        source.logPeerId(), walGroupId, queueSize, logQueueWarnThreshold);
091    }
092    return exists;
093  }
094
095  /**
096   * Get the queue size for the given walGroupId.
097   * @param walGroupId walGroupId
098   */
099  public int getQueueSize(String walGroupId) {
100    Queue queue = queues.get(walGroupId);
101    if (queue == null) {
102      return 0;
103    }
104    return queue.size();
105  }
106
107  /**
108   * Returns number of queues.
109   */
110  public int getNumQueues() {
111    return queues.size();
112  }
113
114  public Map<String, PriorityBlockingQueue<Path>> getQueues() {
115    return queues;
116  }
117
118  /**
119   * Return queue for the given walGroupId Please don't add or remove elements from the returned
120   * queue. Use @enqueueLog and @remove methods respectively.
121   * @param walGroupId walGroupId
122   */
123  public PriorityBlockingQueue<Path> getQueue(String walGroupId) {
124    return queues.get(walGroupId);
125  }
126
127  /**
128   * Remove head from the queue corresponding to given walGroupId.
129   * @param walGroupId walGroupId
130   */
131  public void remove(String walGroupId) {
132    PriorityBlockingQueue<Path> queue = getQueue(walGroupId);
133    if (queue == null || queue.isEmpty()) {
134      return;
135    }
136    queue.remove();
137    // Decrease size logQueue.
138    this.metrics.decrSizeOfLogQueue();
139    // Re-compute age of oldest wal metric.
140    this.metrics.setOldestWalAge(getOldestWalAge());
141  }
142
143  /**
144   * Remove all the elements from the queue corresponding to walGroupId
145   * @param walGroupId walGroupId
146   */
147  public void clear(String walGroupId) {
148    PriorityBlockingQueue<Path> queue = getQueue(walGroupId);
149    while (!queue.isEmpty()) {
150      // Need to iterate since metrics#decrSizeOfLogQueue decrements just by 1.
151      queue.remove();
152      metrics.decrSizeOfLogQueue();
153    }
154    this.metrics.setOldestWalAge(getOldestWalAge());
155  }
156
157  /*
158   * Returns the age of oldest wal.
159   */
160  long getOldestWalAge() {
161    long now = EnvironmentEdgeManager.currentTime();
162    long timestamp = getOldestWalTimestamp();
163    if (timestamp == Long.MAX_VALUE) {
164      // If there are no wals in the queue then set the oldest wal timestamp to current time
165      // so that the oldest wal age will be 0.
166      timestamp = now;
167    }
168    long age = now - timestamp;
169    return age;
170  }
171
172  /*
173   * Get the oldest wal timestamp from all the queues.
174   */
175  private long getOldestWalTimestamp() {
176    long oldestWalTimestamp = Long.MAX_VALUE;
177    for (Map.Entry<String, PriorityBlockingQueue<Path>> entry : queues.entrySet()) {
178      PriorityBlockingQueue<Path> queue = entry.getValue();
179      Path path = queue.peek();
180      // Can path ever be null ?
181      if (path != null) {
182        oldestWalTimestamp =
183          Math.min(oldestWalTimestamp, AbstractFSWALProvider.WALStartTimeComparator.getTS(path));
184      }
185    }
186    return oldestWalTimestamp;
187  }
188
189  public MetricsSource getMetrics() {
190    return metrics;
191  }
192}