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.master.procedure;
20  
21  import java.io.IOException;
22  import java.util.concurrent.atomic.AtomicInteger;
23  import java.util.List;
24  import java.util.TreeSet;
25  
26  import org.apache.commons.logging.Log;
27  import org.apache.commons.logging.LogFactory;
28  import org.apache.hadoop.fs.FileSystem;
29  import org.apache.hadoop.fs.Path;
30  import org.apache.hadoop.hbase.HColumnDescriptor;
31  import org.apache.hadoop.hbase.HRegionInfo;
32  import org.apache.hadoop.hbase.HRegionLocation;
33  import org.apache.hadoop.hbase.HTableDescriptor;
34  import org.apache.hadoop.hbase.MetaTableAccessor;
35  import org.apache.hadoop.hbase.RegionLocations;
36  import org.apache.hadoop.hbase.ServerName;
37  import org.apache.hadoop.hbase.TableName;
38  import org.apache.hadoop.hbase.TableStateManager;
39  import org.apache.hadoop.hbase.client.BufferedMutator;
40  import org.apache.hadoop.hbase.client.Connection;
41  import org.apache.hadoop.hbase.client.Durability;
42  import org.apache.hadoop.hbase.client.Put;
43  import org.apache.hadoop.hbase.client.MetaScanner;
44  import org.apache.hadoop.hbase.client.MetaScanner.MetaScannerVisitor;
45  import org.apache.hadoop.hbase.client.MetaScanner.MetaScannerVisitorBase;
46  import org.apache.hadoop.hbase.client.Result;
47  import org.apache.hadoop.hbase.master.HMaster;
48  import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
49  import org.apache.hadoop.hbase.procedure2.ProcedureTestingUtility;
50  import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos;
51  import org.apache.hadoop.hbase.util.ModifyRegionUtils;
52  import org.apache.hadoop.hbase.util.Bytes;
53  import org.apache.hadoop.hbase.util.FSUtils;
54  import org.apache.hadoop.hbase.util.MD5Hash;
55  
56  import static org.junit.Assert.assertEquals;
57  import static org.junit.Assert.assertFalse;
58  import static org.junit.Assert.assertTrue;
59  
60  public class MasterProcedureTestingUtility {
61    private static final Log LOG = LogFactory.getLog(MasterProcedureTestingUtility.class);
62  
63    private MasterProcedureTestingUtility() {
64    }
65  
66    public static HTableDescriptor createHTD(final TableName tableName, final String... family) {
67      HTableDescriptor htd = new HTableDescriptor(tableName);
68      for (int i = 0; i < family.length; ++i) {
69        htd.addFamily(new HColumnDescriptor(family[i]));
70      }
71      return htd;
72    }
73  
74    public static HRegionInfo[] createTable(final ProcedureExecutor<MasterProcedureEnv> procExec,
75        final TableName tableName, final byte[][] splitKeys, String... family) throws IOException {
76      HTableDescriptor htd = createHTD(tableName, family);
77      HRegionInfo[] regions = ModifyRegionUtils.createHRegionInfos(htd, splitKeys);
78      long procId = ProcedureTestingUtility.submitAndWait(procExec,
79        new CreateTableProcedure(procExec.getEnvironment(), htd, regions));
80      ProcedureTestingUtility.assertProcNotFailed(procExec.getResult(procId));
81      return regions;
82    }
83  
84    public static void validateTableCreation(final HMaster master, final TableName tableName,
85        final HRegionInfo[] regions, String... family) throws IOException {
86      validateTableCreation(master, tableName, regions, true, family);
87    }
88  
89    public static void validateTableCreation(final HMaster master, final TableName tableName,
90        final HRegionInfo[] regions, boolean hasFamilyDirs, String... family) throws IOException {
91      // check filesystem
92      final FileSystem fs = master.getMasterFileSystem().getFileSystem();
93      final Path tableDir = FSUtils.getTableDir(master.getMasterFileSystem().getRootDir(), tableName);
94      assertTrue(fs.exists(tableDir));
95      FSUtils.logFileSystemState(fs, tableDir, LOG);
96      List<Path> allRegionDirs = FSUtils.getRegionDirs(fs, tableDir);
97      for (int i = 0; i < regions.length; ++i) {
98        Path regionDir = new Path(tableDir, regions[i].getEncodedName());
99        assertTrue(regions[i] + " region dir does not exist", fs.exists(regionDir));
100       assertTrue(allRegionDirs.remove(regionDir));
101       List<Path> allFamilyDirs = FSUtils.getFamilyDirs(fs, regionDir);
102       for (int j = 0; j < family.length; ++j) {
103         final Path familyDir = new Path(regionDir, family[j]);
104         if (hasFamilyDirs) {
105           assertTrue(family[j] + " family dir does not exist", fs.exists(familyDir));
106           assertTrue(allFamilyDirs.remove(familyDir));
107         } else {
108           // TODO: WARN: Modify Table/Families does not create a family dir
109           if (!fs.exists(familyDir)) {
110             LOG.warn(family[j] + " family dir does not exist");
111           }
112           allFamilyDirs.remove(familyDir);
113         }
114       }
115       assertTrue("found extraneous families: " + allFamilyDirs, allFamilyDirs.isEmpty());
116     }
117     assertTrue("found extraneous regions: " + allRegionDirs, allRegionDirs.isEmpty());
118 
119     // check meta
120     assertTrue(MetaTableAccessor.tableExists(master.getConnection(), tableName));
121     assertEquals(regions.length, countMetaRegions(master, tableName));
122 
123     // check htd
124     HTableDescriptor htd = master.getTableDescriptors().get(tableName);
125     assertTrue("table descriptor not found", htd != null);
126     for (int i = 0; i < family.length; ++i) {
127       assertTrue("family not found " + family[i], htd.getFamily(Bytes.toBytes(family[i])) != null);
128     }
129     assertEquals(family.length, htd.getFamilies().size());
130   }
131 
132   public static void validateTableDeletion(final HMaster master, final TableName tableName,
133       final HRegionInfo[] regions, String... family) throws IOException {
134     // check filesystem
135     final FileSystem fs = master.getMasterFileSystem().getFileSystem();
136     final Path tableDir = FSUtils.getTableDir(master.getMasterFileSystem().getRootDir(), tableName);
137     assertFalse(fs.exists(tableDir));
138 
139     // check meta
140     assertFalse(MetaTableAccessor.tableExists(master.getConnection(), tableName));
141     assertEquals(0, countMetaRegions(master, tableName));
142 
143     // check htd
144     assertTrue("found htd of deleted table",
145       master.getTableDescriptors().get(tableName) == null);
146   }
147 
148   private static int countMetaRegions(final HMaster master, final TableName tableName)
149       throws IOException {
150     final AtomicInteger actualRegCount = new AtomicInteger(0);
151     final MetaScannerVisitor visitor = new MetaScannerVisitorBase() {
152       @Override
153       public boolean processRow(Result rowResult) throws IOException {
154         RegionLocations list = MetaTableAccessor.getRegionLocations(rowResult);
155         if (list == null) {
156           LOG.warn("No serialized HRegionInfo in " + rowResult);
157           return true;
158         }
159         HRegionLocation l = list.getRegionLocation();
160         if (l == null) {
161           return true;
162         }
163         if (!l.getRegionInfo().getTable().equals(tableName)) {
164           return false;
165         }
166         if (l.getRegionInfo().isOffline() || l.getRegionInfo().isSplit()) return true;
167         HRegionLocation[] locations = list.getRegionLocations();
168         for (HRegionLocation location : locations) {
169           if (location == null) continue;
170           ServerName serverName = location.getServerName();
171           // Make sure that regions are assigned to server
172           if (serverName != null && serverName.getHostAndPort() != null) {
173             actualRegCount.incrementAndGet();
174           }
175         }
176         return true;
177       }
178     };
179     MetaScanner.metaScan(master.getConnection(), visitor, tableName);
180     return actualRegCount.get();
181   }
182 
183   public static void validateTableIsEnabled(final HMaster master, final TableName tableName)
184       throws IOException {
185     TableStateManager tsm = master.getAssignmentManager().getTableStateManager();
186     assertTrue(tsm.isTableState(tableName, ZooKeeperProtos.Table.State.ENABLED));
187   }
188 
189   public static void validateTableIsDisabled(final HMaster master, final TableName tableName)
190       throws IOException {
191     TableStateManager tsm = master.getAssignmentManager().getTableStateManager();
192     assertTrue(tsm.isTableState(tableName, ZooKeeperProtos.Table.State.DISABLED));
193   }
194 
195   /**
196    * Run through all procedure flow states TWICE while also restarting procedure executor at each
197    * step; i.e force a reread of procedure store.
198    *
199    *<p>It does
200    * <ol><li>Execute step N - kill the executor before store update
201    * <li>Restart executor/store
202    * <li>Execute step N - and then save to store
203    * </ol>
204    *
205    *<p>This is a good test for finding state that needs persisting and steps that are not
206    * idempotent. Use this version of the test when a procedure executes all flow steps from start to
207    * finish.
208    * @see #testRecoveryAndDoubleExecution(ProcedureExecutor, long)
209    */
210   public static <TState> void testRecoveryAndDoubleExecution(
211       final ProcedureExecutor<MasterProcedureEnv> procExec, final long procId,
212       final int numSteps, final TState[] states) throws Exception {
213     ProcedureTestingUtility.waitProcedure(procExec, procId);
214     assertEquals(false, procExec.isRunning());
215     for (int i = 0; i < numSteps; ++i) {
216       LOG.info("Restart "+ i +" exec state: " + states[i]);
217       ProcedureTestingUtility.assertProcNotYetCompleted(procExec, procId);
218       ProcedureTestingUtility.restart(procExec);
219       ProcedureTestingUtility.waitProcedure(procExec, procId);
220     }
221     assertEquals(true, procExec.isRunning());
222     ProcedureTestingUtility.assertProcNotFailed(procExec, procId);
223   }
224 
225   /**
226    * Run through all procedure flow states TWICE while also restarting procedure executor at each
227    * step; i.e force a reread of procedure store.
228    *
229    *<p>It does
230    * <ol><li>Execute step N - kill the executor before store update
231    * <li>Restart executor/store
232    * <li>Execute step N - and then save to store
233    * </ol>
234    *
235    *<p>This is a good test for finding state that needs persisting and steps that are not
236    * idempotent. Use this version of the test when the order in which flow steps are executed is
237    * not start to finish; where the procedure may vary the flow steps dependent on circumstance
238    * found.
239    * @see #testRecoveryAndDoubleExecution(ProcedureExecutor, long, int, Object[])
240    */
241   public static <TState> void testRecoveryAndDoubleExecution(
242       final ProcedureExecutor<MasterProcedureEnv> procExec, final long procId)
243   throws Exception {
244     ProcedureTestingUtility.waitProcedure(procExec, procId);
245     assertEquals(false, procExec.isRunning());
246     while (!procExec.isFinished(procId)) {
247       ProcedureTestingUtility.restart(procExec);
248       ProcedureTestingUtility.waitProcedure(procExec, procId);
249     }
250     assertEquals(true, procExec.isRunning());
251     ProcedureTestingUtility.assertProcNotFailed(procExec, procId);
252   }
253 
254   public static <TState> void testRollbackAndDoubleExecution(
255       final ProcedureExecutor<MasterProcedureEnv> procExec, final long procId,
256       final int lastStep, final TState[] states) throws Exception {
257     ProcedureTestingUtility.waitProcedure(procExec, procId);
258 
259     // Restart the executor and execute the step twice
260     //   execute step N - kill before store update
261     //   restart executor/store
262     //   execute step N - save on store
263     for (int i = 0; i < lastStep; ++i) {
264       LOG.info("Restart "+ i +" exec state: " + states[i]);
265       ProcedureTestingUtility.assertProcNotYetCompleted(procExec, procId);
266       ProcedureTestingUtility.restart(procExec);
267       ProcedureTestingUtility.waitProcedure(procExec, procId);
268     }
269 
270     // Restart the executor and rollback the step twice
271     //   rollback step N - kill before store update
272     //   restart executor/store
273     //   rollback step N - save on store
274     InjectAbortOnLoadListener abortListener = new InjectAbortOnLoadListener(procExec);
275     procExec.registerListener(abortListener);
276     try {
277       for (int i = lastStep + 1; i >= 0; --i) {
278         LOG.info("Restart " + i +" rollback state: "+ states[i]);
279         ProcedureTestingUtility.assertProcNotYetCompleted(procExec, procId);
280         ProcedureTestingUtility.restart(procExec);
281         ProcedureTestingUtility.waitProcedure(procExec, procId);
282       }
283     } finally {
284       assertTrue(procExec.unregisterListener(abortListener));
285     }
286 
287     ProcedureTestingUtility.assertIsAbortException(procExec.getResult(procId));
288   }
289 
290   public static <TState> void testRollbackAndDoubleExecutionAfterPONR(
291       final ProcedureExecutor<MasterProcedureEnv> procExec, final long procId,
292       final int lastStep, final TState[] states) throws Exception {
293     ProcedureTestingUtility.waitProcedure(procExec, procId);
294 
295     // Restart the executor and execute the step twice
296     //   execute step N - kill before store update
297     //   restart executor/store
298     //   execute step N - save on store
299     for (int i = 0; i < lastStep; ++i) {
300       LOG.info("Restart "+ i +" exec state: " + states[i]);
301       ProcedureTestingUtility.assertProcNotYetCompleted(procExec, procId);
302       ProcedureTestingUtility.restart(procExec);
303       ProcedureTestingUtility.waitProcedure(procExec, procId);
304     }
305 
306     // try to inject the abort
307     ProcedureTestingUtility.setKillAndToggleBeforeStoreUpdate(procExec, false);
308     InjectAbortOnLoadListener abortListener = new InjectAbortOnLoadListener(procExec);
309     procExec.registerListener(abortListener);
310     try {
311       ProcedureTestingUtility.assertProcNotYetCompleted(procExec, procId);
312       ProcedureTestingUtility.restart(procExec);
313       LOG.info("Restart and execute");
314       ProcedureTestingUtility.waitProcedure(procExec, procId);
315     } finally {
316       assertTrue(procExec.unregisterListener(abortListener));
317     }
318 
319     assertEquals(true, procExec.isRunning());
320     ProcedureTestingUtility.assertProcNotFailed(procExec, procId);
321   }
322 
323   public static <TState> void testRollbackRetriableFailure(
324       final ProcedureExecutor<MasterProcedureEnv> procExec, final long procId,
325       final int lastStep, final TState[] states) throws Exception {
326     ProcedureTestingUtility.waitProcedure(procExec, procId);
327 
328     // Restart the executor and execute the step twice
329     //   execute step N - kill before store update
330     //   restart executor/store
331     //   execute step N - save on store
332     for (int i = 0; i < lastStep; ++i) {
333       LOG.info("Restart "+ i +" exec state: " + states[i]);
334       ProcedureTestingUtility.assertProcNotYetCompleted(procExec, procId);
335       ProcedureTestingUtility.restart(procExec);
336       ProcedureTestingUtility.waitProcedure(procExec, procId);
337     }
338 
339     // execute the rollback
340     ProcedureTestingUtility.setKillAndToggleBeforeStoreUpdate(procExec, false);
341     InjectAbortOnLoadListener abortListener = new InjectAbortOnLoadListener(procExec);
342     procExec.registerListener(abortListener);
343     try {
344       ProcedureTestingUtility.assertProcNotYetCompleted(procExec, procId);
345       ProcedureTestingUtility.restart(procExec);
346       LOG.info("Restart and rollback");
347       ProcedureTestingUtility.waitProcedure(procExec, procId);
348     } finally {
349       assertTrue(procExec.unregisterListener(abortListener));
350     }
351 
352     ProcedureTestingUtility.assertIsAbortException(procExec.getResult(procId));
353   }
354 
355   public static void testRestartWithAbort(ProcedureExecutor<MasterProcedureEnv> procExec,
356       long procId) throws Exception {
357     InjectAbortOnLoadListener abortListener = new InjectAbortOnLoadListener(procExec);
358     abortListener.addProcId(procId);
359     procExec.registerListener(abortListener);
360     try {
361       ProcedureTestingUtility.setKillAndToggleBeforeStoreUpdate(procExec, false);
362       ProcedureTestingUtility.restart(procExec);
363       ProcedureTestingUtility.waitProcedure(procExec, procId);
364     } finally {
365       assertTrue(procExec.unregisterListener(abortListener));
366     }
367   }
368 
369   public static void validateColumnFamilyAddition(final HMaster master, final TableName tableName,
370       final String family) throws IOException {
371     HTableDescriptor htd = master.getTableDescriptors().get(tableName);
372     assertTrue(htd != null);
373     assertTrue(htd.hasFamily(family.getBytes()));
374   }
375 
376   public static void validateColumnFamilyDeletion(final HMaster master, final TableName tableName,
377       final String family) throws IOException {
378     // verify htd
379     HTableDescriptor htd = master.getTableDescriptors().get(tableName);
380     assertTrue(htd != null);
381     assertFalse(htd.hasFamily(family.getBytes()));
382 
383     // verify fs
384     final FileSystem fs = master.getMasterFileSystem().getFileSystem();
385     final Path tableDir = FSUtils.getTableDir(master.getMasterFileSystem().getRootDir(), tableName);
386     for (Path regionDir: FSUtils.getRegionDirs(fs, tableDir)) {
387       final Path familyDir = new Path(regionDir, family);
388       assertFalse(family + " family dir should not exist", fs.exists(familyDir));
389     }
390   }
391 
392   public static void validateColumnFamilyModification(final HMaster master,
393       final TableName tableName, final String family, HColumnDescriptor columnDescriptor)
394       throws IOException {
395     HTableDescriptor htd = master.getTableDescriptors().get(tableName);
396     assertTrue(htd != null);
397 
398     HColumnDescriptor hcfd = htd.getFamily(family.getBytes());
399     assertTrue(hcfd.equals(columnDescriptor));
400   }
401 
402   public static void loadData(final Connection connection, final TableName tableName,
403       int rows, final byte[][] splitKeys,  final String... sfamilies) throws IOException {
404     byte[][] families = new byte[sfamilies.length][];
405     for (int i = 0; i < families.length; ++i) {
406       families[i] = Bytes.toBytes(sfamilies[i]);
407     }
408 
409     BufferedMutator mutator = connection.getBufferedMutator(tableName);
410 
411     // Ensure one row per region
412     assertTrue(rows >= splitKeys.length);
413     for (byte[] k: splitKeys) {
414       byte[] value = Bytes.add(Bytes.toBytes(System.currentTimeMillis()), k);
415       byte[] key = Bytes.add(k, Bytes.toBytes(MD5Hash.getMD5AsHex(value)));
416       mutator.mutate(createPut(families, key, value));
417       rows--;
418     }
419 
420     // Add other extra rows. more rows, more files
421     while (rows-- > 0) {
422       byte[] value = Bytes.add(Bytes.toBytes(System.currentTimeMillis()), Bytes.toBytes(rows));
423       byte[] key = Bytes.toBytes(MD5Hash.getMD5AsHex(value));
424       mutator.mutate(createPut(families, key, value));
425     }
426     mutator.flush();
427   }
428 
429   private static Put createPut(final byte[][] families, final byte[] key, final byte[] value) {
430     byte[] q = Bytes.toBytes("q");
431     Put put = new Put(key);
432     put.setDurability(Durability.SKIP_WAL);
433     for (byte[] family: families) {
434       put.add(family, q, value);
435     }
436     return put;
437   }
438 
439   public static long generateNonceGroup(final HMaster master) {
440     return master.getConnection().getNonceGenerator().getNonceGroup();
441   }
442 
443   public static long generateNonce(final HMaster master) {
444     return master.getConnection().getNonceGenerator().newNonce();
445   }
446 
447   public static class InjectAbortOnLoadListener
448       implements ProcedureExecutor.ProcedureExecutorListener {
449     private final ProcedureExecutor<MasterProcedureEnv> procExec;
450     private TreeSet<Long> procsToAbort = null;
451 
452     public InjectAbortOnLoadListener(final ProcedureExecutor<MasterProcedureEnv> procExec) {
453       this.procExec = procExec;
454     }
455 
456     public void addProcId(long procId) {
457       if (procsToAbort == null) {
458         procsToAbort = new TreeSet<Long>();
459       }
460       procsToAbort.add(procId);
461     }
462 
463     @Override
464     public void procedureLoaded(long procId) {
465       if (procsToAbort != null && !procsToAbort.contains(procId)) {
466         return;
467       }
468       procExec.abort(procId);
469     }
470 
471     @Override
472     public void procedureAdded(long procId) { /* no-op */ }
473 
474     @Override
475     public void procedureFinished(long procId) { /* no-op */ }
476   }
477 }