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 java.io.Closeable;
22 import java.io.IOException;
23 import java.util.ArrayList;
24 import java.util.Collection;
25 import java.util.HashMap;
26 import java.util.HashSet;
27 import java.util.Iterator;
28 import java.util.List;
29 import java.util.Map;
30 import java.util.Map.Entry;
31 import java.util.Set;
32 import java.util.TreeMap;
33
34 import org.apache.commons.lang.StringUtils;
35 import org.apache.commons.logging.Log;
36 import org.apache.commons.logging.LogFactory;
37 import org.apache.hadoop.conf.Configuration;
38 import org.apache.hadoop.hbase.*;
39 import org.apache.hadoop.hbase.classification.InterfaceAudience;
40 import org.apache.hadoop.hbase.classification.InterfaceStability;
41 import org.apache.hadoop.hbase.client.Admin;
42 import org.apache.hadoop.hbase.client.Connection;
43 import org.apache.hadoop.hbase.client.ConnectionFactory;
44 import org.apache.hadoop.hbase.client.RegionLocator;
45 import org.apache.hadoop.hbase.replication.ReplicationException;
46 import org.apache.hadoop.hbase.replication.ReplicationFactory;
47 import org.apache.hadoop.hbase.replication.ReplicationPeer;
48 import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
49 import org.apache.hadoop.hbase.replication.ReplicationPeerZKImpl;
50 import org.apache.hadoop.hbase.replication.ReplicationPeers;
51 import org.apache.hadoop.hbase.replication.ReplicationQueuesClient;
52 import org.apache.hadoop.hbase.replication.ReplicationSerDeHelper;
53 import org.apache.hadoop.hbase.util.Pair;
54 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
55
56 import com.google.common.annotations.VisibleForTesting;
57 import com.google.common.collect.Lists;
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82 @InterfaceAudience.Public
83 @InterfaceStability.Evolving
84 public class ReplicationAdmin implements Closeable {
85 private static final Log LOG = LogFactory.getLog(ReplicationAdmin.class);
86
87 public static final String TNAME = "tableName";
88 public static final String CFNAME = "columnFamlyName";
89 public static final String REPLICATION_WALENTRYFILTER_CONFIG_KEY
90 = "hbase.replication.source.custom.walentryfilters";
91
92
93 public static final String REPLICATIONTYPE = "replicationType";
94 public static final String REPLICATIONGLOBAL = Integer
95 .toString(HConstants.REPLICATION_SCOPE_GLOBAL);
96
97 private final Connection connection;
98
99
100 private final ReplicationQueuesClient replicationQueuesClient;
101 private final ReplicationPeers replicationPeers;
102
103
104
105
106 private final ZooKeeperWatcher zkw;
107
108
109
110
111
112
113
114 public ReplicationAdmin(Configuration conf) throws IOException {
115 if (!conf.getBoolean(HConstants.REPLICATION_ENABLE_KEY,
116 HConstants.REPLICATION_ENABLE_DEFAULT)) {
117 throw new RuntimeException("hbase.replication isn't true, please " +
118 "enable it in order to use replication");
119 }
120 this.connection = ConnectionFactory.createConnection(conf);
121 try {
122 zkw = createZooKeeperWatcher();
123 try {
124 this.replicationQueuesClient =
125 ReplicationFactory.getReplicationQueuesClient(zkw, conf, this.connection);
126 this.replicationQueuesClient.init();
127 this.replicationPeers = ReplicationFactory.getReplicationPeers(zkw, conf,
128 this.replicationQueuesClient, this.connection);
129 this.replicationPeers.init();
130 } catch (Exception exception) {
131 if (zkw != null) {
132 zkw.close();
133 }
134 throw exception;
135 }
136 } catch (Exception exception) {
137 if (connection != null) {
138 connection.close();
139 }
140 if (exception instanceof IOException) {
141 throw (IOException) exception;
142 } else if (exception instanceof RuntimeException) {
143 throw (RuntimeException) exception;
144 } else {
145 throw new IOException("Error initializing the replication admin client.", exception);
146 }
147 }
148 }
149
150 private ZooKeeperWatcher createZooKeeperWatcher() throws IOException {
151
152 return new ZooKeeperWatcher(connection.getConfiguration(), "ReplicationAdmin", new Abortable() {
153 @Override
154 public void abort(String why, Throwable e) {
155 LOG.error(why, e);
156
157
158 }
159
160 @Override
161 public boolean isAborted() {
162 return false;
163 }
164 });
165 }
166
167
168
169
170
171
172
173
174
175
176 @Deprecated
177 public void addPeer(String id, String clusterKey) throws ReplicationException {
178 this.addPeer(id, new ReplicationPeerConfig().setClusterKey(clusterKey), null);
179 }
180
181 @Deprecated
182 public void addPeer(String id, String clusterKey, String tableCFs)
183 throws ReplicationException {
184 this.addPeer(id, new ReplicationPeerConfig().setClusterKey(clusterKey),
185 parseTableCFsFromConfig(tableCFs));
186 }
187
188
189
190
191
192
193
194
195
196
197 public void addPeer(String id, ReplicationPeerConfig peerConfig,
198 Map<TableName, ? extends Collection<String>> tableCfs) throws ReplicationException {
199 if (tableCfs != null) {
200 peerConfig.setTableCFsMap(tableCfs);
201 }
202 checkConfiguredWALEntryFilters(peerConfig);
203 this.replicationPeers.addPeer(id, peerConfig);
204 }
205
206
207
208
209
210
211 public void addPeer(String id, ReplicationPeerConfig peerConfig) throws ReplicationException {
212 checkConfiguredWALEntryFilters(peerConfig);
213 this.replicationPeers.addPeer(id, peerConfig);
214 }
215
216 public void updatePeerConfig(String id, ReplicationPeerConfig peerConfig)
217 throws ReplicationException {
218 checkConfiguredWALEntryFilters(peerConfig);
219 this.replicationPeers.updatePeerConfig(id, peerConfig);
220 }
221
222 public static Map<TableName, List<String>> parseTableCFsFromConfig(String tableCFsConfig) {
223 return ReplicationSerDeHelper.parseTableCFsFromConfig(tableCFsConfig);
224 }
225
226 @VisibleForTesting
227 static String getTableCfsStr(Map<TableName, ? extends Collection<String>> tableCfs) {
228 String tableCfsStr = null;
229 if (tableCfs != null) {
230
231 StringBuilder builder = new StringBuilder();
232 for (Entry<TableName, ? extends Collection<String>> entry : tableCfs.entrySet()) {
233 if (builder.length() > 0) {
234 builder.append(";");
235 }
236 builder.append(entry.getKey());
237 if (entry.getValue() != null && !entry.getValue().isEmpty()) {
238 builder.append(":");
239 builder.append(StringUtils.join(entry.getValue(), ","));
240 }
241 }
242 tableCfsStr = builder.toString();
243 }
244 return tableCfsStr;
245 }
246
247
248
249
250
251 public void removePeer(String id) throws ReplicationException {
252 this.replicationPeers.removePeer(id);
253 }
254
255
256
257
258
259 public void enablePeer(String id) throws ReplicationException {
260 this.replicationPeers.enablePeer(id);
261 }
262
263
264
265
266
267 public void disablePeer(String id) throws ReplicationException {
268 this.replicationPeers.disablePeer(id);
269 }
270
271
272
273
274
275 public int getPeersCount() {
276 return this.replicationPeers.getAllPeerIds().size();
277 }
278
279
280
281
282
283
284 @Deprecated
285 public Map<String, String> listPeers() {
286 Map<String, ReplicationPeerConfig> peers = this.listPeerConfigs();
287 Map<String, String> ret = new HashMap<String, String>(peers.size());
288
289 for (Map.Entry<String, ReplicationPeerConfig> entry : peers.entrySet()) {
290 ret.put(entry.getKey(), entry.getValue().getClusterKey());
291 }
292 return ret;
293 }
294
295 public Map<String, ReplicationPeerConfig> listPeerConfigs() {
296 return this.replicationPeers.getAllPeerConfigs();
297 }
298
299 public ReplicationPeerConfig getPeerConfig(String id) throws ReplicationException {
300 return this.replicationPeers.getReplicationPeerConfig(id);
301 }
302
303
304
305
306
307 public String getPeerTableCFs(String id) throws ReplicationException {
308 return ReplicationSerDeHelper.convertToString(this.replicationPeers.getPeerTableCFsConfig(id));
309 }
310
311
312
313
314
315
316 @Deprecated
317 public void setPeerTableCFs(String id, String tableCFs) throws ReplicationException {
318 this.setPeerTableCFs(id, parseTableCFsFromConfig(tableCFs));
319 }
320
321
322
323
324
325
326 public void appendPeerTableCFs(String id, String tableCfs) throws ReplicationException {
327 appendPeerTableCFs(id, ReplicationSerDeHelper.parseTableCFsFromConfig(tableCfs));
328 }
329
330
331
332
333
334
335 public void appendPeerTableCFs(String id, Map<TableName, ? extends Collection<String>> tableCfs)
336 throws ReplicationException {
337 if (tableCfs == null) {
338 throw new ReplicationException("tableCfs is null");
339 }
340 Map<TableName, List<String>> preTableCfs = this.replicationPeers.getPeerTableCFsConfig(id);
341 if (preTableCfs == null) {
342 setPeerTableCFs(id, tableCfs);
343 return;
344 }
345
346 for (Map.Entry<TableName, ? extends Collection<String>> entry : tableCfs.entrySet()) {
347 TableName table = entry.getKey();
348 Collection<String> appendCfs = entry.getValue();
349 if (preTableCfs.containsKey(table)) {
350 List<String> cfs = preTableCfs.get(table);
351 if (cfs == null || appendCfs == null) {
352 preTableCfs.put(table, null);
353 } else {
354 Set<String> cfSet = new HashSet<String>(cfs);
355 cfSet.addAll(appendCfs);
356 preTableCfs.put(table, Lists.newArrayList(cfSet));
357 }
358 } else {
359 if (appendCfs == null || appendCfs.isEmpty()) {
360 preTableCfs.put(table, null);
361 } else {
362 preTableCfs.put(table, Lists.newArrayList(appendCfs));
363 }
364 }
365 }
366 setPeerTableCFs(id, preTableCfs);
367 }
368
369
370
371
372
373
374
375 public void removePeerTableCFs(String id, String tableCf) throws ReplicationException {
376 removePeerTableCFs(id, ReplicationSerDeHelper.parseTableCFsFromConfig(tableCf));
377 }
378
379
380
381
382
383
384
385 public void removePeerTableCFs(String id, Map<TableName, ? extends Collection<String>> tableCfs)
386 throws ReplicationException {
387 if (tableCfs == null) {
388 throw new ReplicationException("tableCfs is null");
389 }
390
391 Map<TableName, List<String>> preTableCfs = this.replicationPeers.getPeerTableCFsConfig(id);
392 if (preTableCfs == null) {
393 throw new ReplicationException("Table-Cfs for peer" + id + " is null");
394 }
395 for (Map.Entry<TableName, ? extends Collection<String>> entry: tableCfs.entrySet()) {
396 TableName table = entry.getKey();
397 Collection<String> removeCfs = entry.getValue();
398 if (preTableCfs.containsKey(table)) {
399 List<String> cfs = preTableCfs.get(table);
400 if (cfs == null && removeCfs == null) {
401 preTableCfs.remove(table);
402 } else if (cfs != null && removeCfs != null) {
403 Set<String> cfSet = new HashSet<String>(cfs);
404 cfSet.removeAll(removeCfs);
405 if (cfSet.isEmpty()) {
406 preTableCfs.remove(table);
407 } else {
408 preTableCfs.put(table, Lists.newArrayList(cfSet));
409 }
410 } else if (cfs == null && removeCfs != null) {
411 throw new ReplicationException("Cannot remove cf of table: " + table
412 + " which doesn't specify cfs from table-cfs config in peer: " + id);
413 } else if (cfs != null && removeCfs == null) {
414 throw new ReplicationException("Cannot remove table: " + table
415 + " which has specified cfs from table-cfs config in peer: " + id);
416 }
417 } else {
418 throw new ReplicationException("No table: " + table + " in table-cfs config of peer: " + id);
419 }
420 }
421 setPeerTableCFs(id, preTableCfs);
422 }
423
424
425
426
427
428
429
430
431
432 public void setPeerTableCFs(String id, Map<TableName, ? extends Collection<String>> tableCfs)
433 throws ReplicationException {
434 this.replicationPeers.setPeerTableCFsConfig(id, tableCfs);
435 }
436
437
438
439
440
441
442
443 public boolean getPeerState(String id) throws ReplicationException {
444 return this.replicationPeers.getStatusOfPeerFromBackingStore(id);
445 }
446
447 @Override
448 public void close() throws IOException {
449 if (this.zkw != null) {
450 this.zkw.close();
451 }
452 if (this.connection != null) {
453 this.connection.close();
454 }
455 }
456
457
458
459
460
461
462
463
464
465
466
467
468
469 public List<HashMap<String, String>> listReplicated() throws IOException {
470 List<HashMap<String, String>> replicationColFams = new ArrayList<HashMap<String, String>>();
471
472 Admin admin = connection.getAdmin();
473 HTableDescriptor[] tables;
474 try {
475 tables = admin.listTables();
476 } finally {
477 if (admin!= null) admin.close();
478 }
479
480 for (HTableDescriptor table : tables) {
481 HColumnDescriptor[] columns = table.getColumnFamilies();
482 String tableName = table.getNameAsString();
483 for (HColumnDescriptor column : columns) {
484 if (column.getScope() != HConstants.REPLICATION_SCOPE_LOCAL) {
485
486 HashMap<String, String> replicationEntry = new HashMap<String, String>();
487 replicationEntry.put(TNAME, tableName);
488 replicationEntry.put(CFNAME, column.getNameAsString());
489 replicationEntry.put(REPLICATIONTYPE, REPLICATIONGLOBAL);
490 replicationColFams.add(replicationEntry);
491 }
492 }
493 }
494
495 return replicationColFams;
496 }
497
498
499
500
501
502
503 public void enableTableRep(final TableName tableName) throws IOException {
504 if (tableName == null) {
505 throw new IllegalArgumentException("Table name cannot be null");
506 }
507 try (Admin admin = this.connection.getAdmin()) {
508 if (!admin.tableExists(tableName)) {
509 throw new TableNotFoundException("Table '" + tableName.getNameAsString()
510 + "' does not exists.");
511 }
512 }
513 byte[][] splits = getTableSplitRowKeys(tableName);
514 checkAndSyncTableDescToPeers(tableName, splits);
515 setTableRep(tableName, true);
516 }
517
518
519
520
521
522
523 public void disableTableRep(final TableName tableName) throws IOException {
524 if (tableName == null) {
525 throw new IllegalArgumentException("Table name is null");
526 }
527 try (Admin admin = this.connection.getAdmin()) {
528 if (!admin.tableExists(tableName)) {
529 throw new TableNotFoundException("Table '" + tableName.getNamespaceAsString()
530 + "' does not exists.");
531 }
532 }
533 setTableRep(tableName, false);
534 }
535
536
537
538
539
540
541
542 private byte[][] getTableSplitRowKeys(TableName tableName) throws IOException {
543 try (RegionLocator locator = connection.getRegionLocator(tableName);) {
544 byte[][] startKeys = locator.getStartKeys();
545 if (startKeys.length == 1) {
546 return null;
547 }
548 byte[][] splits = new byte[startKeys.length - 1][];
549 for (int i = 1; i < startKeys.length; i++) {
550 splits[i - 1] = startKeys[i];
551 }
552 return splits;
553 }
554 }
555
556
557
558
559
560
561
562
563
564
565
566
567
568 private void checkAndSyncTableDescToPeers(final TableName tableName, final byte[][] splits)
569 throws IOException {
570 List<ReplicationPeer> repPeers = listReplicationPeers();
571 if (repPeers == null || repPeers.size() <= 0) {
572 throw new IllegalArgumentException("Found no peer cluster for replication.");
573 }
574
575 final TableName onlyTableNameQualifier = TableName.valueOf(tableName.getQualifierAsString());
576
577 for (ReplicationPeer repPeer : repPeers) {
578 Map<TableName, List<String>> tableCFMap = repPeer.getTableCFs();
579
580
581 if (tableCFMap != null && !tableCFMap.containsKey(onlyTableNameQualifier)) {
582 continue;
583 }
584
585 Configuration peerConf = repPeer.getConfiguration();
586 HTableDescriptor localHtd = null;
587 try (Connection conn = ConnectionFactory.createConnection(peerConf);
588 Admin admin = this.connection.getAdmin();
589 Admin repHBaseAdmin = conn.getAdmin()) {
590 localHtd = admin.getTableDescriptor(tableName);
591 HTableDescriptor peerHtd = null;
592 if (!repHBaseAdmin.tableExists(tableName)) {
593 repHBaseAdmin.createTable(localHtd, splits);
594 } else {
595 peerHtd = repHBaseAdmin.getTableDescriptor(tableName);
596 if (peerHtd == null) {
597 throw new IllegalArgumentException("Failed to get table descriptor for table "
598 + tableName.getNameAsString() + " from peer cluster " + repPeer.getId());
599 }
600 if (!compareForReplication(peerHtd, localHtd)) {
601 throw new IllegalArgumentException("Table " + tableName.getNameAsString()
602 + " exists in peer cluster " + repPeer.getId()
603 + ", but the table descriptors are not same when compared with source cluster."
604 + " Thus can not enable the table's replication switch.");
605 }
606 }
607 }
608 }
609 }
610
611 @VisibleForTesting
612 public void peerAdded(String id) throws ReplicationException {
613 this.replicationPeers.peerAdded(id);
614 }
615
616 @VisibleForTesting
617 List<ReplicationPeer> listReplicationPeers() {
618 Map<String, ReplicationPeerConfig> peers = listPeerConfigs();
619 if (peers == null || peers.size() <= 0) {
620 return null;
621 }
622 List<ReplicationPeer> listOfPeers = new ArrayList<ReplicationPeer>(peers.size());
623 for (Entry<String, ReplicationPeerConfig> peerEntry : peers.entrySet()) {
624 String peerId = peerEntry.getKey();
625 try {
626 Pair<ReplicationPeerConfig, Configuration> pair = this.replicationPeers.getPeerConf(peerId);
627 Configuration peerConf = pair.getSecond();
628 ReplicationPeer peer = new ReplicationPeerZKImpl(zkw, pair.getSecond(),
629 peerId, pair.getFirst(), this.connection);
630 listOfPeers.add(peer);
631 } catch (ReplicationException e) {
632 LOG.warn("Failed to get valid replication peers. "
633 + "Error connecting to peer cluster with peerId=" + peerId + ". Error message="
634 + e.getMessage());
635 LOG.debug("Failure details to get valid replication peers.", e);
636 continue;
637 }
638 }
639 return listOfPeers;
640 }
641
642
643
644
645
646
647
648 private void setTableRep(final TableName tableName, boolean enableRep) throws IOException {
649 Admin admin = null;
650 try {
651 admin = this.connection.getAdmin();
652 HTableDescriptor htd = admin.getTableDescriptor(tableName);
653 ReplicationState currentReplicationState = getTableReplicationState(htd);
654 if ((enableRep && currentReplicationState != ReplicationState.ENABLED)
655 || (!enableRep && currentReplicationState != ReplicationState.DISABLED)) {
656 boolean isOnlineSchemaUpdateEnabled =
657 this.connection.getConfiguration()
658 .getBoolean("hbase.online.schema.update.enable", true);
659 if (!isOnlineSchemaUpdateEnabled) {
660 admin.disableTable(tableName);
661 }
662 for (HColumnDescriptor hcd : htd.getFamilies()) {
663 hcd.setScope(enableRep ? HConstants.REPLICATION_SCOPE_GLOBAL
664 : HConstants.REPLICATION_SCOPE_LOCAL);
665 }
666 admin.modifyTable(tableName, htd);
667 if (!isOnlineSchemaUpdateEnabled) {
668 admin.enableTable(tableName);
669 }
670 }
671 } finally {
672 if (admin != null) {
673 try {
674 admin.close();
675 } catch (IOException e) {
676 LOG.warn("Failed to close admin connection.");
677 LOG.debug("Details on failure to close admin connection.", e);
678 }
679 }
680 }
681 }
682
683
684
685
686 private enum ReplicationState {
687 ENABLED,
688 MIXED,
689 DISABLED
690 }
691
692
693
694
695
696 private ReplicationState getTableReplicationState(HTableDescriptor htd) {
697 boolean hasEnabled = false;
698 boolean hasDisabled = false;
699
700 for (HColumnDescriptor hcd : htd.getFamilies()) {
701 if (hcd.getScope() != HConstants.REPLICATION_SCOPE_GLOBAL) {
702 hasDisabled = true;
703 } else {
704 hasEnabled = true;
705 }
706 }
707
708 if (hasEnabled && hasDisabled) return ReplicationState.MIXED;
709 if (hasEnabled) return ReplicationState.ENABLED;
710 return ReplicationState.DISABLED;
711 }
712
713 @SuppressWarnings("unchecked")
714 private void checkConfiguredWALEntryFilters(ReplicationPeerConfig peerConfig)
715 throws ReplicationException {
716 String filterCSV = peerConfig.getConfiguration().
717 get(REPLICATION_WALENTRYFILTER_CONFIG_KEY);
718 if (filterCSV != null && !filterCSV.isEmpty()) {
719 String[] filters = filterCSV.split(",");
720 for (String filter : filters) {
721 try {
722 Class<?> clazz = Class.forName(filter);
723 Object o = clazz.getDeclaredConstructor().newInstance();
724 } catch (Exception e) {
725 throw new ReplicationException("Configured WALEntryFilter " + filter +
726 " could not be created. Failing add/update " + "peer operation.", e);
727 }
728 }
729 }
730 }
731
732
733
734
735
736
737
738
739
740 private boolean copyReplicationScope(final HTableDescriptor peerHtd,
741 final HTableDescriptor localHtd) {
742
743
744 int result = peerHtd.getTableName().compareTo(localHtd.getTableName());
745
746 if (result == 0) {
747 Iterator<HColumnDescriptor> remoteHCDIter = peerHtd.getFamilies().iterator();
748 Iterator<HColumnDescriptor> localHCDIter = localHtd.getFamilies().iterator();
749
750 while (remoteHCDIter.hasNext() && localHCDIter.hasNext()) {
751 HColumnDescriptor remoteHCD = remoteHCDIter.next();
752 HColumnDescriptor localHCD = localHCDIter.next();
753
754 String remoteHCDName = remoteHCD.getNameAsString();
755 String localHCDName = localHCD.getNameAsString();
756
757 if (remoteHCDName.equals(localHCDName)) {
758 remoteHCD.setScope(localHCD.getScope());
759 } else {
760 result = -1;
761 break;
762 }
763 }
764 if (remoteHCDIter.hasNext() || localHCDIter.hasNext()) {
765 return false;
766 }
767 }
768
769 return result == 0;
770 }
771
772
773
774
775
776
777
778
779
780 private boolean compareForReplication(HTableDescriptor peerHtd, HTableDescriptor localHtd) {
781 if (peerHtd == null) {
782 return false;
783 }
784 if (peerHtd.equals(localHtd)) {
785 return true;
786 }
787 boolean result = false;
788
789
790
791 HTableDescriptor peerHtdCopy = new HTableDescriptor(peerHtd);
792
793 result = copyReplicationScope(peerHtdCopy, localHtd);
794
795
796 if (result) {
797 result = (peerHtdCopy.compareTo(localHtd) == 0);
798 }
799
800 return result;
801 }
802 }