You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by jd...@apache.org on 2014/05/01 00:07:26 UTC
svn commit: r1591495 - in /hbase/branches/0.94/src:
main/java/org/apache/hadoop/hbase/regionserver/
test/java/org/apache/hadoop/hbase/mapreduce/
test/java/org/apache/hadoop/hbase/regionserver/
test/java/org/apache/hadoop/hbase/regionserver/wal/ test/ja...
Author: jdcryans
Date: Wed Apr 30 22:07:26 2014
New Revision: 1591495
URL: http://svn.apache.org/r1591495
Log:
HBASE-10958 [dataloss] Bulk loading with seqids can prevent some log entries from being replayed
Added:
hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/util/HFileTestUtil.java
Modified:
hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java
hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java
hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFiles.java
hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompaction.java
hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java
Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java?rev=1591495&r1=1591494&r2=1591495&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java (original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java Wed Apr 30 22:07:26 2014
@@ -341,6 +341,81 @@ public class HRegion implements HeapSize
ClassSize.OBJECT + 5 * Bytes.SIZEOF_BOOLEAN);
}
+ /**
+ * Objects from this class are created when flushing to describe all the different states that
+ * that method ends up in. The Result enum describes those states. The sequence id should only
+ * be specified if the flush was successful, and the failure message should only be speficied
+ * if it didn't flush.
+ */
+ public static class FlushResult {
+ enum Result {
+ FLUSHED_NO_COMPACTION_NEEDED,
+ FLUSHED_COMPACTION_NEEDED,
+ // Special case where a flush didn't run because there's nothing in the memstores. Used when
+ // bulk loading to know when we can still load even if a flush didn't happen.
+ CANNOT_FLUSH_MEMSTORE_EMPTY,
+ CANNOT_FLUSH
+ // Be careful adding more to this enum, look at the below methods to make sure
+ }
+
+ final Result result;
+ final String failureReason;
+ final long flushSequenceId;
+
+ /**
+ * Convenience constructor to use when the flush is successful, the failure message is set to
+ * null.
+ * @param result Expecting FLUSHED_NO_COMPACTION_NEEDED or FLUSHED_COMPACTION_NEEDED.
+ * @param flushSequenceId Generated sequence id that comes right after the edits in the
+ * memstores.
+ */
+ FlushResult(Result result, long flushSequenceId) {
+ this(result, flushSequenceId, null);
+ assert result == Result.FLUSHED_NO_COMPACTION_NEEDED || result == Result
+ .FLUSHED_COMPACTION_NEEDED;
+ }
+
+ /**
+ * Convenience constructor to use when we cannot flush.
+ * @param result Expecting CANNOT_FLUSH_MEMSTORE_EMPTY or CANNOT_FLUSH.
+ * @param failureReason Reason why we couldn't flush.
+ */
+ FlushResult(Result result, String failureReason) {
+ this(result, -1, failureReason);
+ assert result == Result.CANNOT_FLUSH_MEMSTORE_EMPTY || result == Result.CANNOT_FLUSH;
+ }
+
+ /**
+ * Constructor with all the parameters.
+ * @param result Any of the Result.
+ * @param flushSequenceId Generated sequence id if the memstores were flushed else -1.
+ * @param failureReason Reason why we couldn't flush, or null.
+ */
+ FlushResult(Result result, long flushSequenceId, String failureReason) {
+ this.result = result;
+ this.flushSequenceId = flushSequenceId;
+ this.failureReason = failureReason;
+ }
+
+ /**
+ * Convenience method, the equivalent of checking if result is
+ * FLUSHED_NO_COMPACTION_NEEDED or FLUSHED_NO_COMPACTION_NEEDED.
+ * @return true if the memstores were flushed, else false.
+ */
+ public boolean isFlushSucceeded() {
+ return result == Result.FLUSHED_NO_COMPACTION_NEEDED || result == Result
+ .FLUSHED_COMPACTION_NEEDED;
+ }
+
+ /**
+ * Convenience method, the equivalent of checking if result is FLUSHED_COMPACTION_NEEDED.
+ * @return True if the flush requested a compaction, else false (doesn't even mean it flushed).
+ */
+ public boolean isCompactionNeeded() {
+ return result == Result.FLUSHED_COMPACTION_NEEDED;
+ }
+ }
+
final WriteState writestate = new WriteState();
long memstoreFlushSize;
@@ -610,15 +685,12 @@ public class HRegion implements HeapSize
for (int i = 0; i < htableDescriptor.getFamilies().size(); i++) {
Future<Store> future = completionService.take();
Store store = future.get();
-
this.stores.put(store.getColumnFamilyName().getBytes(), store);
- // Do not include bulk loaded files when determining seqIdForReplay
- long storeSeqIdForReplay = store.getMaxSequenceId(false);
+
+ long storeSeqIdForReplay = store.getMaxSequenceId();
maxSeqIdInStores.put(store.getColumnFamilyName().getBytes(), storeSeqIdForReplay);
- // Include bulk loaded files when determining seqIdForAssignment
- long storeSeqIdForAssignment = store.getMaxSequenceId(true);
- if (maxSeqId == -1 || storeSeqIdForAssignment > maxSeqId) {
- maxSeqId = storeSeqIdForAssignment;
+ if (maxSeqId == -1 || storeSeqIdForReplay > maxSeqId) {
+ maxSeqId = storeSeqIdForReplay;
}
long maxStoreMemstoreTS = store.getMaxMemstoreTS();
if (maxStoreMemstoreTS > maxMemstoreTS) {
@@ -1483,11 +1555,12 @@ public class HRegion implements HeapSize
* @throws DroppedSnapshotException Thrown when replay of hlog is required
* because a Snapshot was not properly persisted.
*/
- public boolean flushcache() throws IOException {
+ public FlushResult flushcache() throws IOException {
// fail-fast instead of waiting on the lock
if (this.closing.get()) {
- LOG.debug("Skipping flush on " + this + " because closing");
- return false;
+ String msg = "Skipping flush on " + this + " because closing";
+ LOG.debug(msg);
+ return new FlushResult(FlushResult.Result.CANNOT_FLUSH, msg);
}
MonitoredTask status = TaskMonitor.get().createStatus("Flushing " + this);
status.setStatus("Acquiring readlock on region");
@@ -1495,9 +1568,10 @@ public class HRegion implements HeapSize
lock.readLock().lock();
try {
if (this.closed.get()) {
- LOG.debug("Skipping flush on " + this + " because closed");
- status.abort("Skipped: closed");
- return false;
+ String msg = "Skipping flush on " + this + " because closed";
+ LOG.debug(msg);
+ status.abort(msg);
+ return new FlushResult(FlushResult.Result.CANNOT_FLUSH, msg);
}
if (coprocessorHost != null) {
status.setStatus("Running coprocessor pre-flush hooks");
@@ -1516,14 +1590,15 @@ public class HRegion implements HeapSize
+ ", flushing=" + writestate.flushing + ", writesEnabled="
+ writestate.writesEnabled);
}
- status.abort("Not flushing since "
+ String msg = "Not flushing since "
+ (writestate.flushing ? "already flushing"
- : "writes not enabled"));
- return false;
+ : "writes not enabled");
+ status.abort(msg);
+ return new FlushResult(FlushResult.Result.CANNOT_FLUSH, msg);
}
}
try {
- boolean result = internalFlushcache(status);
+ FlushResult fs = internalFlushcache(status);
if (coprocessorHost != null) {
status.setStatus("Running post-flush coprocessor hooks");
@@ -1531,7 +1606,7 @@ public class HRegion implements HeapSize
}
status.markComplete("Flush successful");
- return result;
+ return fs;
} finally {
synchronized (writestate) {
writestate.flushing = false;
@@ -1603,7 +1678,7 @@ public class HRegion implements HeapSize
* @throws DroppedSnapshotException Thrown when replay of hlog is required
* because a Snapshot was not properly persisted.
*/
- protected boolean internalFlushcache(MonitoredTask status)
+ protected FlushResult internalFlushcache(MonitoredTask status)
throws IOException {
return internalFlushcache(this.log, -1, status);
}
@@ -1617,7 +1692,7 @@ public class HRegion implements HeapSize
* @throws IOException
* @see #internalFlushcache(MonitoredTask)
*/
- protected boolean internalFlushcache(
+ protected FlushResult internalFlushcache(
final HLog wal, final long myseqid, MonitoredTask status)
throws IOException {
if (this.rsServices != null && this.rsServices.isAborted()) {
@@ -1630,7 +1705,7 @@ public class HRegion implements HeapSize
this.lastFlushTime = startTime;
// If nothing to flush, return and avoid logging start/stop flush.
if (this.memstoreSize.get() <= 0) {
- return false;
+ return new FlushResult(FlushResult.Result.CANNOT_FLUSH_MEMSTORE_EMPTY, "Nothing to flush");
}
if (LOG.isDebugEnabled()) {
LOG.debug("Started memstore flush for " + this +
@@ -1777,7 +1852,8 @@ public class HRegion implements HeapSize
status.setStatus(msg);
this.recentFlushes.add(new Pair<Long,Long>(time/1000, totalFlushableSize));
- return compactionRequested;
+ return new FlushResult(compactionRequested ? FlushResult.Result.FLUSHED_COMPACTION_NEEDED :
+ FlushResult.Result.FLUSHED_NO_COMPACTION_NEEDED, sequenceId);
}
/**
@@ -3788,6 +3864,22 @@ public class HRegion implements HeapSize
return false;
}
+ long seqId = -1;
+ // We need to assign a sequential ID that's in between two memstores in order to preserve
+ // the guarantee that all the edits lower than the highest sequential ID from all the
+ // HFiles are flushed on disk. See HBASE-10958.
+ if (assignSeqId) {
+ FlushResult fs = this.flushcache();
+ if (fs.isFlushSucceeded()) {
+ seqId = fs.flushSequenceId;
+ } else if (fs.result == FlushResult.Result.CANNOT_FLUSH_MEMSTORE_EMPTY) {
+ seqId = this.log.obtainSeqNum();
+ } else {
+ throw new IOException("Could not bulk load with an assigned sequential ID because the " +
+ "flush didn't run. Reason for not flushing: " + fs.failureReason);
+ }
+ }
+
for (Pair<byte[], String> p : familyPaths) {
byte[] familyName = p.getFirst();
String path = p.getSecond();
@@ -3797,7 +3889,7 @@ public class HRegion implements HeapSize
if(bulkLoadListener != null) {
finalPath = bulkLoadListener.prepareBulkLoad(familyName, path);
}
- store.bulkLoadHFile(finalPath, assignSeqId ? this.log.obtainSeqNum() : -1);
+ store.bulkLoadHFile(finalPath, seqId);
if(bulkLoadListener != null) {
bulkLoadListener.doneBulkLoad(familyName, path);
}
Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java?rev=1591495&r1=1591494&r2=1591495&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java (original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java Wed Apr 30 22:07:26 2014
@@ -2470,7 +2470,7 @@ public class HRegionServer implements HR
throw new IllegalArgumentException("No region : " + new String(regionName)
+ " available");
}
- boolean needsCompaction = region.flushcache();
+ boolean needsCompaction = region.flushcache().isCompactionNeeded();
if (needsCompaction) {
this.compactSplitThread.requestCompaction(region, "Compaction through user triggered flush");
}
@@ -2487,7 +2487,7 @@ public class HRegionServer implements HR
+ " available");
}
if (region.getLastFlushTime() < ifOlderThanTS) {
- boolean needsCompaction = region.flushcache();
+ boolean needsCompaction = region.flushcache().isCompactionNeeded();
if (needsCompaction) {
this.compactSplitThread
.requestCompaction(region, "Compaction through user triggered flush");
@@ -3335,7 +3335,7 @@ public class HRegionServer implements HR
checkOpen();
LOG.info("Flushing " + regionInfo.getRegionNameAsString());
HRegion region = getRegion(regionInfo.getRegionName());
- boolean needsCompaction = region.flushcache();
+ boolean needsCompaction = region.flushcache().isCompactionNeeded();
if (needsCompaction) {
this.compactSplitThread.requestCompaction(region, "Compaction through user triggered flush");
}
Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java?rev=1591495&r1=1591494&r2=1591495&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java (original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java Wed Apr 30 22:07:26 2014
@@ -417,7 +417,7 @@ class MemStoreFlusher extends HasThread
lock.lock();
}
try {
- boolean shouldCompact = region.flushcache();
+ boolean shouldCompact = region.flushcache().isCompactionNeeded();
// We just want to check the size
boolean shouldSplit = region.checkSplit() != null;
if (shouldSplit) {
Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java?rev=1591495&r1=1591494&r2=1591495&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java (original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java Wed Apr 30 22:07:26 2014
@@ -341,8 +341,8 @@ public class Store extends SchemaConfigu
/**
* @return The maximum sequence id in all store files.
*/
- long getMaxSequenceId(boolean includeBulkFiles) {
- return StoreFile.getMaxSequenceIdInList(this.getStorefiles(), includeBulkFiles);
+ long getMaxSequenceId() {
+ return StoreFile.getMaxSequenceIdInList(this.getStorefiles());
}
/**
@@ -1145,7 +1145,7 @@ public class Store extends SchemaConfigu
}
// Max-sequenceID is the last key in the files we're compacting
- long maxId = StoreFile.getMaxSequenceIdInList(filesToCompact, true);
+ long maxId = StoreFile.getMaxSequenceIdInList(filesToCompact);
// Ready to go. Have list of files to compact.
LOG.info("Starting compaction of " + filesToCompact.size() + " file(s) in "
@@ -1213,7 +1213,7 @@ public class Store extends SchemaConfigu
}
filesToCompact = filesToCompact.subList(count - N, count);
- maxId = StoreFile.getMaxSequenceIdInList(filesToCompact, true);
+ maxId = StoreFile.getMaxSequenceIdInList(filesToCompact);
isMajor = (filesToCompact.size() == storefiles.size());
filesCompacting.addAll(filesToCompact);
Collections.sort(filesCompacting, StoreFile.Comparators.SEQ_ID);
Modified: hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java?rev=1591495&r1=1591494&r2=1591495&view=diff
==============================================================================
--- hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java (original)
+++ hbase/branches/0.94/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFile.java Wed Apr 30 22:07:26 2014
@@ -439,13 +439,10 @@ public class StoreFile extends SchemaCon
* @return 0 if no non-bulk-load files are provided or, this is Store that
* does not yet have any store files.
*/
- public static long getMaxSequenceIdInList(Collection<StoreFile> sfs,
- boolean includeBulkLoadedFiles) {
+ public static long getMaxSequenceIdInList(Collection<StoreFile> sfs) {
long max = 0;
for (StoreFile sf : sfs) {
- if (includeBulkLoadedFiles || !sf.isBulkLoadResult()) {
- max = Math.max(max, sf.getMaxSequenceId());
- }
+ max = Math.max(max, sf.getMaxSequenceId());
}
return max;
}
Modified: hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFiles.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFiles.java?rev=1591495&r1=1591494&r2=1591495&view=diff
==============================================================================
--- hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFiles.java (original)
+++ hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFiles.java Wed Apr 30 22:07:26 2014
@@ -33,19 +33,18 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseTestingUtility;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
-import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.LargeTests;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.UserProvider;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
-import org.apache.hadoop.hbase.io.hfile.Compression;
import org.apache.hadoop.hbase.io.hfile.HFile;
import org.apache.hadoop.hbase.io.hfile.HFileScanner;
import org.apache.hadoop.hbase.regionserver.StoreFile;
import org.apache.hadoop.hbase.regionserver.StoreFile.BloomType;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.HFileTestUtil;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -67,10 +66,6 @@ public class TestLoadIncrementalHFiles {
Bytes.toBytes("ppp")
};
- public static int BLOCKSIZE = 64*1024;
- public static String COMPRESSION =
- Compression.Algorithm.NONE.getName();
-
static HBaseTestingUtility util = new HBaseTestingUtility();
@BeforeClass
@@ -144,7 +139,7 @@ public class TestLoadIncrementalHFiles {
for (byte[][] range : hfileRanges) {
byte[] from = range[0];
byte[] to = range[1];
- createHFile(util.getConfiguration(), fs, new Path(familyDir, "hfile_"
+ HFileTestUtil.createHFile(util.getConfiguration(), fs, new Path(familyDir, "hfile_"
+ hfileIdx++), FAMILY, QUALIFIER, from, to, 1000);
}
int expectedRows = hfileIdx * 1000;
@@ -180,8 +175,8 @@ public class TestLoadIncrementalHFiles {
for (byte[][] range : hfileRanges) {
byte[] from = range[0];
byte[] to = range[1];
- createHFile(util.getConfiguration(), fs, new Path(familyDir, "hfile_" + hfileIdx++), FAMILY,
- QUALIFIER, from, to, 1000);
+ HFileTestUtil.createHFile(util.getConfiguration(), fs, new Path(familyDir,
+ "hfile_" + hfileIdx++), FAMILY, QUALIFIER, from, to, 1000);
}
final byte[] TABLE = Bytes.toBytes("mytable_" + testName);
@@ -236,7 +231,7 @@ public class TestLoadIncrementalHFiles {
for (byte[][] range : hfileRanges) {
byte[] from = range[0];
byte[] to = range[1];
- createHFile(util.getConfiguration(), fs, new Path(familyDir, "hfile_"
+ HFileTestUtil.createHFile(util.getConfiguration(), fs, new Path(familyDir, "hfile_"
+ hfileIdx++), FAMILY, QUALIFIER, from, to, 1000);
}
@@ -268,7 +263,7 @@ public class TestLoadIncrementalHFiles {
FileSystem fs = util.getTestFileSystem();
Path testIn = new Path(dir, "testhfile");
HColumnDescriptor familyDesc = new HColumnDescriptor(FAMILY);
- createHFile(util.getConfiguration(), fs, testIn, FAMILY, QUALIFIER,
+ HFileTestUtil.createHFile(util.getConfiguration(), fs, testIn, FAMILY, QUALIFIER,
Bytes.toBytes("aaa"), Bytes.toBytes("zzz"), 1000);
Path bottomOut = new Path(dir, "bottom.out");
@@ -301,36 +296,6 @@ public class TestLoadIncrementalHFiles {
return count;
}
-
- /**
- * Create an HFile with the given number of rows between a given
- * start key and end key.
- * TODO put me in an HFileTestUtil or something?
- */
- static void createHFile(
- Configuration conf,
- FileSystem fs, Path path,
- byte[] family, byte[] qualifier,
- byte[] startKey, byte[] endKey, int numRows) throws IOException
- {
- HFile.Writer writer = HFile.getWriterFactory(conf, new CacheConfig(conf))
- .withPath(fs, path)
- .withBlockSize(BLOCKSIZE)
- .withCompression(COMPRESSION)
- .withComparator(KeyValue.KEY_COMPARATOR)
- .create();
- long now = System.currentTimeMillis();
- try {
- // subtract 2 since iterateOnSplits doesn't include boundary keys
- for (byte[] key : Bytes.iterateOnSplits(startKey, endKey, numRows-2)) {
- KeyValue kv = new KeyValue(key, family, qualifier, now, key);
- writer.append(kv);
- }
- } finally {
- writer.close();
- }
- }
-
private void addStartEndKeysForTest(TreeMap<byte[], Integer> map, byte[] first, byte[] last) {
Integer value = map.containsKey(first)?(Integer)map.get(first):0;
map.put(first, value+1);
Modified: hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompaction.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompaction.java?rev=1591495&r1=1591494&r2=1591495&view=diff
==============================================================================
--- hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompaction.java (original)
+++ hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompaction.java Wed Apr 30 22:07:26 2014
@@ -594,7 +594,7 @@ public class TestCompaction extends HBas
Store store = r.getStore(COLUMN_FAMILY);
List<StoreFile> storeFiles = store.getStorefiles();
- long maxId = StoreFile.getMaxSequenceIdInList(storeFiles, true);
+ long maxId = StoreFile.getMaxSequenceIdInList(storeFiles);
Compactor tool = new Compactor(this.conf);
StoreFile.Writer compactedFile = tool.compactForTesting(store, this.conf, storeFiles, false,
Modified: hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java?rev=1591495&r1=1591494&r2=1591495&view=diff
==============================================================================
--- hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java (original)
+++ hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java Wed Apr 30 22:07:26 2014
@@ -3540,7 +3540,7 @@ public class TestHRegion extends HBaseTe
private int flushesSinceCompact = 0;
private final int maxFlushesSinceCompact = 20;
public void doAnAction() throws Exception {
- if (region.flushcache()) {
+ if (region.flushcache().isCompactionNeeded()) {
++flushesSinceCompact;
}
// Compact regularly to avoid creating too many files and exceeding the ulimit.
@@ -4263,6 +4263,42 @@ public class TestHRegion extends HBaseTe
}
}
+ /**
+ * Test that we get the expected flush results back
+ * @throws IOException
+ */
+ @Test
+ public void testFlushResult() throws IOException {
+ String method = "testFlushResult";
+ byte[] tableName = Bytes.toBytes(method);
+ byte[] family = Bytes.toBytes("family");
+
+ this.region = initHRegion(tableName, method, conf, family);
+
+ // empty memstore, flush doesn't run
+ HRegion.FlushResult fr = region.flushcache();
+ assertFalse(fr.isFlushSucceeded());
+ assertFalse(fr.isCompactionNeeded());
+
+ // Flush enough files to get up to the threshold, doesn't need compactions
+ for (int i = 0; i < 3; i++) {
+ Put put = new Put(tableName).add(family, family, tableName);
+ region.put(put);
+ fr = region.flushcache();
+ assertTrue(fr.isFlushSucceeded());
+ assertFalse(fr.isCompactionNeeded());
+ }
+
+ // Two flushes after the threshold, compactions are needed
+ for (int i = 0; i < 2; i++) {
+ Put put = new Put(tableName).add(family, family, tableName);
+ region.put(put);
+ fr = region.flushcache();
+ assertTrue(fr.isFlushSucceeded());
+ assertTrue(fr.isCompactionNeeded());
+ }
+ }
+
private Configuration initSplit() {
Configuration conf = HBaseConfiguration.create(this.conf);
Modified: hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java?rev=1591495&r1=1591494&r2=1591495&view=diff
==============================================================================
--- hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java (original)
+++ hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java Wed Apr 30 22:07:26 2014
@@ -57,7 +57,6 @@ import org.apache.hadoop.hbase.client.Pu
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.io.hfile.HFile;
import org.apache.hadoop.hbase.master.HMaster;
import org.apache.hadoop.hbase.monitoring.MonitoredTask;
import org.apache.hadoop.hbase.regionserver.FlushRequester;
@@ -71,6 +70,7 @@ import org.apache.hadoop.hbase.security.
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.EnvironmentEdge;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.HFileTestUtil;
import org.apache.hadoop.hbase.util.Pair;
import org.junit.After;
import org.junit.AfterClass;
@@ -306,7 +306,7 @@ public class TestWALReplay {
public void testRegionMadeOfBulkLoadedFilesOnly()
throws IOException, SecurityException, IllegalArgumentException,
NoSuchFieldException, IllegalAccessException, InterruptedException {
- final String tableNameStr = "testReplayEditsWrittenViaHRegion";
+ final String tableNameStr = "testRegionMadeOfBulkLoadedFilesOnly";
final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableNameStr);
final Path basedir = new Path(this.hbaseRootDir, tableNameStr);
deleteDir(basedir);
@@ -317,19 +317,22 @@ public class TestWALReplay {
region2.getLog().closeAndDelete();
HLog wal = createWAL(this.conf);
HRegion region = HRegion.openHRegion(hri, htd, wal, this.conf);
- Path f = new Path(basedir, "hfile");
- HFile.Writer writer =
- HFile.getWriterFactoryNoCache(conf).withPath(fs, f).create();
+
byte [] family = htd.getFamilies().iterator().next().getName();
- byte [] row = Bytes.toBytes(tableNameStr);
- writer.append(new KeyValue(row, family, family, row));
- writer.close();
+ Path f = new Path(basedir, "hfile");
+ HFileTestUtil.createHFile(this.conf, fs, f, family, family, Bytes.toBytes(""),
+ Bytes.toBytes("z"), 10);
List <Pair<byte[],String>> hfs= new ArrayList<Pair<byte[],String>>(1);
hfs.add(Pair.newPair(family, f.toString()));
region.bulkLoadHFiles(hfs, true);
+
// Add an edit so something in the WAL
+ byte [] row = Bytes.toBytes(tableNameStr);
region.put((new Put(row)).add(family, family, family));
wal.sync();
+ final int rowsInsertedCount = 11;
+
+ assertEquals(rowsInsertedCount, getScannedCount(region.getScanner(new Scan())));
// Now 'crash' the region by stealing its wal
final Configuration newConf = HBaseConfiguration.create(this.conf);
@@ -343,6 +346,77 @@ public class TestWALReplay {
newConf, hri, htd, null);
long seqid2 = region2.initialize();
assertTrue(seqid2 > -1);
+ assertEquals(rowsInsertedCount, getScannedCount(region2.getScanner(new Scan())));
+
+ // I can't close wal1. Its been appropriated when we split.
+ region2.close();
+ wal2.closeAndDelete();
+ return null;
+ }
+ });
+ }
+
+ /**
+ * HRegion test case that is made of a major compacted HFile (created with three bulk loaded
+ * files) and an edit in the memstore.
+ * This is for HBASE-10958 "[dataloss] Bulk loading with seqids can prevent some log entries
+ * from being replayed"
+ * @throws IOException
+ * @throws IllegalAccessException
+ * @throws NoSuchFieldException
+ * @throws IllegalArgumentException
+ * @throws SecurityException
+ */
+ @Test
+ public void testCompactedBulkLoadedFiles()
+ throws IOException, SecurityException, IllegalArgumentException,
+ NoSuchFieldException, IllegalAccessException, InterruptedException {
+ final String tableNameStr = "testCompactedBulkLoadedFiles";
+ final HRegionInfo hri = createBasic3FamilyHRegionInfo(tableNameStr);
+ final Path basedir = new Path(this.hbaseRootDir, tableNameStr);
+ deleteDir(basedir);
+ final HTableDescriptor htd = createBasic3FamilyHTD(tableNameStr);
+ HRegion region2 = HRegion.createHRegion(hri,
+ hbaseRootDir, this.conf, htd);
+ region2.close();
+ region2.getLog().closeAndDelete();
+ HLog wal = createWAL(this.conf);
+ HRegion region = HRegion.openHRegion(hri, htd, wal, this.conf);
+
+ // Add an edit so something in the WAL
+ byte [] row = Bytes.toBytes(tableNameStr);
+ byte [] family = htd.getFamilies().iterator().next().getName();
+ region.put((new Put(row)).add(family, family, family));
+ wal.sync();
+
+ List <Pair<byte[],String>> hfs= new ArrayList<Pair<byte[],String>>(1);
+ for (int i = 0; i < 3; i++) {
+ Path f = new Path(basedir, "hfile"+i);
+ HFileTestUtil.createHFile(this.conf, fs, f, family, family, Bytes.toBytes(i + "00"),
+ Bytes.toBytes(i + "50"), 10);
+ hfs.add(Pair.newPair(family, f.toString()));
+ }
+ region.bulkLoadHFiles(hfs, true);
+ final int rowsInsertedCount = 31;
+ assertEquals(rowsInsertedCount, getScannedCount(region.getScanner(new Scan())));
+
+ // major compact to turn all the bulk loaded files into one normal file
+ region.compactStores(true);
+ assertEquals(rowsInsertedCount, getScannedCount(region.getScanner(new Scan())));
+
+ // Now 'crash' the region by stealing its wal
+ final Configuration newConf = HBaseConfiguration.create(this.conf);
+ User user = HBaseTestingUtility.getDifferentUser(newConf,
+ tableNameStr);
+ user.runAs(new PrivilegedExceptionAction() {
+ public Object run() throws Exception {
+ runWALSplit(newConf);
+ HLog wal2 = createWAL(newConf);
+ HRegion region2 = new HRegion(basedir, wal2, FileSystem.get(newConf),
+ newConf, hri, htd, null);
+ long seqid2 = region2.initialize();
+ assertTrue(seqid2 > -1);
+ assertEquals(rowsInsertedCount, getScannedCount(region2.getScanner(new Scan())));
// I can't close wal1. Its been appropriated when we split.
region2.close();
@@ -736,14 +810,14 @@ public class TestWALReplay {
try {
final HRegion region =
new HRegion(basedir, newWal, newFS, newConf, hri, htd, null) {
- protected boolean internalFlushcache(
+ protected FlushResult internalFlushcache(
final HLog wal, final long myseqid, MonitoredTask status)
throws IOException {
LOG.info("InternalFlushCache Invoked");
- boolean b = super.internalFlushcache(wal, myseqid,
+ FlushResult fs = super.internalFlushcache(wal, myseqid,
Mockito.mock(MonitoredTask.class));
flushcount.incrementAndGet();
- return b;
+ return fs;
};
};
long seqid = region.initialize();
Added: hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/util/HFileTestUtil.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/util/HFileTestUtil.java?rev=1591495&view=auto
==============================================================================
--- hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/util/HFileTestUtil.java (added)
+++ hbase/branches/0.94/src/test/java/org/apache/hadoop/hbase/util/HFileTestUtil.java Wed Apr 30 22:07:26 2014
@@ -0,0 +1,63 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.util;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.io.hfile.CacheConfig;
+import org.apache.hadoop.hbase.io.hfile.HFile;
+import org.apache.hadoop.hbase.regionserver.StoreFile;
+
+import java.io.IOException;
+
+/**
+ * Utility class for HFile-related testing.
+ */
+public class HFileTestUtil {
+
+ /**
+ * Create an HFile with the given number of rows between a given
+ * start key and end key.
+ */
+ public static void createHFile(
+ Configuration configuration,
+ FileSystem fs, Path path,
+ byte[] family, byte[] qualifier,
+ byte[] startKey, byte[] endKey, int numRows) throws IOException
+ {
+ HFile.Writer writer = HFile.getWriterFactory(configuration, new CacheConfig(configuration))
+ .withPath(fs, path)
+ .withComparator(KeyValue.KEY_COMPARATOR)
+ .create();
+ long now = System.currentTimeMillis();
+ try {
+ // subtract 2 since iterateOnSplits doesn't include boundary keys
+ for (byte[] key : Bytes.iterateOnSplits(startKey, endKey, numRows-2)) {
+ KeyValue kv = new KeyValue(key, family, qualifier, now, key);
+ writer.append(kv);
+ }
+ } finally {
+ writer.appendFileInfo(StoreFile.BULKLOAD_TIME_KEY,
+ Bytes.toBytes(System.currentTimeMillis()));
+ writer.close();
+ }
+ }
+}