View Javadoc

1   /*
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.replication.regionserver;
20  
21  import java.io.IOException;
22  import java.io.InterruptedIOException;
23  import java.util.ArrayList;
24  import java.util.Collection;
25  import java.util.HashMap;
26  import java.util.List;
27  import java.util.Map;
28  import java.util.Map.Entry;
29  import java.util.TreeMap;
30  import java.util.UUID;
31  import java.util.concurrent.atomic.AtomicLong;
32  
33  import org.apache.commons.lang.StringUtils;
34  import org.apache.commons.logging.Log;
35  import org.apache.commons.logging.LogFactory;
36  import org.apache.hadoop.classification.InterfaceAudience;
37  import org.apache.hadoop.conf.Configuration;
38  import org.apache.hadoop.hbase.Cell;
39  import org.apache.hadoop.hbase.CellScanner;
40  import org.apache.hadoop.hbase.CellUtil;
41  import org.apache.hadoop.hbase.TableName;
42  import org.apache.hadoop.hbase.HBaseConfiguration;
43  import org.apache.hadoop.hbase.HConstants;
44  import org.apache.hadoop.hbase.KeyValueUtil;
45  import org.apache.hadoop.hbase.Stoppable;
46  import org.apache.hadoop.hbase.client.Delete;
47  import org.apache.hadoop.hbase.client.HConnection;
48  import org.apache.hadoop.hbase.client.HConnectionManager;
49  import org.apache.hadoop.hbase.client.Mutation;
50  import org.apache.hadoop.hbase.client.Put;
51  import org.apache.hadoop.hbase.client.Row;
52  import org.apache.hadoop.hbase.client.Table;
53  import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.WALEntry;
54  import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
55  
56  /**
57   * This class is responsible for replicating the edits coming
58   * from another cluster.
59   * <p/>
60   * This replication process is currently waiting for the edits to be applied
61   * before the method can return. This means that the replication of edits
62   * is synchronized (after reading from HLogs in ReplicationSource) and that a
63   * single region server cannot receive edits from two sources at the same time
64   * <p/>
65   * This class uses the native HBase client in order to replicate entries.
66   * <p/>
67   *
68   * TODO make this class more like ReplicationSource wrt log handling
69   */
70  @InterfaceAudience.Private
71  public class ReplicationSink {
72  
73    private static final Log LOG = LogFactory.getLog(ReplicationSink.class);
74    private final Configuration conf;
75    private final HConnection sharedHtableCon;
76    private final MetricsSink metrics;
77    private final AtomicLong totalReplicatedEdits = new AtomicLong();
78  
79    /**
80     * Create a sink for replication
81     *
82     * @param conf                conf object
83     * @param stopper             boolean to tell this thread to stop
84     * @throws IOException thrown when HDFS goes bad or bad file name
85     */
86    public ReplicationSink(Configuration conf, Stoppable stopper)
87        throws IOException {
88      this.conf = HBaseConfiguration.create(conf);
89      decorateConf();
90      this.metrics = new MetricsSink();
91      this.sharedHtableCon = HConnectionManager.createConnection(this.conf);
92    }
93  
94    /**
95     * decorate the Configuration object to make replication more receptive to delays:
96     * lessen the timeout and numTries.
97     */
98    private void decorateConf() {
99      this.conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
100         this.conf.getInt("replication.sink.client.retries.number", 4));
101     this.conf.setInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT,
102         this.conf.getInt("replication.sink.client.ops.timeout", 10000));
103     String replicationCodec = this.conf.get(HConstants.REPLICATION_CODEC_CONF_KEY);
104     if (StringUtils.isNotEmpty(replicationCodec)) {
105       this.conf.set(HConstants.RPC_CODEC_CONF_KEY, replicationCodec);
106     }
107    }
108 
109   /**
110    * Replicate this array of entries directly into the local cluster using the native client. Only
111    * operates against raw protobuf type saving on a conversion from pb to pojo.
112    * @param entries
113    * @param cells
114    * @throws IOException
115    */
116   public void replicateEntries(List<WALEntry> entries, final CellScanner cells) throws IOException {
117     if (entries.isEmpty()) return;
118     if (cells == null) throw new NullPointerException("TODO: Add handling of null CellScanner");
119     // Very simple optimization where we batch sequences of rows going
120     // to the same table.
121     try {
122       long totalReplicated = 0;
123       // Map of table => list of Rows, grouped by cluster id, we only want to flushCommits once per
124       // invocation of this method per table and cluster id.
125       Map<TableName, Map<List<UUID>, List<Row>>> rowMap =
126           new TreeMap<TableName, Map<List<UUID>, List<Row>>>();
127       for (WALEntry entry : entries) {
128         TableName table =
129             TableName.valueOf(entry.getKey().getTableName().toByteArray());
130         Cell previousCell = null;
131         Mutation m = null;
132         int count = entry.getAssociatedCellCount();
133         for (int i = 0; i < count; i++) {
134           // Throw index out of bounds if our cell count is off
135           if (!cells.advance()) {
136             throw new ArrayIndexOutOfBoundsException("Expected=" + count + ", index=" + i);
137           }
138           Cell cell = cells.current();
139           if (isNewRowOrType(previousCell, cell)) {
140             // Create new mutation
141             m = CellUtil.isDelete(cell)?
142               new Delete(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()):
143               new Put(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
144             List<UUID> clusterIds = new ArrayList<UUID>();
145             for(HBaseProtos.UUID clusterId : entry.getKey().getClusterIdsList()){
146               clusterIds.add(toUUID(clusterId));
147             }
148             m.setClusterIds(clusterIds);
149             addToHashMultiMap(rowMap, table, clusterIds, m);
150           }
151           if (CellUtil.isDelete(cell)) {
152             ((Delete)m).addDeleteMarker(KeyValueUtil.ensureKeyValue(cell));
153           } else {
154             ((Put)m).add(KeyValueUtil.ensureKeyValue(cell));
155           }
156           previousCell = cell;
157         }
158         totalReplicated++;
159       }
160       for (Entry<TableName, Map<List<UUID>,List<Row>>> entry : rowMap.entrySet()) {
161         batch(entry.getKey(), entry.getValue().values());
162       }
163       int size = entries.size();
164       this.metrics.setAgeOfLastAppliedOp(entries.get(size - 1).getKey().getWriteTime());
165       this.metrics.applyBatch(size);
166       this.totalReplicatedEdits.addAndGet(totalReplicated);
167     } catch (IOException ex) {
168       LOG.error("Unable to accept edit because:", ex);
169       throw ex;
170     }
171   }
172 
173   /**
174    * @param previousCell
175    * @param cell
176    * @return True if we have crossed over onto a new row or type
177    */
178   private boolean isNewRowOrType(final Cell previousCell, final Cell cell) {
179     return previousCell == null || previousCell.getTypeByte() != cell.getTypeByte() ||
180         !CellUtil.matchingRow(previousCell, cell);
181   }
182 
183   private java.util.UUID toUUID(final HBaseProtos.UUID uuid) {
184     return new java.util.UUID(uuid.getMostSigBits(), uuid.getLeastSigBits());
185   }
186 
187   /**
188    * Simple helper to a map from key to (a list of) values
189    * TODO: Make a general utility method
190    * @param map
191    * @param key1
192    * @param key2
193    * @param value
194    * @return the list of values corresponding to key1 and key2
195    */
196   private <K1, K2, V> List<V> addToHashMultiMap(Map<K1, Map<K2,List<V>>> map, K1 key1, K2 key2, V value) {
197     Map<K2,List<V>> innerMap = map.get(key1);
198     if (innerMap == null) {
199       innerMap = new HashMap<K2, List<V>>();
200       map.put(key1, innerMap);
201     }
202     List<V> values = innerMap.get(key2);
203     if (values == null) {
204       values = new ArrayList<V>();
205       innerMap.put(key2, values);
206     }
207     values.add(value);
208     return values;
209   }
210 
211   /**
212    * stop the thread pool executor. It is called when the regionserver is stopped.
213    */
214   public void stopReplicationSinkServices() {
215     try {
216       this.sharedHtableCon.close();
217     } catch (IOException e) {
218       LOG.warn("IOException while closing the connection", e); // ignoring as we are closing.
219     }
220   }
221 
222 
223   /**
224    * Do the changes and handle the pool
225    * @param tableName table to insert into
226    * @param allRows list of actions
227    * @throws IOException
228    */
229   protected void batch(TableName tableName, Collection<List<Row>> allRows) throws IOException {
230     if (allRows.isEmpty()) {
231       return;
232     }
233     Table table = null;
234     try {
235       table = this.sharedHtableCon.getTable(tableName);
236       for (List<Row> rows : allRows) {
237         table.batch(rows);
238       }
239     } catch (InterruptedException ix) {
240       throw (InterruptedIOException)new InterruptedIOException().initCause(ix);
241     } finally {
242       if (table != null) {
243         table.close();
244       }
245     }
246   }
247 
248   /**
249    * Get a string representation of this sink's metrics
250    * @return string with the total replicated edits count and the date
251    * of the last edit that was applied
252    */
253   public String getStats() {
254     return this.totalReplicatedEdits.get() == 0 ? "" : "Sink: " +
255       "age in ms of last applied edit: " + this.metrics.refreshAgeOfLastAppliedOp() +
256       ", total replicated edits: " + this.totalReplicatedEdits;
257   }
258 }