View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
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   * A {@link BaseReplicationEndpoint} for replication endpoints whose
43   * target cluster is an HBase cluster.
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     * A private method used to re-establish a zookeeper session with a peer cluster.
64     * @param ke
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    // Synchronize peer cluster connection attempts to avoid races and rate
97    // limit connections when multiple replication sources try to connect to
98    // the peer cluster. If the peer cluster is down we can get out of control
99    // over time.
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    * Get the ZK connection to this peer
112    * @return zk connection
113    */
114   protected synchronized ZooKeeperWatcher getZkw() {
115     return zkw;
116   }
117 
118   /**
119    * Closes the current ZKW (if not null) and creates a new one
120    * @throws IOException If anything goes wrong connecting
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     // Currently this is never "Aborted", we just log when the abort method is called.
138     return false;
139   }
140 
141   /**
142    * Get the list of all the region servers from the specified peer
143    * @param zkw zk connection to use
144    * @return list of region server addresses or an empty list if the slave is unavailable
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    * Get a list of all the addresses of all the region servers
161    * for this peer cluster
162    * @return list of addresses
163    */
164   // Synchronize peer cluster connection attempts to avoid races and rate
165   // limit connections when multiple replication sources try to connect to
166   // the peer cluster. If the peer cluster is down we can get out of control
167   // over time.
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    * Set the list of region servers for that peer
182    * @param regionServers list of addresses for the region servers
183    */
184   public synchronized void setRegionServers(List<ServerName> regionServers) {
185     this.regionServers = regionServers;
186     lastRegionServerUpdate = System.currentTimeMillis();
187   }
188 
189   /**
190    * Get the timestamp at which the last change occurred to the list of region servers to replicate
191    * to.
192    * @return The System.currentTimeMillis at the last time the list of peer region servers changed.
193    */
194   public long getLastRegionServerUpdate() {
195     return lastRegionServerUpdate;
196   }
197 
198   /**
199    * Tracks changes to the list of region servers in a peer's cluster.
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 }