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 java.io.IOException;
23  import java.util.List;
24  import java.util.Set;
25  
26  import org.apache.commons.logging.Log;
27  import org.apache.commons.logging.LogFactory;
28  import org.apache.hadoop.conf.Configuration;
29  import org.apache.hadoop.fs.FileStatus;
30  import org.apache.hadoop.hbase.Abortable;
31  import org.apache.hadoop.hbase.HConstants;
32  import org.apache.hadoop.hbase.client.HConnectionManager;
33  import org.apache.hadoop.hbase.master.cleaner.BaseLogCleanerDelegate;
34  import org.apache.hadoop.hbase.replication.ReplicationZookeeper;
35  import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
36  import org.apache.zookeeper.KeeperException;
37  
38  import com.google.common.base.Predicate;
39  import com.google.common.collect.ImmutableList;
40  import com.google.common.collect.ImmutableSet;
41  import com.google.common.collect.Iterables;
42  import com.google.common.collect.Sets;
43  
44  /**
45   * Implementation of a log cleaner that checks if a log is still scheduled for
46   * replication before deleting it when its TTL is over.
47   */
48  public class ReplicationLogCleaner extends BaseLogCleanerDelegate implements Abortable {
49    private static final Log LOG = LogFactory.getLog(ReplicationLogCleaner.class);
50    private ReplicationZookeeper zkHelper;
51    private boolean stopped = false;
52    private boolean aborted;
53  
54    /**
55     * Instantiates the cleaner, does nothing more.
56     */
57    public ReplicationLogCleaner() {}
58  
59    @Override
60    public Iterable<FileStatus> getDeletableFiles(Iterable<FileStatus> files) {
61      try {
62        if (!zkHelper.getReplication()) {
63          return ImmutableList.of();
64        }
65      } catch (KeeperException e) {
66        abort("Cannot get the state of replication", e);
67        return ImmutableList.of();
68      }
69  
70     // all members of this class are null if replication is disabled, 
71     // so we cannot filter the files
72      if (this.getConf() == null) {
73        return files;
74      }
75      
76      final Set<String> hlogs = loadHLogsFromQueues();
77      return Iterables.filter(files, new Predicate<FileStatus>() {
78        @Override
79        public boolean apply(FileStatus file) {
80          String hlog = file.getPath().getName();
81          boolean logInReplicationQueue = hlogs.contains(hlog);
82          if (LOG.isDebugEnabled()) {
83            if (logInReplicationQueue) {
84              LOG.debug("Found log in ZK, keeping: " + hlog);
85            } else {
86              LOG.debug("Didn't find this log in ZK, deleting: " + hlog);
87            }
88          }
89         return !logInReplicationQueue;
90        }});
91    }
92  
93    /**
94     * Load all hlogs in all replication queues from ZK
95     */
96    private Set<String> loadHLogsFromQueues() {
97      List<String> rss = zkHelper.getListOfReplicators();
98      if (rss == null) {
99        LOG.debug("Didn't find any region server that replicates, won't prevent any deletions.");
100       return ImmutableSet.of();
101     }
102     Set<String> hlogs = Sets.newHashSet();
103     for (String rs: rss) {
104       List<String> listOfPeers = zkHelper.getListPeersForRS(rs);
105       // if rs just died, this will be null
106       if (listOfPeers == null) {
107         continue;
108       }
109       for (String id : listOfPeers) {
110         List<String> peersHlogs = zkHelper.getListHLogsForPeerForRS(rs, id);
111         if (peersHlogs != null) {
112           hlogs.addAll(peersHlogs);
113         }
114       }
115     }
116     return hlogs;
117   }
118 
119   @Override
120   public void setConf(Configuration config) {
121     // If replication is disabled, keep all members null
122     if (!config.getBoolean(HConstants.REPLICATION_ENABLE_KEY, false)) {
123       LOG.warn("Not configured - allowing all hlogs to be deleted");
124       return;
125     }
126     // Make my own Configuration.  Then I'll have my own connection to zk that
127     // I can close myself when comes time.
128     Configuration conf = new Configuration(config);
129     super.setConf(conf);
130     try {
131       ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "replicationLogCleaner", null);
132       this.zkHelper = new ReplicationZookeeper(this, conf, zkw);
133     } catch (KeeperException e) {
134       LOG.error("Error while configuring " + this.getClass().getName(), e);
135     } catch (IOException e) {
136       LOG.error("Error while configuring " + this.getClass().getName(), e);
137     }
138   }
139 
140 
141   @Override
142   public void stop(String why) {
143     if (this.stopped) return;
144     this.stopped = true;
145     if (this.zkHelper != null) {
146       LOG.info("Stopping " + this.zkHelper.getZookeeperWatcher());
147       this.zkHelper.getZookeeperWatcher().close();
148     }
149   }
150 
151   @Override
152   public boolean isStopped() {
153     return this.stopped;
154   }
155 
156   @Override
157   public void abort(String why, Throwable e) {
158     LOG.warn("Aborting ReplicationLogCleaner because " + why, e);
159     this.aborted = true;
160     stop(why);
161   }
162 
163   @Override
164   public boolean isAborted() {
165     return this.aborted;
166   }
167 }