001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.master.cleaner;
019
020import static org.junit.Assert.assertEquals;
021import static org.junit.Assert.assertFalse;
022import static org.junit.Assert.assertTrue;
023import static org.junit.Assert.fail;
024import static org.mockito.Mockito.doThrow;
025import static org.mockito.Mockito.spy;
026
027import java.io.IOException;
028import java.util.ArrayList;
029import java.util.Iterator;
030import java.util.List;
031import org.apache.hadoop.conf.Configuration;
032import org.apache.hadoop.fs.FileStatus;
033import org.apache.hadoop.fs.FileSystem;
034import org.apache.hadoop.fs.Path;
035import org.apache.hadoop.hbase.Abortable;
036import org.apache.hadoop.hbase.ChoreService;
037import org.apache.hadoop.hbase.CoordinatedStateManager;
038import org.apache.hadoop.hbase.HBaseClassTestRule;
039import org.apache.hadoop.hbase.HBaseTestingUtility;
040import org.apache.hadoop.hbase.HConstants;
041import org.apache.hadoop.hbase.Server;
042import org.apache.hadoop.hbase.ServerName;
043import org.apache.hadoop.hbase.ZooKeeperConnectionException;
044import org.apache.hadoop.hbase.client.AsyncClusterConnection;
045import org.apache.hadoop.hbase.client.Connection;
046import org.apache.hadoop.hbase.master.HMaster;
047import org.apache.hadoop.hbase.replication.ReplicationException;
048import org.apache.hadoop.hbase.replication.ReplicationFactory;
049import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
050import org.apache.hadoop.hbase.replication.ReplicationPeers;
051import org.apache.hadoop.hbase.replication.ReplicationQueueStorage;
052import org.apache.hadoop.hbase.replication.ReplicationStorageFactory;
053import org.apache.hadoop.hbase.replication.SyncReplicationState;
054import org.apache.hadoop.hbase.replication.master.ReplicationHFileCleaner;
055import org.apache.hadoop.hbase.testclassification.MasterTests;
056import org.apache.hadoop.hbase.testclassification.SmallTests;
057import org.apache.hadoop.hbase.util.Pair;
058import org.apache.hadoop.hbase.zookeeper.RecoverableZooKeeper;
059import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
060import org.apache.zookeeper.KeeperException;
061import org.apache.zookeeper.data.Stat;
062import org.junit.After;
063import org.junit.AfterClass;
064import org.junit.Before;
065import org.junit.BeforeClass;
066import org.junit.ClassRule;
067import org.junit.Test;
068import org.junit.experimental.categories.Category;
069import org.slf4j.Logger;
070import org.slf4j.LoggerFactory;
071
072import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
073
074@Category({ MasterTests.class, SmallTests.class })
075public class TestReplicationHFileCleaner {
076
077  @ClassRule
078  public static final HBaseClassTestRule CLASS_RULE =
079      HBaseClassTestRule.forClass(TestReplicationHFileCleaner.class);
080
081  private static final Logger LOG = LoggerFactory.getLogger(TestReplicationHFileCleaner.class);
082  private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
083  private static Server server;
084  private static ReplicationQueueStorage rq;
085  private static ReplicationPeers rp;
086  private static final String peerId = "TestReplicationHFileCleaner";
087  private static Configuration conf = TEST_UTIL.getConfiguration();
088  static FileSystem fs = null;
089  Path root;
090
091  @BeforeClass
092  public static void setUpBeforeClass() throws Exception {
093    TEST_UTIL.startMiniZKCluster();
094    server = new DummyServer();
095    conf.setBoolean(HConstants.REPLICATION_BULKLOAD_ENABLE_KEY, true);
096    HMaster.decorateMasterConfiguration(conf);
097    rp = ReplicationFactory.getReplicationPeers(server.getZooKeeper(), conf);
098    rp.init();
099    rq = ReplicationStorageFactory.getReplicationQueueStorage(server.getZooKeeper(), conf);
100    fs = FileSystem.get(conf);
101  }
102
103  @AfterClass
104  public static void tearDownAfterClass() throws Exception {
105    TEST_UTIL.shutdownMiniZKCluster();
106  }
107
108  @Before
109  public void setup() throws ReplicationException, IOException {
110    root = TEST_UTIL.getDataTestDirOnTestFS();
111    rp.getPeerStorage().addPeer(peerId,
112      ReplicationPeerConfig.newBuilder().setClusterKey(TEST_UTIL.getClusterKey()).build(), true,
113      SyncReplicationState.NONE);
114    rq.addPeerToHFileRefs(peerId);
115  }
116
117  @After
118  public void cleanup() throws ReplicationException {
119    try {
120      fs.delete(root, true);
121    } catch (IOException e) {
122      LOG.warn("Failed to delete files recursively from path " + root);
123    }
124    rp.getPeerStorage().removePeer(peerId);
125  }
126
127  @Test
128  public void testIsFileDeletable() throws IOException, ReplicationException {
129    // 1. Create a file
130    Path file = new Path(root, "testIsFileDeletableWithNoHFileRefs");
131    fs.createNewFile(file);
132    // 2. Assert file is successfully created
133    assertTrue("Test file not created!", fs.exists(file));
134    ReplicationHFileCleaner cleaner = new ReplicationHFileCleaner();
135    cleaner.setConf(conf);
136    // 3. Assert that file as is should be deletable
137    assertTrue("Cleaner should allow to delete this file as there is no hfile reference node "
138        + "for it in the queue.",
139      cleaner.isFileDeletable(fs.getFileStatus(file)));
140
141    List<Pair<Path, Path>> files = new ArrayList<>(1);
142    files.add(new Pair<>(null, file));
143    // 4. Add the file to hfile-refs queue
144    rq.addHFileRefs(peerId, files);
145    // 5. Assert file should not be deletable
146    assertFalse("Cleaner should not allow to delete this file as there is a hfile reference node "
147        + "for it in the queue.",
148      cleaner.isFileDeletable(fs.getFileStatus(file)));
149  }
150
151  @Test
152  public void testGetDeletableFiles() throws Exception {
153    // 1. Create two files and assert that they do not exist
154    Path notDeletablefile = new Path(root, "testGetDeletableFiles_1");
155    fs.createNewFile(notDeletablefile);
156    assertTrue("Test file not created!", fs.exists(notDeletablefile));
157    Path deletablefile = new Path(root, "testGetDeletableFiles_2");
158    fs.createNewFile(deletablefile);
159    assertTrue("Test file not created!", fs.exists(deletablefile));
160
161    List<FileStatus> files = new ArrayList<>(2);
162    FileStatus f = new FileStatus();
163    f.setPath(deletablefile);
164    files.add(f);
165    f = new FileStatus();
166    f.setPath(notDeletablefile);
167    files.add(f);
168
169    List<Pair<Path, Path>> hfiles = new ArrayList<>(1);
170    hfiles.add(new Pair<>(null, notDeletablefile));
171    // 2. Add one file to hfile-refs queue
172    rq.addHFileRefs(peerId, hfiles);
173
174    ReplicationHFileCleaner cleaner = new ReplicationHFileCleaner();
175    cleaner.setConf(conf);
176    Iterator<FileStatus> deletableFilesIterator = cleaner.getDeletableFiles(files).iterator();
177    int i = 0;
178    while (deletableFilesIterator.hasNext() && i < 2) {
179      i++;
180    }
181    // 5. Assert one file should not be deletable and it is present in the list returned
182    if (i > 2) {
183      fail("File " + notDeletablefile
184          + " should not be deletable as its hfile reference node is not added.");
185    }
186    assertTrue(deletableFilesIterator.next().getPath().equals(deletablefile));
187  }
188
189  /**
190   * ReplicationHFileCleaner should be able to ride over ZooKeeper errors without aborting.
191   */
192  @Test
193  public void testZooKeeperAbort() throws Exception {
194    ReplicationHFileCleaner cleaner = new ReplicationHFileCleaner();
195
196    List<FileStatus> dummyFiles =
197        Lists.newArrayList(new FileStatus(100, false, 3, 100, System.currentTimeMillis(), new Path(
198            "hfile1")), new FileStatus(100, false, 3, 100, System.currentTimeMillis(), new Path(
199            "hfile2")));
200
201    FaultyZooKeeperWatcher faultyZK =
202        new FaultyZooKeeperWatcher(conf, "testZooKeeperAbort-faulty", null);
203    try {
204      faultyZK.init();
205      cleaner.setConf(conf, faultyZK);
206      // should keep all files due to a ConnectionLossException getting the queues znodes
207      Iterable<FileStatus> toDelete = cleaner.getDeletableFiles(dummyFiles);
208      assertFalse(toDelete.iterator().hasNext());
209      assertFalse(cleaner.isStopped());
210    } finally {
211      faultyZK.close();
212    }
213
214    // when zk is working both files should be returned
215    cleaner = new ReplicationHFileCleaner();
216    ZKWatcher zkw = new ZKWatcher(conf, "testZooKeeperAbort-normal", null);
217    try {
218      cleaner.setConf(conf, zkw);
219      Iterable<FileStatus> filesToDelete = cleaner.getDeletableFiles(dummyFiles);
220      Iterator<FileStatus> iter = filesToDelete.iterator();
221      assertTrue(iter.hasNext());
222      assertEquals(new Path("hfile1"), iter.next().getPath());
223      assertTrue(iter.hasNext());
224      assertEquals(new Path("hfile2"), iter.next().getPath());
225      assertFalse(iter.hasNext());
226    } finally {
227      zkw.close();
228    }
229  }
230
231  static class DummyServer implements Server {
232
233    @Override
234    public Configuration getConfiguration() {
235      return TEST_UTIL.getConfiguration();
236    }
237
238    @Override
239    public ZKWatcher getZooKeeper() {
240      try {
241        return new ZKWatcher(getConfiguration(), "dummy server", this);
242      } catch (IOException e) {
243        e.printStackTrace();
244      }
245      return null;
246    }
247
248    @Override
249    public CoordinatedStateManager getCoordinatedStateManager() {
250      return null;
251    }
252
253    @Override
254    public Connection getConnection() {
255      return null;
256    }
257
258    @Override
259    public ServerName getServerName() {
260      return ServerName.valueOf("regionserver,60020,000000");
261    }
262
263    @Override
264    public void abort(String why, Throwable e) {
265    }
266
267    @Override
268    public boolean isAborted() {
269      return false;
270    }
271
272    @Override
273    public void stop(String why) {
274    }
275
276    @Override
277    public boolean isStopped() {
278      return false;
279    }
280
281    @Override
282    public ChoreService getChoreService() {
283      return null;
284    }
285
286    @Override
287    public FileSystem getFileSystem() {
288      return null;
289    }
290
291    @Override
292    public boolean isStopping() {
293      return false;
294    }
295
296    @Override
297    public Connection createConnection(Configuration conf) throws IOException {
298      return null;
299    }
300
301    @Override
302    public AsyncClusterConnection getAsyncClusterConnection() {
303      return null;
304    }
305  }
306
307  static class FaultyZooKeeperWatcher extends ZKWatcher {
308    private RecoverableZooKeeper zk;
309    public FaultyZooKeeperWatcher(Configuration conf, String identifier, Abortable abortable)
310        throws ZooKeeperConnectionException, IOException {
311      super(conf, identifier, abortable);
312    }
313
314    public void init() throws Exception {
315      this.zk = spy(super.getRecoverableZooKeeper());
316      doThrow(new KeeperException.ConnectionLossException())
317          .when(zk).getData("/hbase/replication/hfile-refs", null, new Stat());
318    }
319
320    @Override
321    public RecoverableZooKeeper getRecoverableZooKeeper() {
322      return zk;
323    }
324  }
325}