1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.replication;
20
21 import java.io.IOException;
22 import java.util.ArrayList;
23 import java.util.Collections;
24 import java.util.List;
25 import java.util.UUID;
26
27 import org.apache.commons.logging.Log;
28 import org.apache.commons.logging.LogFactory;
29 import org.apache.hadoop.hbase.classification.InterfaceAudience;
30 import org.apache.hadoop.hbase.Abortable;
31 import org.apache.hadoop.hbase.ServerName;
32 import org.apache.hadoop.hbase.zookeeper.ZKClusterId;
33 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
34 import org.apache.hadoop.hbase.zookeeper.ZooKeeperListener;
35 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
36 import org.apache.zookeeper.KeeperException;
37 import org.apache.zookeeper.KeeperException.AuthFailedException;
38 import org.apache.zookeeper.KeeperException.ConnectionLossException;
39 import org.apache.zookeeper.KeeperException.SessionExpiredException;
40
41
42
43
44
45 @InterfaceAudience.Private
46 public abstract class HBaseReplicationEndpoint extends BaseReplicationEndpoint
47 implements Abortable {
48
49 private static final Log LOG = LogFactory.getLog(HBaseReplicationEndpoint.class);
50
51 private ZooKeeperWatcher zkw = null;
52
53 private List<ServerName> regionServers = new ArrayList<ServerName>(0);
54 private long lastRegionServerUpdate;
55
56 protected synchronized void disconnect() {
57 if (zkw != null) {
58 zkw.close();
59 }
60 }
61
62
63
64
65
66 protected void reconnect(KeeperException ke) {
67 if (ke instanceof ConnectionLossException || ke instanceof SessionExpiredException
68 || ke instanceof AuthFailedException) {
69 String clusterKey = ctx.getPeerConfig().getClusterKey();
70 LOG.warn("Lost the ZooKeeper connection for peer " + clusterKey, ke);
71 try {
72 reloadZkWatcher();
73 } catch (IOException io) {
74 LOG.warn("Creation of ZookeeperWatcher failed for peer " + clusterKey, io);
75 }
76 }
77 }
78
79 @Override
80 protected void doStart() {
81 try {
82 reloadZkWatcher();
83 notifyStarted();
84 } catch (IOException e) {
85 notifyFailed(e);
86 }
87 }
88
89 @Override
90 protected void doStop() {
91 disconnect();
92 notifyStopped();
93 }
94
95 @Override
96
97
98
99
100 public synchronized UUID getPeerUUID() {
101 UUID peerUUID = null;
102 try {
103 peerUUID = ZKClusterId.getUUIDForCluster(zkw);
104 } catch (KeeperException ke) {
105 reconnect(ke);
106 }
107 return peerUUID;
108 }
109
110
111
112
113
114 protected synchronized ZooKeeperWatcher getZkw() {
115 return zkw;
116 }
117
118
119
120
121
122 synchronized void reloadZkWatcher() throws IOException {
123 if (zkw != null) zkw.close();
124 zkw = new ZooKeeperWatcher(ctx.getConfiguration(),
125 "connection to cluster: " + ctx.getPeerId(), this);
126 getZkw().registerListener(new PeerRegionServerListener(this));
127 }
128
129 @Override
130 public void abort(String why, Throwable e) {
131 LOG.error("The HBaseReplicationEndpoint corresponding to peer " + ctx.getPeerId()
132 + " was aborted for the following reason(s):" + why, e);
133 }
134
135 @Override
136 public boolean isAborted() {
137
138 return false;
139 }
140
141
142
143
144
145
146 protected static List<ServerName> fetchSlavesAddresses(ZooKeeperWatcher zkw)
147 throws KeeperException {
148 List<String> children = ZKUtil.listChildrenAndWatchForNewChildren(zkw, zkw.rsZNode);
149 if (children == null) {
150 return Collections.emptyList();
151 }
152 List<ServerName> addresses = new ArrayList<ServerName>(children.size());
153 for (String child : children) {
154 addresses.add(ServerName.parseServerName(child));
155 }
156 return addresses;
157 }
158
159
160
161
162
163
164
165
166
167
168 public synchronized List<ServerName> getRegionServers() {
169 try {
170 setRegionServers(fetchSlavesAddresses(this.getZkw()));
171 } catch (KeeperException ke) {
172 if (LOG.isDebugEnabled()) {
173 LOG.debug("Fetch slaves addresses failed", ke);
174 }
175 reconnect(ke);
176 }
177 return regionServers;
178 }
179
180
181
182
183
184 public synchronized void setRegionServers(List<ServerName> regionServers) {
185 this.regionServers = regionServers;
186 lastRegionServerUpdate = System.currentTimeMillis();
187 }
188
189
190
191
192
193
194 public long getLastRegionServerUpdate() {
195 return lastRegionServerUpdate;
196 }
197
198
199
200
201 public static class PeerRegionServerListener extends ZooKeeperListener {
202
203 private final HBaseReplicationEndpoint replicationEndpoint;
204 private final String regionServerListNode;
205
206 public PeerRegionServerListener(HBaseReplicationEndpoint replicationPeer) {
207 super(replicationPeer.getZkw());
208 this.replicationEndpoint = replicationPeer;
209 this.regionServerListNode = replicationEndpoint.getZkw().rsZNode;
210 }
211
212 @Override
213 public synchronized void nodeChildrenChanged(String path) {
214 if (path.equals(regionServerListNode)) {
215 try {
216 LOG.info("Detected change to peer region servers, fetching updated list");
217 replicationEndpoint.setRegionServers(fetchSlavesAddresses(replicationEndpoint.getZkw()));
218 } catch (KeeperException e) {
219 LOG.error("Error reading slave addresses", e);
220 }
221 }
222 }
223 }
224 }