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.ClusterConnection; 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.master.ReplicationHFileCleaner; 054import org.apache.hadoop.hbase.testclassification.MasterTests; 055import org.apache.hadoop.hbase.testclassification.SmallTests; 056import org.apache.hadoop.hbase.util.Pair; 057import org.apache.hadoop.hbase.zookeeper.MetaTableLocator; 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 rq.addPeerToHFileRefs(peerId); 114 } 115 116 @After 117 public void cleanup() throws ReplicationException { 118 try { 119 fs.delete(root, true); 120 } catch (IOException e) { 121 LOG.warn("Failed to delete files recursively from path " + root); 122 } 123 rp.getPeerStorage().removePeer(peerId); 124 } 125 126 @Test 127 public void testIsFileDeletable() throws IOException, ReplicationException { 128 // 1. Create a file 129 Path file = new Path(root, "testIsFileDeletableWithNoHFileRefs"); 130 fs.createNewFile(file); 131 // 2. Assert file is successfully created 132 assertTrue("Test file not created!", fs.exists(file)); 133 ReplicationHFileCleaner cleaner = new ReplicationHFileCleaner(); 134 cleaner.setConf(conf); 135 // 3. Assert that file as is should be deletable 136 assertTrue("Cleaner should allow to delete this file as there is no hfile reference node " 137 + "for it in the queue.", 138 cleaner.isFileDeletable(fs.getFileStatus(file))); 139 140 List<Pair<Path, Path>> files = new ArrayList<>(1); 141 files.add(new Pair<>(null, file)); 142 // 4. Add the file to hfile-refs queue 143 rq.addHFileRefs(peerId, files); 144 // 5. Assert file should not be deletable 145 assertFalse("Cleaner should not allow to delete this file as there is a hfile reference node " 146 + "for it in the queue.", 147 cleaner.isFileDeletable(fs.getFileStatus(file))); 148 } 149 150 @Test 151 public void testGetDeletableFiles() throws Exception { 152 // 1. Create two files and assert that they do not exist 153 Path notDeletablefile = new Path(root, "testGetDeletableFiles_1"); 154 fs.createNewFile(notDeletablefile); 155 assertTrue("Test file not created!", fs.exists(notDeletablefile)); 156 Path deletablefile = new Path(root, "testGetDeletableFiles_2"); 157 fs.createNewFile(deletablefile); 158 assertTrue("Test file not created!", fs.exists(deletablefile)); 159 160 List<FileStatus> files = new ArrayList<>(2); 161 FileStatus f = new FileStatus(); 162 f.setPath(deletablefile); 163 files.add(f); 164 f = new FileStatus(); 165 f.setPath(notDeletablefile); 166 files.add(f); 167 168 List<Pair<Path, Path>> hfiles = new ArrayList<>(1); 169 hfiles.add(new Pair<>(null, notDeletablefile)); 170 // 2. Add one file to hfile-refs queue 171 rq.addHFileRefs(peerId, hfiles); 172 173 ReplicationHFileCleaner cleaner = new ReplicationHFileCleaner(); 174 cleaner.setConf(conf); 175 Iterator<FileStatus> deletableFilesIterator = cleaner.getDeletableFiles(files).iterator(); 176 int i = 0; 177 while (deletableFilesIterator.hasNext() && i < 2) { 178 i++; 179 } 180 // 5. Assert one file should not be deletable and it is present in the list returned 181 if (i > 2) { 182 fail("File " + notDeletablefile 183 + " should not be deletable as its hfile reference node is not added."); 184 } 185 assertTrue(deletableFilesIterator.next().getPath().equals(deletablefile)); 186 } 187 188 /** 189 * ReplicationHFileCleaner should be able to ride over ZooKeeper errors without aborting. 190 */ 191 @Test 192 public void testZooKeeperAbort() throws Exception { 193 ReplicationHFileCleaner cleaner = new ReplicationHFileCleaner(); 194 195 List<FileStatus> dummyFiles = 196 Lists.newArrayList(new FileStatus(100, false, 3, 100, System.currentTimeMillis(), new Path( 197 "hfile1")), new FileStatus(100, false, 3, 100, System.currentTimeMillis(), new Path( 198 "hfile2"))); 199 200 FaultyZooKeeperWatcher faultyZK = 201 new FaultyZooKeeperWatcher(conf, "testZooKeeperAbort-faulty", null); 202 try { 203 faultyZK.init(); 204 cleaner.setConf(conf, faultyZK); 205 // should keep all files due to a ConnectionLossException getting the queues znodes 206 Iterable<FileStatus> toDelete = cleaner.getDeletableFiles(dummyFiles); 207 assertFalse(toDelete.iterator().hasNext()); 208 assertFalse(cleaner.isStopped()); 209 } finally { 210 faultyZK.close(); 211 } 212 213 // when zk is working both files should be returned 214 cleaner = new ReplicationHFileCleaner(); 215 ZKWatcher zkw = new ZKWatcher(conf, "testZooKeeperAbort-normal", null); 216 try { 217 cleaner.setConf(conf, zkw); 218 Iterable<FileStatus> filesToDelete = cleaner.getDeletableFiles(dummyFiles); 219 Iterator<FileStatus> iter = filesToDelete.iterator(); 220 assertTrue(iter.hasNext()); 221 assertEquals(new Path("hfile1"), iter.next().getPath()); 222 assertTrue(iter.hasNext()); 223 assertEquals(new Path("hfile2"), iter.next().getPath()); 224 assertFalse(iter.hasNext()); 225 } finally { 226 zkw.close(); 227 } 228 } 229 230 static class DummyServer implements Server { 231 232 @Override 233 public Configuration getConfiguration() { 234 return TEST_UTIL.getConfiguration(); 235 } 236 237 @Override 238 public ZKWatcher getZooKeeper() { 239 try { 240 return new ZKWatcher(getConfiguration(), "dummy server", this); 241 } catch (IOException e) { 242 e.printStackTrace(); 243 } 244 return null; 245 } 246 247 @Override 248 public CoordinatedStateManager getCoordinatedStateManager() { 249 return null; 250 } 251 252 @Override 253 public ClusterConnection getConnection() { 254 return null; 255 } 256 257 @Override 258 public MetaTableLocator getMetaTableLocator() { 259 return null; 260 } 261 262 @Override 263 public ServerName getServerName() { 264 return ServerName.valueOf("regionserver,60020,000000"); 265 } 266 267 @Override 268 public void abort(String why, Throwable e) { 269 } 270 271 @Override 272 public boolean isAborted() { 273 return false; 274 } 275 276 @Override 277 public void stop(String why) { 278 } 279 280 @Override 281 public boolean isStopped() { 282 return false; 283 } 284 285 @Override 286 public ChoreService getChoreService() { 287 return null; 288 } 289 290 @Override 291 public ClusterConnection getClusterConnection() { 292 // TODO Auto-generated method stub 293 return null; 294 } 295 296 @Override 297 public FileSystem getFileSystem() { 298 return null; 299 } 300 301 @Override 302 public boolean isStopping() { 303 return false; 304 } 305 306 @Override 307 public Connection createConnection(Configuration conf) throws IOException { 308 return null; 309 } 310 } 311 312 static class FaultyZooKeeperWatcher extends ZKWatcher { 313 private RecoverableZooKeeper zk; 314 public FaultyZooKeeperWatcher(Configuration conf, String identifier, Abortable abortable) 315 throws ZooKeeperConnectionException, IOException { 316 super(conf, identifier, abortable); 317 } 318 319 public void init() throws Exception { 320 this.zk = spy(super.getRecoverableZooKeeper()); 321 doThrow(new KeeperException.ConnectionLossException()) 322 .when(zk).getData("/hbase/replication/hfile-refs", null, new Stat()); 323 } 324 325 @Override 326 public RecoverableZooKeeper getRecoverableZooKeeper() { 327 return zk; 328 } 329 } 330}