View Javadoc

1   /*
2    * Copyright 2010 The Apache Software Foundation
3    *
4    * Licensed to the Apache Software Foundation (ASF) under one
5    * or more contributor license agreements.  See the NOTICE file
6    * distributed with this work for additional information
7    * regarding copyright ownership.  The ASF licenses this file
8    * to you under the Apache License, Version 2.0 (the
9    * "License"); you may not use this file except in compliance
10   * with the License.  You may obtain a copy of the License at
11   *
12   *     http://www.apache.org/licenses/LICENSE-2.0
13   *
14   * Unless required by applicable law or agreed to in writing, software
15   * distributed under the License is distributed on an "AS IS" BASIS,
16   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17   * See the License for the specific language governing permissions and
18   * limitations under the License.
19   */
20  package org.apache.hadoop.hbase.replication.master;
21  
22  import org.apache.commons.logging.Log;
23  import org.apache.commons.logging.LogFactory;
24  import org.apache.hadoop.conf.Configuration;
25  import org.apache.hadoop.fs.Path;
26  import org.apache.hadoop.hbase.Abortable;
27  import org.apache.hadoop.hbase.HConstants;
28  import org.apache.hadoop.hbase.client.HConnectionManager;
29  import org.apache.hadoop.hbase.master.LogCleanerDelegate;
30  import org.apache.hadoop.hbase.replication.ReplicationZookeeper;
31  import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
32  import org.apache.zookeeper.KeeperException;
33  
34  import java.io.IOException;
35  import java.util.HashSet;
36  import java.util.List;
37  import java.util.Set;
38  
39  /**
40   * Implementation of a log cleaner that checks if a log is still scheduled for
41   * replication before deleting it when its TTL is over.
42   */
43  public class ReplicationLogCleaner implements LogCleanerDelegate, Abortable {
44    private static final Log LOG = LogFactory.getLog(ReplicationLogCleaner.class);
45    private Configuration conf;
46    private ReplicationZookeeper zkHelper;
47    private Set<String> hlogs = new HashSet<String>();
48    private boolean stopped = false;
49  
50    /**
51     * Instantiates the cleaner, does nothing more.
52     */
53    public ReplicationLogCleaner() {}
54  
55    @Override
56    public boolean isLogDeletable(Path filePath) {
57  
58      try {
59        if (!zkHelper.getReplication()) {
60          return false;
61        }
62      } catch (KeeperException e) {
63        abort("Cannot get the state of replication", e);
64        return false;
65      }
66  
67      // all members of this class are null if replication is disabled, and we
68      // return true since false would render the LogsCleaner useless
69      if (this.conf == null) {
70        return true;
71      }
72      String log = filePath.getName();
73      // If we saw the hlog previously, let's consider it's still used
74      // At some point in the future we will refresh the list and it will be gone
75      if (this.hlogs.contains(log)) {
76        return false;
77      }
78  
79      // Let's see it's still there
80      // This solution makes every miss very expensive to process since we
81      // almost completely refresh the cache each time
82      return !refreshHLogsAndSearch(log);
83    }
84  
85    /**
86     * Search through all the hlogs we have in ZK to refresh the cache
87     * If a log is specified and found, then we early out and return true
88     * @param searchedLog log we are searching for, pass null to cache everything
89     *                    that's in zookeeper.
90     * @return false until a specified log is found.
91     */
92    private boolean refreshHLogsAndSearch(String searchedLog) {
93      this.hlogs.clear();
94      final boolean lookForLog = searchedLog != null;
95      List<String> rss = zkHelper.getListOfReplicators();
96      if (rss == null) {
97        LOG.debug("Didn't find any region server that replicates, deleting: " +
98            searchedLog);
99        return false;
100     }
101     for (String rs: rss) {
102       List<String> listOfPeers = zkHelper.getListPeersForRS(rs);
103       // if rs just died, this will be null
104       if (listOfPeers == null) {
105         continue;
106       }
107       for (String id : listOfPeers) {
108         List<String> peersHlogs = zkHelper.getListHLogsForPeerForRS(rs, id);
109         if (peersHlogs != null) {
110           this.hlogs.addAll(peersHlogs);
111         }
112         // early exit if we found the log
113         if(lookForLog && this.hlogs.contains(searchedLog)) {
114           LOG.debug("Found log in ZK, keeping: " + searchedLog);
115           return true;
116         }
117       }
118     }
119     LOG.debug("Didn't find this log in ZK, deleting: " + searchedLog);
120     return false;
121   }
122 
123   @Override
124   public void setConf(Configuration conf) {
125     // If replication is disabled, keep all members null
126     if (!conf.getBoolean(HConstants.REPLICATION_ENABLE_KEY, false)) {
127       return;
128     }
129     // Make my own Configuration.  Then I'll have my own connection to zk that
130     // I can close myself when comes time.
131     this.conf = new Configuration(conf);
132     try {
133       ZooKeeperWatcher zkw =
134           new ZooKeeperWatcher(this.conf, "replicationLogCleaner", null);
135       this.zkHelper = new ReplicationZookeeper(this, this.conf, zkw);
136     } catch (KeeperException e) {
137       LOG.error("Error while configuring " + this.getClass().getName(), e);
138     } catch (IOException e) {
139       LOG.error("Error while configuring " + this.getClass().getName(), e);
140     }
141     refreshHLogsAndSearch(null);
142   }
143 
144   @Override
145   public Configuration getConf() {
146     return conf;
147   }
148 
149   @Override
150   public void stop(String why) {
151     if (this.stopped) return;
152     this.stopped = true;
153     if (this.zkHelper != null) {
154       LOG.info("Stopping " + this.zkHelper.getZookeeperWatcher());
155       this.zkHelper.getZookeeperWatcher().close();
156     }
157     HConnectionManager.deleteConnection(this.conf, true);
158   }
159 
160   @Override
161   public boolean isStopped() {
162     return this.stopped;
163   }
164 
165   @Override
166   public void abort(String why, Throwable e) {
167     LOG.warn("Aborting ReplicationLogCleaner because " + why, e);
168     stop(why);
169   }
170 }