View Javadoc

1   /**
2    * Copyright The Apache Software Foundation
3    *
4    * Licensed to the Apache Software Foundation (ASF) under one
5    * or more contributor license agreements.  See the NOTICE file
6    * distributed with this work for additional information
7    * regarding copyright ownership.  The ASF licenses this file
8    * to you under the Apache License, Version 2.0 (the
9    * "License"); you may not use this file except in compliance
10   * with the License.  You may obtain a copy of the License at
11   *
12   *     http://www.apache.org/licenses/LICENSE-2.0
13   *
14   * Unless required by applicable law or agreed to in writing, software
15   * distributed under the License is distributed on an "AS IS" BASIS,
16   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17   * See the License for the specific language governing permissions and
18   * limitations under the License.
19   */
20  package org.apache.hadoop.hbase.client;
21  
22  import com.google.common.annotations.VisibleForTesting;
23  import com.google.common.util.concurrent.ThreadFactoryBuilder;
24  
25  import java.io.IOException;
26  import java.util.AbstractMap.SimpleEntry;
27  import java.util.ArrayList;
28  import java.util.Collections;
29  import java.util.HashMap;
30  import java.util.List;
31  import java.util.Map;
32  import java.util.concurrent.ConcurrentHashMap;
33  import java.util.concurrent.ExecutorService;
34  import java.util.concurrent.Executors;
35  import java.util.concurrent.LinkedBlockingQueue;
36  import java.util.concurrent.ScheduledExecutorService;
37  import java.util.concurrent.TimeUnit;
38  import java.util.concurrent.atomic.AtomicInteger;
39  import java.util.concurrent.atomic.AtomicLong;
40  
41  import org.apache.commons.logging.Log;
42  import org.apache.commons.logging.LogFactory;
43  import org.apache.hadoop.conf.Configuration;
44  import org.apache.hadoop.hbase.HBaseConfiguration;
45  import org.apache.hadoop.hbase.HConstants;
46  import org.apache.hadoop.hbase.HRegionInfo;
47  import org.apache.hadoop.hbase.HRegionLocation;
48  import org.apache.hadoop.hbase.ServerName;
49  import org.apache.hadoop.hbase.TableName;
50  import org.apache.hadoop.hbase.classification.InterfaceAudience;
51  import org.apache.hadoop.hbase.classification.InterfaceStability;
52  import org.apache.hadoop.hbase.client.AsyncProcess.AsyncRequestFuture;
53  import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
54  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
55  
56  /**
57   * HTableMultiplexer provides a thread-safe non blocking PUT API across all the tables.
58   * Each put will be sharded into different buffer queues based on its destination region server.
59   * So each region server buffer queue will only have the puts which share the same destination.
60   * And each queue will have a flush worker thread to flush the puts request to the region server.
61   * If any queue is full, the HTableMultiplexer starts to drop the Put requests for that
62   * particular queue.
63   *
64   * Also all the puts will be retried as a configuration number before dropping.
65   * And the HTableMultiplexer can report the number of buffered requests and the number of the
66   * failed (dropped) requests in total or on per region server basis.
67   *
68   * This class is thread safe.
69   */
70  @InterfaceAudience.Public
71  @InterfaceStability.Evolving
72  public class HTableMultiplexer {
73    private static final Log LOG = LogFactory.getLog(HTableMultiplexer.class.getName());
74  
75    public static final String TABLE_MULTIPLEXER_FLUSH_PERIOD_MS =
76        "hbase.tablemultiplexer.flush.period.ms";
77    public static final String TABLE_MULTIPLEXER_INIT_THREADS = "hbase.tablemultiplexer.init.threads";
78    public static final String TABLE_MULTIPLEXER_MAX_RETRIES_IN_QUEUE =
79        "hbase.client.max.retries.in.queue";
80  
81    /** The map between each region server to its flush worker */
82    private final Map<HRegionLocation, FlushWorker> serverToFlushWorkerMap =
83        new ConcurrentHashMap<>();
84  
85    private final Configuration workerConf;
86    private final ClusterConnection conn;
87    private final ExecutorService pool;
88    private final int retryNum;
89    private final int perRegionServerBufferQueueSize;
90    private final int maxKeyValueSize;
91    private final ScheduledExecutorService executor;
92    private final long flushPeriod;
93  
94    /**
95     * @param conf The HBaseConfiguration
96     * @param perRegionServerBufferQueueSize determines the max number of the buffered Put ops for
97     *          each region server before dropping the request.
98     */
99    public HTableMultiplexer(Configuration conf, int perRegionServerBufferQueueSize)
100       throws IOException {
101     this(ConnectionFactory.createConnection(conf), conf, perRegionServerBufferQueueSize);
102   }
103 
104   /**
105    * @param conn The HBase connection.
106    * @param conf The HBase configuration
107    * @param perRegionServerBufferQueueSize determines the max number of the buffered Put ops for
108    *          each region server before dropping the request.
109    */
110   public HTableMultiplexer(Connection conn, Configuration conf,
111       int perRegionServerBufferQueueSize) {
112     this.conn = (ClusterConnection) conn;
113     this.pool = HTable.getDefaultExecutor(conf);
114     this.retryNum = conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
115         HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
116     this.perRegionServerBufferQueueSize = perRegionServerBufferQueueSize;
117     this.maxKeyValueSize = HTable.getMaxKeyValueSize(conf);
118     this.flushPeriod = conf.getLong(TABLE_MULTIPLEXER_FLUSH_PERIOD_MS, 100);
119     int initThreads = conf.getInt(TABLE_MULTIPLEXER_INIT_THREADS, 10);
120     this.executor =
121         Executors.newScheduledThreadPool(initThreads,
122           new ThreadFactoryBuilder().setDaemon(true).setNameFormat("HTableFlushWorker-%d").build());
123 
124     this.workerConf = HBaseConfiguration.create(conf);
125     // We do not do the retry because we need to reassign puts to different queues if regions are
126     // moved.
127     this.workerConf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 0);
128   }
129 
130   /**
131    * Closes the internal {@link Connection}. Does nothing if the {@link Connection} has already
132    * been closed.
133    * @throws IOException If there is an error closing the connection.
134    */
135   @SuppressWarnings("deprecation")
136   public synchronized void close() throws IOException {
137     if (!getConnection().isClosed()) {
138       getConnection().close();
139     }
140   }
141 
142   /**
143    * The put request will be buffered by its corresponding buffer queue. Return false if the queue
144    * is already full.
145    * @param tableName
146    * @param put
147    * @return true if the request can be accepted by its corresponding buffer queue.
148    * @throws IOException
149    */
150   public boolean put(TableName tableName, final Put put) {
151     return put(tableName, put, this.retryNum);
152   }
153 
154   /**
155    * The puts request will be buffered by their corresponding buffer queue.
156    * Return the list of puts which could not be queued.
157    * @param tableName
158    * @param puts
159    * @return the list of puts which could not be queued
160    * @throws IOException
161    */
162   public List<Put> put(TableName tableName, final List<Put> puts) {
163     if (puts == null)
164       return null;
165 
166     List <Put> failedPuts = null;
167     boolean result;
168     for (Put put : puts) {
169       result = put(tableName, put, this.retryNum);
170       if (result == false) {
171 
172         // Create the failed puts list if necessary
173         if (failedPuts == null) {
174           failedPuts = new ArrayList<Put>();
175         }
176         // Add the put to the failed puts list
177         failedPuts.add(put);
178       }
179     }
180     return failedPuts;
181   }
182 
183   /**
184    * Deprecated. Use {@link #put(TableName, List) } instead.
185    */
186   @Deprecated
187   public List<Put> put(byte[] tableName, final List<Put> puts) {
188     return put(TableName.valueOf(tableName), puts);
189   }
190 
191   /**
192    * The put request will be buffered by its corresponding buffer queue. And the put request will be
193    * retried before dropping the request.
194    * Return false if the queue is already full.
195    * @return true if the request can be accepted by its corresponding buffer queue.
196    * @throws IOException
197    */
198   public boolean put(final TableName tableName, final Put put, int retry) {
199     if (retry <= 0) {
200       return false;
201     }
202 
203     try {
204       HTable.validatePut(put, maxKeyValueSize);
205       // Allow mocking to get at the connection, but don't expose the connection to users.
206       ClusterConnection conn = (ClusterConnection) getConnection();
207       // AsyncProcess in the FlushWorker should take care of refreshing the location cache
208       // as necessary. We shouldn't have to do that here.
209       HRegionLocation loc = conn.getRegionLocation(tableName, put.getRow(), false);
210       if (loc != null) {
211         // Add the put pair into its corresponding queue.
212         LinkedBlockingQueue<PutStatus> queue = getQueue(loc);
213 
214         // Generate a MultiPutStatus object and offer it into the queue
215         PutStatus s = new PutStatus(loc.getRegionInfo(), put, retry);
216 
217         return queue.offer(s);
218       }
219     } catch (IOException e) {
220       LOG.debug("Cannot process the put " + put, e);
221     }
222     return false;
223   }
224 
225   /**
226    * Deprecated. Use {@link #put(TableName, Put) } instead.
227    */
228   @Deprecated
229   public boolean put(final byte[] tableName, final Put put, int retry) {
230     return put(TableName.valueOf(tableName), put, retry);
231   }
232 
233   /**
234    * Deprecated. Use {@link #put(TableName, Put)} instead.
235    */
236   @Deprecated
237   public boolean put(final byte[] tableName, Put put) {
238     return put(TableName.valueOf(tableName), put);
239   }
240 
241   /**
242    * @return the current HTableMultiplexerStatus
243    */
244   public HTableMultiplexerStatus getHTableMultiplexerStatus() {
245     return new HTableMultiplexerStatus(serverToFlushWorkerMap);
246   }
247 
248   @VisibleForTesting
249   LinkedBlockingQueue<PutStatus> getQueue(HRegionLocation addr) {
250     FlushWorker worker = serverToFlushWorkerMap.get(addr);
251     if (worker == null) {
252       synchronized (this.serverToFlushWorkerMap) {
253         worker = serverToFlushWorkerMap.get(addr);
254         if (worker == null) {
255           // Create the flush worker
256           worker = new FlushWorker(workerConf, this.conn, addr, this, perRegionServerBufferQueueSize,
257                   pool, executor);
258           this.serverToFlushWorkerMap.put(addr, worker);
259           executor.scheduleAtFixedRate(worker, flushPeriod, flushPeriod, TimeUnit.MILLISECONDS);
260         }
261       }
262     }
263     return worker.getQueue();
264   }
265 
266   @VisibleForTesting
267   ClusterConnection getConnection() {
268     return this.conn;
269   }
270 
271   /**
272    * HTableMultiplexerStatus keeps track of the current status of the HTableMultiplexer.
273    * report the number of buffered requests and the number of the failed (dropped) requests
274    * in total or on per region server basis.
275    */
276   @InterfaceAudience.Public
277   @InterfaceStability.Evolving
278   public static class HTableMultiplexerStatus {
279     private long totalFailedPutCounter;
280     private long totalBufferedPutCounter;
281     private long maxLatency;
282     private long overallAverageLatency;
283     private Map<String, Long> serverToFailedCounterMap;
284     private Map<String, Long> serverToBufferedCounterMap;
285     private Map<String, Long> serverToAverageLatencyMap;
286     private Map<String, Long> serverToMaxLatencyMap;
287 
288     public HTableMultiplexerStatus(
289         Map<HRegionLocation, FlushWorker> serverToFlushWorkerMap) {
290       this.totalBufferedPutCounter = 0;
291       this.totalFailedPutCounter = 0;
292       this.maxLatency = 0;
293       this.overallAverageLatency = 0;
294       this.serverToBufferedCounterMap = new HashMap<String, Long>();
295       this.serverToFailedCounterMap = new HashMap<String, Long>();
296       this.serverToAverageLatencyMap = new HashMap<String, Long>();
297       this.serverToMaxLatencyMap = new HashMap<String, Long>();
298       this.initialize(serverToFlushWorkerMap);
299     }
300 
301     private void initialize(
302         Map<HRegionLocation, FlushWorker> serverToFlushWorkerMap) {
303       if (serverToFlushWorkerMap == null) {
304         return;
305       }
306 
307       long averageCalcSum = 0;
308       int averageCalcCount = 0;
309       for (Map.Entry<HRegionLocation, FlushWorker> entry : serverToFlushWorkerMap
310           .entrySet()) {
311         HRegionLocation addr = entry.getKey();
312         FlushWorker worker = entry.getValue();
313 
314         long bufferedCounter = worker.getTotalBufferedCount();
315         long failedCounter = worker.getTotalFailedCount();
316         long serverMaxLatency = worker.getMaxLatency();
317         AtomicAverageCounter averageCounter = worker.getAverageLatencyCounter();
318         // Get sum and count pieces separately to compute overall average
319         SimpleEntry<Long, Integer> averageComponents = averageCounter
320             .getComponents();
321         long serverAvgLatency = averageCounter.getAndReset();
322 
323         this.totalBufferedPutCounter += bufferedCounter;
324         this.totalFailedPutCounter += failedCounter;
325         if (serverMaxLatency > this.maxLatency) {
326           this.maxLatency = serverMaxLatency;
327         }
328         averageCalcSum += averageComponents.getKey();
329         averageCalcCount += averageComponents.getValue();
330 
331         this.serverToBufferedCounterMap.put(addr.getHostnamePort(),
332             bufferedCounter);
333         this.serverToFailedCounterMap
334             .put(addr.getHostnamePort(),
335             failedCounter);
336         this.serverToAverageLatencyMap.put(addr.getHostnamePort(),
337             serverAvgLatency);
338         this.serverToMaxLatencyMap
339             .put(addr.getHostnamePort(),
340             serverMaxLatency);
341       }
342       this.overallAverageLatency = averageCalcCount != 0 ? averageCalcSum
343           / averageCalcCount : 0;
344     }
345 
346     public long getTotalBufferedCounter() {
347       return this.totalBufferedPutCounter;
348     }
349 
350     public long getTotalFailedCounter() {
351       return this.totalFailedPutCounter;
352     }
353 
354     public long getMaxLatency() {
355       return this.maxLatency;
356     }
357 
358     public long getOverallAverageLatency() {
359       return this.overallAverageLatency;
360     }
361 
362     public Map<String, Long> getBufferedCounterForEachRegionServer() {
363       return this.serverToBufferedCounterMap;
364     }
365 
366     public Map<String, Long> getFailedCounterForEachRegionServer() {
367       return this.serverToFailedCounterMap;
368     }
369 
370     public Map<String, Long> getMaxLatencyForEachRegionServer() {
371       return this.serverToMaxLatencyMap;
372     }
373 
374     public Map<String, Long> getAverageLatencyForEachRegionServer() {
375       return this.serverToAverageLatencyMap;
376     }
377   }
378 
379   @VisibleForTesting
380   static class PutStatus {
381     public final HRegionInfo regionInfo;
382     public final Put put;
383     public final int retryCount;
384 
385     public PutStatus(HRegionInfo regionInfo, Put put, int retryCount) {
386       this.regionInfo = regionInfo;
387       this.put = put;
388       this.retryCount = retryCount;
389     }
390   }
391 
392   /**
393    * Helper to count the average over an interval until reset.
394    */
395   private static class AtomicAverageCounter {
396     private long sum;
397     private int count;
398 
399     public AtomicAverageCounter() {
400       this.sum = 0L;
401       this.count = 0;
402     }
403 
404     public synchronized long getAndReset() {
405       long result = this.get();
406       this.reset();
407       return result;
408     }
409 
410     public synchronized long get() {
411       if (this.count == 0) {
412         return 0;
413       }
414       return this.sum / this.count;
415     }
416 
417     public synchronized SimpleEntry<Long, Integer> getComponents() {
418       return new SimpleEntry<Long, Integer>(sum, count);
419     }
420 
421     public synchronized void reset() {
422       this.sum = 0l;
423       this.count = 0;
424     }
425 
426     public synchronized void add(long value) {
427       this.sum += value;
428       this.count++;
429     }
430   }
431 
432   @VisibleForTesting
433   static class FlushWorker implements Runnable {
434     private final HRegionLocation addr;
435     private final LinkedBlockingQueue<PutStatus> queue;
436     private final HTableMultiplexer multiplexer;
437     private final AtomicLong totalFailedPutCount = new AtomicLong(0);
438     private final AtomicInteger currentProcessingCount = new AtomicInteger(0);
439     private final AtomicAverageCounter averageLatency = new AtomicAverageCounter();
440     private final AtomicLong maxLatency = new AtomicLong(0);
441 
442     private final AsyncProcess ap;
443     private final List<PutStatus> processingList = new ArrayList<>();
444     private final ScheduledExecutorService executor;
445     private final int maxRetryInQueue;
446     private final AtomicInteger retryInQueue = new AtomicInteger(0);
447 
448     public FlushWorker(Configuration conf, ClusterConnection conn, HRegionLocation addr,
449         HTableMultiplexer htableMultiplexer, int perRegionServerBufferQueueSize,
450         ExecutorService pool, ScheduledExecutorService executor) {
451       this.addr = addr;
452       this.multiplexer = htableMultiplexer;
453       this.queue = new LinkedBlockingQueue<>(perRegionServerBufferQueueSize);
454       RpcRetryingCallerFactory rpcCallerFactory = RpcRetryingCallerFactory.instantiate(conf);
455       RpcControllerFactory rpcControllerFactory = RpcControllerFactory.instantiate(conf);
456       this.ap = new AsyncProcess(conn, conf, pool, rpcCallerFactory, false, rpcControllerFactory);
457       this.executor = executor;
458       this.maxRetryInQueue = conf.getInt(TABLE_MULTIPLEXER_MAX_RETRIES_IN_QUEUE, 10000);
459     }
460 
461     protected LinkedBlockingQueue<PutStatus> getQueue() {
462       return this.queue;
463     }
464 
465     public long getTotalFailedCount() {
466       return totalFailedPutCount.get();
467     }
468 
469     public long getTotalBufferedCount() {
470       return queue.size() + currentProcessingCount.get();
471     }
472 
473     public AtomicAverageCounter getAverageLatencyCounter() {
474       return this.averageLatency;
475     }
476 
477     public long getMaxLatency() {
478       return this.maxLatency.getAndSet(0);
479     }
480 
481     boolean resubmitFailedPut(PutStatus ps, HRegionLocation oldLoc) throws IOException {
482       // Decrease the retry count
483       final int retryCount = ps.retryCount - 1;
484 
485       if (retryCount <= 0) {
486         // Update the failed counter and no retry any more.
487         return false;
488       }
489 
490       int cnt = getRetryInQueue().incrementAndGet();
491       if (cnt > getMaxRetryInQueue()) {
492         // Too many Puts in queue for resubmit, give up this
493         getRetryInQueue().decrementAndGet();
494         return false;
495       }
496 
497       final Put failedPut = ps.put;
498       // The currentPut is failed. So get the table name for the currentPut.
499       final TableName tableName = ps.regionInfo.getTable();
500 
501       long delayMs = getNextDelay(retryCount);
502       if (LOG.isDebugEnabled()) {
503         LOG.debug("resubmitting after " + delayMs + "ms: " + retryCount);
504       }
505 
506       // HBASE-12198, HBASE-15221, HBASE-15232: AsyncProcess should be responsible for updating
507       // the region location cache when the Put original failed with some exception. If we keep
508       // re-trying the same Put to the same location, AsyncProcess isn't doing the right stuff
509       // that we expect it to.
510       getExecutor().schedule(new Runnable() {
511         @Override
512         public void run() {
513           boolean succ = false;
514           try {
515             succ = FlushWorker.this.getMultiplexer().put(tableName, failedPut, retryCount);
516           } finally {
517             FlushWorker.this.getRetryInQueue().decrementAndGet();
518             if (!succ) {
519               FlushWorker.this.getTotalFailedPutCount().incrementAndGet();
520             }
521           }
522         }
523       }, delayMs, TimeUnit.MILLISECONDS);
524       return true;
525     }
526 
527     @VisibleForTesting
528     long getNextDelay(int retryCount) {
529       return ConnectionUtils.getPauseTime(multiplexer.flushPeriod,
530           multiplexer.retryNum - retryCount - 1);
531     }
532 
533     @VisibleForTesting
534     AtomicInteger getRetryInQueue() {
535       return this.retryInQueue;
536     }
537 
538     @VisibleForTesting
539     int getMaxRetryInQueue() {
540       return this.maxRetryInQueue;
541     }
542 
543     @VisibleForTesting
544     AtomicLong getTotalFailedPutCount() {
545       return this.totalFailedPutCount;
546     }
547 
548     @VisibleForTesting
549     HTableMultiplexer getMultiplexer() {
550       return this.multiplexer;
551     }
552 
553     @VisibleForTesting
554     ScheduledExecutorService getExecutor() {
555       return this.executor;
556     }
557 
558     @Override
559     public void run() {
560       int failedCount = 0;
561       try {
562         long start = EnvironmentEdgeManager.currentTime();
563 
564         // drain all the queued puts into the tmp list
565         processingList.clear();
566         queue.drainTo(processingList);
567         if (processingList.size() == 0) {
568           // Nothing to flush
569           return;
570         }
571 
572         currentProcessingCount.set(processingList.size());
573         // failedCount is decreased whenever a Put is success or resubmit.
574         failedCount = processingList.size();
575 
576         List<Action<Row>> retainedActions = new ArrayList<>(processingList.size());
577         MultiAction<Row> actions = new MultiAction<>();
578         for (int i = 0; i < processingList.size(); i++) {
579           PutStatus putStatus = processingList.get(i);
580           Action<Row> action = new Action<Row>(putStatus.put, i);
581           actions.add(putStatus.regionInfo.getRegionName(), action);
582           retainedActions.add(action);
583         }
584 
585         // Process this multi-put request
586         List<PutStatus> failed = null;
587         Object[] results = new Object[actions.size()];
588         ServerName server = addr.getServerName();
589         Map<ServerName, MultiAction<Row>> actionsByServer =
590             Collections.singletonMap(server, actions);
591         try {
592           AsyncRequestFuture arf =
593               ap.submitMultiActions(null, retainedActions, 0L, null, results, true, null,
594                 null, actionsByServer, null);
595           arf.waitUntilDone();
596           if (arf.hasError()) {
597             // We just log and ignore the exception here since failed Puts will be resubmit again.
598             LOG.debug("Caught some exceptions when flushing puts to region server "
599                 + addr.getHostnamePort(), arf.getErrors());
600           }
601         } finally {
602           for (int i = 0; i < results.length; i++) {
603             if (results[i] instanceof Result) {
604               failedCount--;
605             } else {
606               if (failed == null) {
607                 failed = new ArrayList<PutStatus>();
608               }
609               failed.add(processingList.get(i));
610             }
611           }
612         }
613 
614         if (failed != null) {
615           // Resubmit failed puts
616           for (PutStatus putStatus : failed) {
617             if (resubmitFailedPut(putStatus, this.addr)) {
618               failedCount--;
619             }
620           }
621         }
622 
623         long elapsed = EnvironmentEdgeManager.currentTime() - start;
624         // Update latency counters
625         averageLatency.add(elapsed);
626         if (elapsed > maxLatency.get()) {
627           maxLatency.set(elapsed);
628         }
629 
630         // Log some basic info
631         if (LOG.isDebugEnabled()) {
632           LOG.debug("Processed " + currentProcessingCount + " put requests for "
633               + addr.getHostnamePort() + " and " + failedCount + " failed"
634               + ", latency for this send: " + elapsed);
635         }
636 
637         // Reset the current processing put count
638         currentProcessingCount.set(0);
639       } catch (RuntimeException e) {
640         // To make findbugs happy
641         // Log all the exceptions and move on
642         LOG.debug(
643           "Caught some exceptions " + e + " when flushing puts to region server "
644               + addr.getHostnamePort(), e);
645       } catch (Exception e) {
646         if (e instanceof InterruptedException) {
647           Thread.currentThread().interrupt();
648         }
649         // Log all the exceptions and move on
650         LOG.debug(
651           "Caught some exceptions " + e + " when flushing puts to region server "
652               + addr.getHostnamePort(), e);
653       } finally {
654         // Update the totalFailedCount
655         this.totalFailedPutCount.addAndGet(failedCount);
656       }
657     }
658   }
659 }