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.client.replication;
20  
21  import com.google.common.annotations.VisibleForTesting;
22  import com.google.common.collect.Lists;
23  
24  import java.io.Closeable;
25  import java.io.IOException;
26  import java.util.ArrayList;
27  import java.util.Collection;
28  import java.util.HashMap;
29  import java.util.HashSet;
30  import java.util.List;
31  import java.util.Map;
32  import java.util.Map.Entry;
33  import java.util.Set;
34  
35  import org.apache.commons.lang.StringUtils;
36  import org.apache.commons.logging.Log;
37  import org.apache.commons.logging.LogFactory;
38  import org.apache.hadoop.conf.Configuration;
39  import org.apache.hadoop.hbase.Abortable;
40  import org.apache.hadoop.hbase.HColumnDescriptor;
41  import org.apache.hadoop.hbase.HConstants;
42  import org.apache.hadoop.hbase.HTableDescriptor;
43  import org.apache.hadoop.hbase.TableName;
44  import org.apache.hadoop.hbase.TableNotFoundException;
45  import org.apache.hadoop.hbase.classification.InterfaceAudience;
46  import org.apache.hadoop.hbase.classification.InterfaceStability;
47  import org.apache.hadoop.hbase.client.Admin;
48  import org.apache.hadoop.hbase.client.Connection;
49  import org.apache.hadoop.hbase.client.ConnectionFactory;
50  import org.apache.hadoop.hbase.client.RegionLocator;
51  import org.apache.hadoop.hbase.replication.ReplicationException;
52  import org.apache.hadoop.hbase.replication.ReplicationFactory;
53  import org.apache.hadoop.hbase.replication.ReplicationPeer;
54  import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
55  import org.apache.hadoop.hbase.replication.ReplicationPeerZKImpl;
56  import org.apache.hadoop.hbase.replication.ReplicationPeers;
57  import org.apache.hadoop.hbase.replication.ReplicationQueuesClient;
58  import org.apache.hadoop.hbase.util.Pair;
59  import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
60  
61  /**
62   * <p>
63   * This class provides the administrative interface to HBase cluster
64   * replication. In order to use it, the cluster and the client using
65   * ReplicationAdmin must be configured with <code>hbase.replication</code>
66   * set to true.
67   * </p>
68   * <p>
69   * Adding a new peer results in creating new outbound connections from every
70   * region server to a subset of region servers on the slave cluster. Each
71   * new stream of replication will start replicating from the beginning of the
72   * current WAL, meaning that edits from that past will be replicated.
73   * </p>
74   * <p>
75   * Removing a peer is a destructive and irreversible operation that stops
76   * all the replication streams for the given cluster and deletes the metadata
77   * used to keep track of the replication state.
78   * </p>
79   * <p>
80   * To see which commands are available in the shell, type
81   * <code>replication</code>.
82   * </p>
83   */
84  @InterfaceAudience.Public
85  @InterfaceStability.Evolving
86  public class ReplicationAdmin implements Closeable {
87    private static final Log LOG = LogFactory.getLog(ReplicationAdmin.class);
88  
89    public static final String TNAME = "tableName";
90    public static final String CFNAME = "columnFamlyName";
91  
92    // only Global for now, can add other type
93    // such as, 1) no global replication, or 2) the table is replicated to this cluster, etc.
94    public static final String REPLICATIONTYPE = "replicationType";
95    public static final String REPLICATIONGLOBAL = Integer
96        .toString(HConstants.REPLICATION_SCOPE_GLOBAL);
97  
98    private final Connection connection;
99    // TODO: replication should be managed by master. All the classes except ReplicationAdmin should
100   // be moved to hbase-server. Resolve it in HBASE-11392.
101   private final ReplicationQueuesClient replicationQueuesClient;
102   private final ReplicationPeers replicationPeers;
103   /**
104    * A watcher used by replicationPeers and replicationQueuesClient. Keep reference so can dispose
105    * on {@link #close()}.
106    */
107   private final ZooKeeperWatcher zkw;
108 
109   /**
110    * Constructor that creates a connection to the local ZooKeeper ensemble.
111    * @param conf Configuration to use
112    * @throws IOException if an internal replication error occurs
113    * @throws RuntimeException if replication isn't enabled.
114    */
115   public ReplicationAdmin(Configuration conf) throws IOException {
116     if (!conf.getBoolean(HConstants.REPLICATION_ENABLE_KEY,
117         HConstants.REPLICATION_ENABLE_DEFAULT)) {
118       throw new RuntimeException("hbase.replication isn't true, please " +
119           "enable it in order to use replication");
120     }
121     this.connection = ConnectionFactory.createConnection(conf);
122     try {
123       zkw = createZooKeeperWatcher();
124       try {
125         this.replicationQueuesClient =
126             ReplicationFactory.getReplicationQueuesClient(zkw, conf, this.connection);
127         this.replicationQueuesClient.init();
128         this.replicationPeers = ReplicationFactory.getReplicationPeers(zkw, conf,
129           this.replicationQueuesClient, this.connection);
130         this.replicationPeers.init();
131       } catch (Exception exception) {
132         if (zkw != null) {
133           zkw.close();
134         }
135         throw exception;
136       }
137     } catch (Exception exception) {
138       if (connection != null) {
139         connection.close();
140       }
141       if (exception instanceof IOException) {
142         throw (IOException) exception;
143       } else if (exception instanceof RuntimeException) {
144         throw (RuntimeException) exception;
145       } else {
146         throw new IOException("Error initializing the replication admin client.", exception);
147       }
148     }
149   }
150 
151   private ZooKeeperWatcher createZooKeeperWatcher() throws IOException {
152     // This Abortable doesn't 'abort'... it just logs.
153     return new ZooKeeperWatcher(connection.getConfiguration(), "ReplicationAdmin", new Abortable() {
154       @Override
155       public void abort(String why, Throwable e) {
156         LOG.error(why, e);
157         // We used to call system.exit here but this script can be embedded by other programs that
158         // want to do replication stuff... so inappropriate calling System.exit. Just log for now.
159       }
160 
161       @Override
162       public boolean isAborted() {
163         return false;
164       }
165     });
166   }
167 
168   /**
169    * Add a new peer cluster to replicate to.
170    * @param id a short name that identifies the cluster
171    * @param clusterKey the concatenation of the slave cluster's
172    * <code>hbase.zookeeper.quorum:hbase.zookeeper.property.clientPort:zookeeper.znode.parent</code>
173    * @throws IllegalStateException if there's already one slave since
174    * multi-slave isn't supported yet.
175    * @deprecated Use addPeer(String, ReplicationPeerConfig, Map) instead.
176    */
177   @Deprecated
178   public void addPeer(String id, String clusterKey) throws ReplicationException {
179     this.addPeer(id, new ReplicationPeerConfig().setClusterKey(clusterKey), null);
180   }
181 
182   @Deprecated
183   public void addPeer(String id, String clusterKey, String tableCFs)
184     throws ReplicationException {
185     this.replicationPeers.addPeer(id,
186       new ReplicationPeerConfig().setClusterKey(clusterKey), tableCFs);
187   }
188 
189   /**
190    * Add a new remote slave cluster for replication.
191    * @param id a short name that identifies the cluster
192    * @param peerConfig configuration for the replication slave cluster
193    * @param tableCfs the table and column-family list which will be replicated for this peer.
194    * A map from tableName to column family names. An empty collection can be passed
195    * to indicate replicating all column families. Pass null for replicating all table and column
196    * families
197    */
198   public void addPeer(String id, ReplicationPeerConfig peerConfig,
199       Map<TableName, ? extends Collection<String>> tableCfs) throws ReplicationException {
200     this.replicationPeers.addPeer(id, peerConfig, getTableCfsStr(tableCfs));
201   }
202 
203   public static Map<TableName, List<String>> parseTableCFsFromConfig(String tableCFsConfig) {
204     if (tableCFsConfig == null || tableCFsConfig.trim().length() == 0) {
205       return null;
206     }
207 
208     Map<TableName, List<String>> tableCFsMap = null;
209     // TODO: This should be a PB object rather than a String to be parsed!! See HBASE-11393
210     // parse out (table, cf-list) pairs from tableCFsConfig
211     // format: "table1:cf1,cf2;table2:cfA,cfB"
212     String[] tables = tableCFsConfig.split(";");
213     for (String tab : tables) {
214       // 1 ignore empty table config
215       tab = tab.trim();
216       if (tab.length() == 0) {
217         continue;
218       }
219       // 2 split to "table" and "cf1,cf2"
220       //   for each table: "table:cf1,cf2" or "table"
221       String[] pair = tab.split(":");
222       String tabName = pair[0].trim();
223       if (pair.length > 2 || tabName.length() == 0) {
224         LOG.error("ignore invalid tableCFs setting: " + tab);
225         continue;
226       }
227 
228       // 3 parse "cf1,cf2" part to List<cf>
229       List<String> cfs = null;
230       if (pair.length == 2) {
231         String[] cfsList = pair[1].split(",");
232         for (String cf : cfsList) {
233           String cfName = cf.trim();
234           if (cfName.length() > 0) {
235             if (cfs == null) {
236               cfs = new ArrayList<String>();
237             }
238             cfs.add(cfName);
239           }
240         }
241       }
242 
243       // 4 put <table, List<cf>> to map
244       if (tableCFsMap == null) {
245         tableCFsMap = new HashMap<TableName, List<String>>();
246       }
247       tableCFsMap.put(TableName.valueOf(tabName), cfs);
248     }
249     return tableCFsMap;
250   }
251 
252   @VisibleForTesting
253   static String getTableCfsStr(Map<TableName, ? extends Collection<String>> tableCfs) {
254     String tableCfsStr = null;
255     if (tableCfs != null) {
256       // Format: table1:cf1,cf2;table2:cfA,cfB;table3
257       StringBuilder builder = new StringBuilder();
258       for (Entry<TableName, ? extends Collection<String>> entry : tableCfs.entrySet()) {
259         if (builder.length() > 0) {
260           builder.append(";");
261         }
262         builder.append(entry.getKey());
263         if (entry.getValue() != null && !entry.getValue().isEmpty()) {
264           builder.append(":");
265           builder.append(StringUtils.join(entry.getValue(), ","));
266         }
267       }
268       tableCfsStr = builder.toString();
269     }
270     return tableCfsStr;
271   }
272 
273   /**
274    * Removes a peer cluster and stops the replication to it.
275    * @param id a short name that identifies the cluster
276    */
277   public void removePeer(String id) throws ReplicationException {
278     this.replicationPeers.removePeer(id);
279   }
280 
281   /**
282    * Restart the replication stream to the specified peer.
283    * @param id a short name that identifies the cluster
284    */
285   public void enablePeer(String id) throws ReplicationException {
286     this.replicationPeers.enablePeer(id);
287   }
288 
289   /**
290    * Stop the replication stream to the specified peer.
291    * @param id a short name that identifies the cluster
292    */
293   public void disablePeer(String id) throws ReplicationException {
294     this.replicationPeers.disablePeer(id);
295   }
296 
297   /**
298    * Get the number of slave clusters the local cluster has.
299    * @return number of slave clusters
300    */
301   public int getPeersCount() {
302     return this.replicationPeers.getAllPeerIds().size();
303   }
304 
305   /**
306    * Map of this cluster's peers for display.
307    * @return A map of peer ids to peer cluster keys
308    * @deprecated use {@link #listPeerConfigs()}
309    */
310   @Deprecated
311   public Map<String, String> listPeers() {
312     Map<String, ReplicationPeerConfig> peers = this.listPeerConfigs();
313     Map<String, String> ret = new HashMap<String, String>(peers.size());
314 
315     for (Map.Entry<String, ReplicationPeerConfig> entry : peers.entrySet()) {
316       ret.put(entry.getKey(), entry.getValue().getClusterKey());
317     }
318     return ret;
319   }
320 
321   public Map<String, ReplicationPeerConfig> listPeerConfigs() {
322     return this.replicationPeers.getAllPeerConfigs();
323   }
324 
325   public ReplicationPeerConfig getPeerConfig(String id) throws ReplicationException {
326     return this.replicationPeers.getReplicationPeerConfig(id);
327   }
328 
329   /**
330    * Get the replicable table-cf config of the specified peer.
331    * @param id a short name that identifies the cluster
332    */
333   public String getPeerTableCFs(String id) throws ReplicationException {
334     return this.replicationPeers.getPeerTableCFsConfig(id);
335   }
336 
337   /**
338    * Set the replicable table-cf config of the specified peer
339    * @param id a short name that identifies the cluster
340    * @deprecated use {@link #setPeerTableCFs(String, Map)}
341    */
342   @Deprecated
343   public void setPeerTableCFs(String id, String tableCFs) throws ReplicationException {
344     this.replicationPeers.setPeerTableCFsConfig(id, tableCFs);
345   }
346 
347   /**
348    * Append the replicable table-cf config of the specified peer
349    * @param id a short that identifies the cluster
350    * @param tableCfs table-cfs config str
351    */
352   public void appendPeerTableCFs(String id, String tableCfs) throws ReplicationException {
353     appendPeerTableCFs(id, parseTableCFsFromConfig(tableCfs));
354   }
355 
356   /**
357    * Append the replicable table-cf config of the specified peer
358    * @param id a short that identifies the cluster
359    * @param tableCfs A map from tableName to column family names
360    */
361   public void appendPeerTableCFs(String id, Map<TableName, ? extends Collection<String>> tableCfs)
362       throws ReplicationException {
363     if (tableCfs == null) {
364       throw new ReplicationException("tableCfs is null");
365     }
366     Map<TableName, List<String>> preTableCfs = parseTableCFsFromConfig(getPeerTableCFs(id));
367     if (preTableCfs == null) {
368       setPeerTableCFs(id, tableCfs);
369       return;
370     }
371 
372     for (Map.Entry<TableName, ? extends Collection<String>> entry : tableCfs.entrySet()) {
373       TableName table = entry.getKey();
374       Collection<String> appendCfs = entry.getValue();
375       if (preTableCfs.containsKey(table)) {
376         List<String> cfs = preTableCfs.get(table);
377         if (cfs == null || appendCfs == null) {
378           preTableCfs.put(table, null);
379         } else {
380           Set<String> cfSet = new HashSet<String>(cfs);
381           cfSet.addAll(appendCfs);
382           preTableCfs.put(table, Lists.newArrayList(cfSet));
383         }
384       } else {
385         if (appendCfs == null || appendCfs.isEmpty()) {
386           preTableCfs.put(table, null);
387         } else {
388           preTableCfs.put(table, Lists.newArrayList(appendCfs));
389         }
390       }
391     }
392     setPeerTableCFs(id, preTableCfs);
393   }
394 
395   /**
396    * Remove some table-cfs from table-cfs config of the specified peer
397    * @param id a short name that identifies the cluster
398    * @param tableCf table-cfs config str
399    * @throws ReplicationException
400    */
401   public void removePeerTableCFs(String id, String tableCf) throws ReplicationException {
402     removePeerTableCFs(id, parseTableCFsFromConfig(tableCf));
403   }
404 
405   /**
406    * Remove some table-cfs from config of the specified peer
407    * @param id a short name that identifies the cluster
408    * @param tableCfs A map from tableName to column family names
409    * @throws ReplicationException
410    */
411   public void removePeerTableCFs(String id, Map<TableName, ? extends Collection<String>> tableCfs)
412       throws ReplicationException {
413     if (tableCfs == null) {
414       throw new ReplicationException("tableCfs is null");
415     }
416 
417     Map<TableName, List<String>> preTableCfs = parseTableCFsFromConfig(getPeerTableCFs(id));
418     if (preTableCfs == null) {
419       throw new ReplicationException("Table-Cfs for peer" + id + " is null");
420     }
421     for (Map.Entry<TableName, ? extends Collection<String>> entry: tableCfs.entrySet()) {
422       TableName table = entry.getKey();
423       Collection<String> removeCfs = entry.getValue();
424       if (preTableCfs.containsKey(table)) {
425         List<String> cfs = preTableCfs.get(table);
426         if (cfs == null && removeCfs == null) {
427           preTableCfs.remove(table);
428         } else if (cfs != null && removeCfs != null) {
429           Set<String> cfSet = new HashSet<String>(cfs);
430           cfSet.removeAll(removeCfs);
431           if (cfSet.isEmpty()) {
432             preTableCfs.remove(table);
433           } else {
434             preTableCfs.put(table, Lists.newArrayList(cfSet));
435           }
436         } else if (cfs == null && removeCfs != null) {
437           throw new ReplicationException("Cannot remove cf of table: " + table
438               + " which doesn't specify cfs from table-cfs config in peer: " + id);
439         } else if (cfs != null && removeCfs == null) {
440           throw new ReplicationException("Cannot remove table: " + table
441               + " which has specified cfs from table-cfs config in peer: " + id);
442         }
443       } else {
444         throw new ReplicationException("No table: " + table + " in table-cfs config of peer: " + id);
445       }
446     }
447     setPeerTableCFs(id, preTableCfs);
448   }
449 
450   /**
451    * Set the replicable table-cf config of the specified peer
452    * @param id a short name that identifies the cluster
453    * @param tableCfs the table and column-family list which will be replicated for this peer.
454    * A map from tableName to column family names. An empty collection can be passed
455    * to indicate replicating all column families. Pass null for replicating all table and column
456    * families
457    */
458   public void setPeerTableCFs(String id, Map<TableName, ? extends Collection<String>> tableCfs)
459       throws ReplicationException {
460     this.replicationPeers.setPeerTableCFsConfig(id, getTableCfsStr(tableCfs));
461   }
462 
463   /**
464    * Get the state of the specified peer cluster
465    * @param id String format of the Short name that identifies the peer,
466    * an IllegalArgumentException is thrown if it doesn't exist
467    * @return true if replication is enabled to that peer, false if it isn't
468    */
469   public boolean getPeerState(String id) throws ReplicationException {
470     return this.replicationPeers.getStatusOfPeerFromBackingStore(id);
471   }
472 
473   @Override
474   public void close() throws IOException {
475     if (this.zkw != null) {
476       this.zkw.close();
477     }
478     if (this.connection != null) {
479       this.connection.close();
480     }
481   }
482 
483 
484   /**
485    * Find all column families that are replicated from this cluster
486    * @return the full list of the replicated column families of this cluster as:
487    *        tableName, family name, replicationType
488    *
489    * Currently replicationType is Global. In the future, more replication
490    * types may be extended here. For example
491    *  1) the replication may only apply to selected peers instead of all peers
492    *  2) the replicationType may indicate the host Cluster servers as Slave
493    *     for the table:columnFam.
494    */
495   public List<HashMap<String, String>> listReplicated() throws IOException {
496     List<HashMap<String, String>> replicationColFams = new ArrayList<HashMap<String, String>>();
497 
498     Admin admin = connection.getAdmin();
499     HTableDescriptor[] tables;
500     try {
501       tables = admin.listTables();
502     } finally {
503       if (admin!= null) admin.close();
504     }
505 
506     for (HTableDescriptor table : tables) {
507       HColumnDescriptor[] columns = table.getColumnFamilies();
508       String tableName = table.getNameAsString();
509       for (HColumnDescriptor column : columns) {
510         if (column.getScope() != HConstants.REPLICATION_SCOPE_LOCAL) {
511           // At this moment, the columfam is replicated to all peers
512           HashMap<String, String> replicationEntry = new HashMap<String, String>();
513           replicationEntry.put(TNAME, tableName);
514           replicationEntry.put(CFNAME, column.getNameAsString());
515           replicationEntry.put(REPLICATIONTYPE, REPLICATIONGLOBAL);
516           replicationColFams.add(replicationEntry);
517         }
518       }
519     }
520 
521     return replicationColFams;
522   }
523 
524   /**
525    * Enable a table's replication switch.
526    * @param tableName name of the table
527    * @throws IOException if a remote or network exception occurs
528    */
529   public void enableTableRep(final TableName tableName) throws IOException {
530     if (tableName == null) {
531       throw new IllegalArgumentException("Table name cannot be null");
532     }
533     try (Admin admin = this.connection.getAdmin()) {
534       if (!admin.tableExists(tableName)) {
535         throw new TableNotFoundException("Table '" + tableName.getNameAsString()
536             + "' does not exists.");
537       }
538     }
539     byte[][] splits = getTableSplitRowKeys(tableName);
540     checkAndSyncTableDescToPeers(tableName, splits);
541     setTableRep(tableName, true);
542   }
543 
544   /**
545    * Disable a table's replication switch.
546    * @param tableName name of the table
547    * @throws IOException if a remote or network exception occurs
548    */
549   public void disableTableRep(final TableName tableName) throws IOException {
550     if (tableName == null) {
551       throw new IllegalArgumentException("Table name is null");
552     }
553     try (Admin admin = this.connection.getAdmin()) {
554       if (!admin.tableExists(tableName)) {
555         throw new TableNotFoundException("Table '" + tableName.getNamespaceAsString()
556             + "' does not exists.");
557       }
558     }
559     setTableRep(tableName, false);
560   }
561 
562   /**
563    * Get the split row keys of table
564    * @param tableName table name
565    * @return array of split row keys
566    * @throws IOException
567    */
568   private byte[][] getTableSplitRowKeys(TableName tableName) throws IOException {
569     try (RegionLocator locator = connection.getRegionLocator(tableName);) {
570       byte[][] startKeys = locator.getStartKeys();
571       if (startKeys.length == 1) {
572         return null;
573       }
574       byte[][] splits = new byte[startKeys.length - 1][];
575       for (int i = 1; i < startKeys.length; i++) {
576         splits[i - 1] = startKeys[i];
577       }
578       return splits;
579     }
580   }
581 
582   /**
583    * Connect to peer and check the table descriptor on peer:
584    * <ol>
585    * <li>Create the same table on peer when not exist.</li>
586    * <li>Throw exception if the table exists on peer cluster but descriptors are not same.</li>
587    * </ol>
588    * @param tableName name of the table to sync to the peer
589    * @param splits table split keys
590    * @throws IOException
591    */
592   private void checkAndSyncTableDescToPeers(final TableName tableName, final byte[][] splits)
593       throws IOException {
594     List<ReplicationPeer> repPeers = listReplicationPeers();
595     if (repPeers == null || repPeers.size() <= 0) {
596       throw new IllegalArgumentException("Found no peer cluster for replication.");
597     }
598     
599     final TableName onlyTableNameQualifier = TableName.valueOf(tableName.getQualifierAsString());
600     
601     for (ReplicationPeer repPeer : repPeers) {
602       Map<TableName, List<String>> tableCFMap = repPeer.getTableCFs();
603       // TODO Currently peer TableCFs will not include namespace so we need to check only for table
604       // name without namespace in it. Need to correct this logic once we fix HBASE-11386.
605       if (tableCFMap != null && !tableCFMap.containsKey(onlyTableNameQualifier)) {
606         continue;
607       }
608 
609       Configuration peerConf = repPeer.getConfiguration();
610       HTableDescriptor htd = null;
611       try (Connection conn = ConnectionFactory.createConnection(peerConf);
612           Admin admin = this.connection.getAdmin();
613           Admin repHBaseAdmin = conn.getAdmin()) {
614         htd = admin.getTableDescriptor(tableName);
615         HTableDescriptor peerHtd = null;
616         if (!repHBaseAdmin.tableExists(tableName)) {
617           repHBaseAdmin.createTable(htd, splits);
618         } else {
619           peerHtd = repHBaseAdmin.getTableDescriptor(tableName);
620           if (peerHtd == null) {
621             throw new IllegalArgumentException("Failed to get table descriptor for table "
622                 + tableName.getNameAsString() + " from peer cluster " + repPeer.getId());
623           } else if (!peerHtd.equals(htd)) {
624             throw new IllegalArgumentException("Table " + tableName.getNameAsString()
625                 + " exists in peer cluster " + repPeer.getId()
626                 + ", but the table descriptors are not same when compared with source cluster."
627                 + " Thus can not enable the table's replication switch.");
628           }
629         }
630       }
631     }
632   }
633 
634   @VisibleForTesting
635   List<ReplicationPeer> listReplicationPeers() {
636     Map<String, ReplicationPeerConfig> peers = listPeerConfigs();
637     if (peers == null || peers.size() <= 0) {
638       return null;
639     }
640     List<ReplicationPeer> listOfPeers = new ArrayList<ReplicationPeer>(peers.size());
641     for (Entry<String, ReplicationPeerConfig> peerEntry : peers.entrySet()) {
642       String peerId = peerEntry.getKey();
643       try {
644         Pair<ReplicationPeerConfig, Configuration> pair = this.replicationPeers.getPeerConf(peerId);
645         Configuration peerConf = pair.getSecond();
646         ReplicationPeer peer = new ReplicationPeerZKImpl(peerConf, peerId, pair.getFirst(),
647             parseTableCFsFromConfig(this.getPeerTableCFs(peerId)));
648         listOfPeers.add(peer);
649       } catch (ReplicationException e) {
650         LOG.warn("Failed to get valid replication peers. "
651             + "Error connecting to peer cluster with peerId=" + peerId + ". Error message="
652             + e.getMessage());
653         LOG.debug("Failure details to get valid replication peers.", e);
654         continue;
655       }
656     }
657     return listOfPeers;
658   }
659 
660   /**
661    * Set the table's replication switch if the table's replication switch is already not set.
662    * @param tableName name of the table
663    * @param isRepEnabled is replication switch enable or disable
664    * @throws IOException if a remote or network exception occurs
665    */
666   private void setTableRep(final TableName tableName, boolean isRepEnabled) throws IOException {
667     Admin admin = null;
668     try {
669       admin = this.connection.getAdmin();
670       HTableDescriptor htd = admin.getTableDescriptor(tableName);
671       if (isTableRepEnabled(htd) ^ isRepEnabled) {
672         boolean isOnlineSchemaUpdateEnabled =
673             this.connection.getConfiguration()
674                 .getBoolean("hbase.online.schema.update.enable", true);
675         if (!isOnlineSchemaUpdateEnabled) {
676           admin.disableTable(tableName);
677         }
678         for (HColumnDescriptor hcd : htd.getFamilies()) {
679           hcd.setScope(isRepEnabled ? HConstants.REPLICATION_SCOPE_GLOBAL
680               : HConstants.REPLICATION_SCOPE_LOCAL);
681         }
682         admin.modifyTable(tableName, htd);
683         if (!isOnlineSchemaUpdateEnabled) {
684           admin.enableTable(tableName);
685         }
686       }
687     } finally {
688       if (admin != null) {
689         try {
690           admin.close();
691         } catch (IOException e) {
692           LOG.warn("Failed to close admin connection.");
693           LOG.debug("Details on failure to close admin connection.", e);
694         }
695       }
696     }
697   }
698 
699   /**
700    * @param htd table descriptor details for the table to check
701    * @return true if table's replication switch is enabled
702    */
703   private boolean isTableRepEnabled(HTableDescriptor htd) {
704     for (HColumnDescriptor hcd : htd.getFamilies()) {
705       if (hcd.getScope() != HConstants.REPLICATION_SCOPE_GLOBAL) {
706         return false;
707       }
708     }
709     return true;
710   }
711 }