1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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
93
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
100
101 private final ReplicationQueuesClient replicationQueuesClient;
102 private final ReplicationPeers replicationPeers;
103
104
105
106
107 private final ZooKeeperWatcher zkw;
108
109
110
111
112
113
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
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
158
159 }
160
161 @Override
162 public boolean isAborted() {
163 return false;
164 }
165 });
166 }
167
168
169
170
171
172
173
174
175
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
191
192
193
194
195
196
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
210
211
212 String[] tables = tableCFsConfig.split(";");
213 for (String tab : tables) {
214
215 tab = tab.trim();
216 if (tab.length() == 0) {
217 continue;
218 }
219
220
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
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
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
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
275
276
277 public void removePeer(String id) throws ReplicationException {
278 this.replicationPeers.removePeer(id);
279 }
280
281
282
283
284
285 public void enablePeer(String id) throws ReplicationException {
286 this.replicationPeers.enablePeer(id);
287 }
288
289
290
291
292
293 public void disablePeer(String id) throws ReplicationException {
294 this.replicationPeers.disablePeer(id);
295 }
296
297
298
299
300
301 public int getPeersCount() {
302 return this.replicationPeers.getAllPeerIds().size();
303 }
304
305
306
307
308
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
331
332
333 public String getPeerTableCFs(String id) throws ReplicationException {
334 return this.replicationPeers.getPeerTableCFsConfig(id);
335 }
336
337
338
339
340
341
342 @Deprecated
343 public void setPeerTableCFs(String id, String tableCFs) throws ReplicationException {
344 this.replicationPeers.setPeerTableCFsConfig(id, tableCFs);
345 }
346
347
348
349
350
351
352 public void appendPeerTableCFs(String id, String tableCfs) throws ReplicationException {
353 appendPeerTableCFs(id, parseTableCFsFromConfig(tableCfs));
354 }
355
356
357
358
359
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
397
398
399
400
401 public void removePeerTableCFs(String id, String tableCf) throws ReplicationException {
402 removePeerTableCFs(id, parseTableCFsFromConfig(tableCf));
403 }
404
405
406
407
408
409
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
452
453
454
455
456
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
465
466
467
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
486
487
488
489
490
491
492
493
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
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
526
527
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
546
547
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
564
565
566
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
584
585
586
587
588
589
590
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
604
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
662
663
664
665
666 private void setTableRep(final TableName tableName, boolean enableRep) throws IOException {
667 Admin admin = null;
668 try {
669 admin = this.connection.getAdmin();
670 HTableDescriptor htd = admin.getTableDescriptor(tableName);
671 ReplicationState currentReplicationState = getTableReplicationState(htd);
672 if (enableRep && currentReplicationState != ReplicationState.ENABLED
673 || !enableRep && currentReplicationState != ReplicationState.DISABLED) {
674 boolean isOnlineSchemaUpdateEnabled =
675 this.connection.getConfiguration()
676 .getBoolean("hbase.online.schema.update.enable", true);
677 if (!isOnlineSchemaUpdateEnabled) {
678 admin.disableTable(tableName);
679 }
680 for (HColumnDescriptor hcd : htd.getFamilies()) {
681 hcd.setScope(enableRep ? HConstants.REPLICATION_SCOPE_GLOBAL
682 : HConstants.REPLICATION_SCOPE_LOCAL);
683 }
684 admin.modifyTable(tableName, htd);
685 if (!isOnlineSchemaUpdateEnabled) {
686 admin.enableTable(tableName);
687 }
688 }
689 } finally {
690 if (admin != null) {
691 try {
692 admin.close();
693 } catch (IOException e) {
694 LOG.warn("Failed to close admin connection.");
695 LOG.debug("Details on failure to close admin connection.", e);
696 }
697 }
698 }
699 }
700
701
702
703
704 private enum ReplicationState {
705 ENABLED,
706 MIXED,
707 DISABLED
708 }
709
710
711
712
713
714 private ReplicationState getTableReplicationState(HTableDescriptor htd) {
715 boolean hasEnabled = false;
716 boolean hasDisabled = false;
717
718 for (HColumnDescriptor hcd : htd.getFamilies()) {
719 if (hcd.getScope() != HConstants.REPLICATION_SCOPE_GLOBAL) {
720 hasDisabled = true;
721 } else {
722 hasEnabled = true;
723 }
724 }
725
726 if (hasEnabled && hasDisabled) return ReplicationState.MIXED;
727 if (hasEnabled) return ReplicationState.ENABLED;
728 return ReplicationState.DISABLED;
729 }
730 }