You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by te...@apache.org on 2011/11/08 00:50:50 UTC
svn commit: r1199016 - in /hbase/branches/0.92: ./
src/main/java/org/apache/hadoop/hbase/ipc/
src/main/java/org/apache/hadoop/hbase/mapreduce/
src/main/java/org/apache/hadoop/hbase/regionserver/ src/main/resources/
src/test/java/org/apache/hadoop/hbase...
Author: tedyu
Date: Mon Nov 7 23:50:49 2011
New Revision: 1199016
URL: http://svn.apache.org/viewvc?rev=1199016&view=rev
Log:
HBASE-4740 [bulk load] the HBASE-4552 API can't tell if errors on region server are recoverable
(Jonathan Hsieh)
Modified:
hbase/branches/0.92/CHANGES.txt
hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/ipc/HRegionInterface.java
hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java
hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
hbase/branches/0.92/src/main/resources/hbase-default.xml
hbase/branches/0.92/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFilesSplitRecovery.java
Modified: hbase/branches/0.92/CHANGES.txt
URL: http://svn.apache.org/viewvc/hbase/branches/0.92/CHANGES.txt?rev=1199016&r1=1199015&r2=1199016&view=diff
==============================================================================
--- hbase/branches/0.92/CHANGES.txt (original)
+++ hbase/branches/0.92/CHANGES.txt Mon Nov 7 23:50:49 2011
@@ -409,6 +409,8 @@ Release 0.92.0 - Unreleased
HBASE-4753 org.apache.hadoop.hbase.regionserver.TestHRegionInfo#testGetSetOfHTD
throws NPE on trunk (nkeywal)
HBASE-4754 FSTableDescriptors.getTableInfoPath() should handle FileNotFoundException
+ HBASE-4740 [bulk load] the HBASE-4552 API can't tell if errors on region server are recoverable
+ (Jonathan Hsieh)
TESTS
HBASE-4492 TestRollingRestart fails intermittently
Modified: hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/ipc/HRegionInterface.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/ipc/HRegionInterface.java?rev=1199016&r1=1199015&r2=1199016&view=diff
==============================================================================
--- hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/ipc/HRegionInterface.java (original)
+++ hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/ipc/HRegionInterface.java Mon Nov 7 23:50:49 2011
@@ -315,8 +315,10 @@ public interface HRegionInterface extend
*
* @param familyPaths List of (family, hfile path) pairs
* @param regionName name of region to load hfiles into
+ * @return true if successful, false if failed recoverably
+ * @throws IOException if fails unrecoverably
*/
- public void bulkLoadHFiles(List<Pair<byte[], String>> familyPaths, byte[] regionName)
+ public boolean bulkLoadHFiles(List<Pair<byte[], String>> familyPaths, byte[] regionName)
throws IOException;
// Master methods
Modified: hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java?rev=1199016&r1=1199015&r2=1199016&view=diff
==============================================================================
--- hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java (original)
+++ hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/mapreduce/LoadIncrementalHFiles.java Mon Nov 7 23:50:49 2011
@@ -165,7 +165,8 @@ public class LoadIncrementalHFiles exten
/**
* Perform a bulk load of the given directory into the given
- * pre-existing table.
+ * pre-existing table. This method is not threadsafe.
+ *
* @param hfofDir the directory that was provided as the output path
* of a job using HFileOutputFormat
* @param table the table to load into
@@ -200,17 +201,24 @@ public class LoadIncrementalHFiles exten
discoverLoadQueue(queue, hfofDir);
int count = 0;
+ if (queue.isEmpty()) {
+ LOG.warn("Bulk load operation did not find any files to load in " +
+ "directory " + hfofDir.toUri() + ". Does it contain files in " +
+ "subdirectories that correspond to column family names?");
+ return;
+ }
+
// Assumes that region splits can happen while this occurs.
while (!queue.isEmpty()) {
// need to reload split keys each iteration.
final Pair<byte[][], byte[][]> startEndKeys = table.getStartEndKeys();
if (count != 0) {
LOG.info("Split occured while grouping HFiles, retry attempt " +
- + count + " with " + queue.size() + " files remaining to load");
+ + count + " with " + queue.size() + " files remaining to group or split");
}
-
- int maxRetries = cfg.getInt("hbase.bulkload.retries.number", 10);
- if (count >= maxRetries) {
+
+ int maxRetries = cfg.getInt("hbase.bulkload.retries.number", 0);
+ if (maxRetries != 0 && count >= maxRetries) {
LOG.error("Retry attempted " + count + " times without completing, bailing out");
return;
}
@@ -247,7 +255,7 @@ public class LoadIncrementalHFiles exten
* them. Any failures are re-queued for another pass with the
* groupOrSplitPhase.
*/
- private void bulkLoadPhase(final HTable table, final HConnection conn,
+ protected void bulkLoadPhase(final HTable table, final HConnection conn,
ExecutorService pool, Deque<LoadQueueItem> queue,
final Multimap<ByteBuffer, LoadQueueItem> regionGroups) throws IOException {
// atomically bulk load the groups.
@@ -269,16 +277,16 @@ public class LoadIncrementalHFiles exten
for (Future<List<LoadQueueItem>> future : loadingFutures) {
try {
List<LoadQueueItem> toRetry = future.get();
- if (toRetry != null && toRetry.size() != 0) {
- // LQIs that are requeued to be regrouped.
- queue.addAll(toRetry);
- }
+
+ // LQIs that are requeued to be regrouped.
+ queue.addAll(toRetry);
} catch (ExecutionException e1) {
Throwable t = e1.getCause();
if (t instanceof IOException) {
- LOG.error("IOException during bulk load", e1);
- throw (IOException)t; // would have been thrown if not parallelized,
+ // At this point something unrecoverable has happened.
+ // TODO Implement bulk load recovery
+ throw new IOException("BulkLoad encountered an unrecoverable problem", t);
}
LOG.error("Unexpected execution exception during bulk load", e1);
throw new IllegalStateException(t);
@@ -444,9 +452,12 @@ public class LoadIncrementalHFiles exten
* succeed atomically and fails atomically.
*
* Protected for testing.
+ *
+ * @return empty list if success, list of items to retry on recoverable
+ * failure
*/
protected List<LoadQueueItem> tryAtomicRegionLoad(final HConnection conn,
- byte[] tableName, final byte[] first, Collection<LoadQueueItem> lqis) {
+ byte[] tableName, final byte[] first, Collection<LoadQueueItem> lqis) throws IOException {
final List<Pair<byte[], String>> famPaths =
new ArrayList<Pair<byte[], String>>(lqis.size());
@@ -454,29 +465,33 @@ public class LoadIncrementalHFiles exten
famPaths.add(Pair.newPair(lqi.family, lqi.hfilePath.toString()));
}
- final ServerCallable<Void> svrCallable = new ServerCallable<Void>(conn,
+ final ServerCallable<Boolean> svrCallable = new ServerCallable<Boolean>(conn,
tableName, first) {
@Override
- public Void call() throws Exception {
+ public Boolean call() throws Exception {
LOG.debug("Going to connect to server " + location + " for row "
+ Bytes.toStringBinary(row));
byte[] regionName = location.getRegionInfo().getRegionName();
- server.bulkLoadHFiles(famPaths, regionName);
- return null;
+ return server.bulkLoadHFiles(famPaths, regionName);
}
};
- List<LoadQueueItem> toRetry = new ArrayList<LoadQueueItem>();
try {
- conn.getRegionServerWithRetries(svrCallable);
+ List<LoadQueueItem> toRetry = new ArrayList<LoadQueueItem>();
+ boolean success = conn.getRegionServerWithRetries(svrCallable);
+ if (!success) {
+ LOG.warn("Attempt to bulk load region containing "
+ + Bytes.toStringBinary(first) + " into table "
+ + Bytes.toStringBinary(tableName) + " with files " + lqis
+ + " failed. This is recoverable and they will be retried.");
+ toRetry.addAll(lqis); // return lqi's to retry
+ }
+ // success
+ return toRetry;
} catch (IOException e) {
- LOG.warn("Attempt to bulk load region containing "
- + Bytes.toStringBinary(first) + " into table "
- + Bytes.toStringBinary(tableName) + " with files " + lqis
- + " failed");
- toRetry.addAll(lqis);
+ LOG.error("Encountered unrecoverable error from region server", e);
+ throw e;
}
- return toRetry;
}
/**
Modified: hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java?rev=1199016&r1=1199015&r2=1199016&view=diff
==============================================================================
--- hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java (original)
+++ hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java Mon Nov 7 23:50:49 2011
@@ -2634,20 +2634,22 @@ public class HRegion implements HeapSize
* rows with multiple column families atomically.
*
* @param familyPaths List of Pair<byte[] column family, String hfilePath>
+ * @return true if successful, false if failed recoverably
+ * @throws IOException if failed unrecoverably.
*/
- public void bulkLoadHFiles(List<Pair<byte[], String>> familyPaths)
+ public boolean bulkLoadHFiles(List<Pair<byte[], String>> familyPaths)
throws IOException {
Preconditions.checkNotNull(familyPaths);
// we need writeLock for multi-family bulk load
startBulkRegionOperation(hasMultipleColumnFamilies(familyPaths));
- this.writeRequestsCount.increment();
- List<IOException> ioes = new ArrayList<IOException>();
- List<Pair<byte[], String>> failures = new ArrayList<Pair<byte[], String>>();
- boolean rangesOk = true;
try {
+ this.writeRequestsCount.increment();
+
// There possibly was a split that happend between when the split keys
// were gathered and before the HReiogn's write lock was taken. We need
// to validate the HFile region before attempting to bulk load all of them
+ List<IOException> ioes = new ArrayList<IOException>();
+ List<Pair<byte[], String>> failures = new ArrayList<Pair<byte[], String>>();
for (Pair<byte[], String> p : familyPaths) {
byte[] familyName = p.getFirst();
String path = p.getSecond();
@@ -2662,16 +2664,34 @@ public class HRegion implements HeapSize
try {
store.assertBulkLoadHFileOk(new Path(path));
+ } catch (WrongRegionException wre) {
+ // recoverable (file doesn't fit in region)
+ failures.add(p);
} catch (IOException ioe) {
- rangesOk = false;
+ // unrecoverable (hdfs problem)
ioes.add(ioe);
- failures.add(p);
}
}
+
+ // validation failed, bail out before doing anything permanent.
+ if (failures.size() != 0) {
+ StringBuilder list = new StringBuilder();
+ for (Pair<byte[], String> p : failures) {
+ list.append("\n").append(Bytes.toString(p.getFirst())).append(" : ")
+ .append(p.getSecond());
+ }
+ // problem when validating
+ LOG.warn("There was a recoverable bulk load failure likely due to a" +
+ " split. These (family, HFile) pairs were not loaded: " + list);
+ return false;
+ }
+
+ // validation failed because of some sort of IO problem.
if (ioes.size() != 0) {
- // validation failed, bail out before doing anything permanent.
- return;
+ LOG.error("There were IO errors when checking if bulk load is ok. " +
+ "throwing exception!");
+ throw MultipleIOException.createIOException(ioes);
}
for (Pair<byte[], String> p : familyPaths) {
@@ -2683,35 +2703,16 @@ public class HRegion implements HeapSize
} catch (IOException ioe) {
// a failure here causes an atomicity violation that we currently
// cannot recover from since it is likely a failed hdfs operation.
- ioes.add(ioe);
- failures.add(p);
- break;
- }
- }
- } finally {
- closeBulkRegionOperation();
- if (ioes.size() != 0) {
- StringBuilder list = new StringBuilder();
- for (Pair<byte[], String> p : failures) {
- list.append("\n").append(Bytes.toString(p.getFirst())).append(" : ")
- .append(p.getSecond());
- }
- if (rangesOk) {
// TODO Need a better story for reverting partial failures due to HDFS.
- LOG.error("There was a partial failure due to IO. These " +
- "(family,hfile) pairs were not loaded: " + list);
- } else {
- // problem when validating
- LOG.info("There was a recoverable bulk load failure likely due to a" +
- " split. These (family, HFile) pairs were not loaded: " + list);
- }
-
- if (ioes.size() == 1) {
- throw ioes.get(0);
+ LOG.error("There was a partial failure due to IO when attempting to" +
+ " load " + Bytes.toString(p.getFirst()) + " : "+ p.getSecond());
+ throw ioe;
}
- throw MultipleIOException.createIOException(ioes);
}
+ return true;
+ } finally {
+ closeBulkRegionOperation();
}
}
Modified: hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java?rev=1199016&r1=1199015&r2=1199016&view=diff
==============================================================================
--- hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java (original)
+++ hbase/branches/0.92/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java Mon Nov 7 23:50:49 2011
@@ -2344,13 +2344,15 @@ public class HRegionServer implements HR
/**
* Atomically bulk load several HFiles into an open region
+ * @return true if successful, false is failed but recoverably (no action)
+ * @throws IOException if failed unrecoverably
*/
@Override
- public void bulkLoadHFiles(List<Pair<byte[], String>> familyPaths,
+ public boolean bulkLoadHFiles(List<Pair<byte[], String>> familyPaths,
byte[] regionName) throws IOException {
checkOpen();
HRegion region = getRegion(regionName);
- region.bulkLoadHFiles(familyPaths);
+ return region.bulkLoadHFiles(familyPaths);
}
Map<String, Integer> rowlocks = new ConcurrentHashMap<String, Integer>();
Modified: hbase/branches/0.92/src/main/resources/hbase-default.xml
URL: http://svn.apache.org/viewvc/hbase/branches/0.92/src/main/resources/hbase-default.xml?rev=1199016&r1=1199015&r2=1199016&view=diff
==============================================================================
--- hbase/branches/0.92/src/main/resources/hbase-default.xml (original)
+++ hbase/branches/0.92/src/main/resources/hbase-default.xml Mon Nov 7 23:50:49 2011
@@ -132,10 +132,10 @@
</property>
<property>
<name>hbase.bulkload.retries.number</name>
- <value>10</value>
+ <value>0</value>
<description>Maximum retries. This is maximum number of iterations
to atomic bulk loads are attempted in the face of splitting operations
- Default: 10.
+ 0 means never give up. Default: 0.
</description>
</property>
<property>
Modified: hbase/branches/0.92/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFilesSplitRecovery.java
URL: http://svn.apache.org/viewvc/hbase/branches/0.92/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFilesSplitRecovery.java?rev=1199016&r1=1199015&r2=1199016&view=diff
==============================================================================
--- hbase/branches/0.92/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFilesSplitRecovery.java (original)
+++ hbase/branches/0.92/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFilesSplitRecovery.java Mon Nov 7 23:50:49 2011
@@ -1,6 +1,4 @@
/**
- * Copyright 2010 The Apache Software Foundation
- *
* 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
@@ -29,8 +27,10 @@ import static org.mockito.Mockito.mock;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collection;
+import java.util.Deque;
import java.util.List;
import java.util.NavigableMap;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
@@ -120,6 +120,63 @@ public class TestLoadIncrementalHFilesSp
}
}
+ private Path buildBulkFiles(String table, int value) throws Exception {
+ Path dir = util.getDataTestDir(table);
+ Path bulk1 = new Path(dir, table+value);
+ FileSystem fs = util.getTestFileSystem();
+ buildHFiles(fs, bulk1, value);
+ return bulk1;
+ }
+
+ /**
+ * Populate table with known values.
+ */
+ private void populateTable(String table, int value) throws Exception {
+ // create HFiles for different column families
+ LoadIncrementalHFiles lih = new LoadIncrementalHFiles(
+ util.getConfiguration());
+ Path bulk1 = buildBulkFiles(table, value);
+ HTable t = new HTable(util.getConfiguration(), Bytes.toBytes(table));
+ lih.doBulkLoad(bulk1, t);
+ }
+
+ /**
+ * Split the known table in half. (this is hard coded for this test suite)
+ */
+ private void forceSplit(String table) {
+ try {
+ // need to call regions server to by synchronous but isn't visible.
+ HRegionServer hrs = util.getRSForFirstRegionInTable(Bytes
+ .toBytes(table));
+
+ for (HRegionInfo hri : hrs.getOnlineRegions()) {
+ if (Bytes.equals(hri.getTableName(), Bytes.toBytes(table))) {
+ // splitRegion doesn't work if startkey/endkey are null
+ hrs.splitRegion(hri, rowkey(ROWCOUNT / 2)); // hard code split
+ }
+ }
+
+ // verify that split completed.
+ int regions;
+ do {
+ regions = 0;
+ for (HRegionInfo hri : hrs.getOnlineRegions()) {
+ if (Bytes.equals(hri.getTableName(), Bytes.toBytes(table))) {
+ regions++;
+ }
+ }
+ if (regions != 2) {
+ LOG.info("Taking some time to complete split...");
+ Thread.sleep(250);
+ }
+ } while (regions != 2);
+ } catch (IOException e) {
+ e.printStackTrace();
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+
@BeforeClass
public static void setupCluster() throws Exception {
util = new HBaseTestingUtility();
@@ -131,6 +188,10 @@ public class TestLoadIncrementalHFilesSp
util.shutdownMiniCluster();
}
+ /**
+ * Checks that all columns have the expected value and that there is the
+ * expected number of rows.
+ */
void assertExpectedTable(String table, int count, int value) {
try {
assertEquals(util.getHBaseAdmin().listTables(table).length, 1);
@@ -146,18 +207,20 @@ public class TestLoadIncrementalHFilesSp
assertTrue(Bytes.equals(val, value(value)));
}
}
-
}
assertEquals(count, i);
-
} catch (IOException e) {
fail("Failed due to exception");
}
}
- @Test
- public void testBulkLoadPhaseRecovery() throws Exception {
- String table = "bulkPhaseRetry";
+ /**
+ * Test that shows that exception thrown from the RS side will result in an
+ * exception on the LIHFile client.
+ */
+ @Test(expected=IOException.class)
+ public void testBulkLoadPhaseFailure() throws Exception {
+ String table = "bulkLoadPhaseFailure";
setupTable(table, 10);
final AtomicInteger attmptedCalls = new AtomicInteger();
@@ -166,7 +229,7 @@ public class TestLoadIncrementalHFilesSp
util.getConfiguration()) {
protected List<LoadQueueItem> tryAtomicRegionLoad(final HConnection conn,
- byte[] tableName, final byte[] first, Collection<LoadQueueItem> lqis) {
+ byte[] tableName, final byte[] first, Collection<LoadQueueItem> lqis) throws IOException {
int i = attmptedCalls.incrementAndGet();
if (i == 1) {
HConnection errConn = mock(HConnection.class);
@@ -186,16 +249,11 @@ public class TestLoadIncrementalHFilesSp
};
// create HFiles for different column families
- FileSystem fs = util.getTestFileSystem();
- Path dir = util.getDataTestDir(table);
- buildHFiles(fs, dir, 1);
+ Path dir = buildBulkFiles(table, 1);
HTable t = new HTable(util.getConfiguration(), Bytes.toBytes(table));
lih.doBulkLoad(dir, t);
- // check that data was loaded
- assertEquals(attmptedCalls.get(), 2);
- assertEquals(failedCalls.get(), 1);
- assertExpectedTable(table, ROWCOUNT, 1);
+ fail("doBulkLoad should have thrown an exception");
}
/**
@@ -206,119 +264,109 @@ public class TestLoadIncrementalHFilesSp
*/
@Test
public void testSplitWhileBulkLoadPhase() throws Exception {
- final String table = "bulkPhaseSplit";
+ final String table = "splitWhileBulkloadPhase";
setupTable(table, 10);
- LoadIncrementalHFiles lih = new LoadIncrementalHFiles(
- util.getConfiguration());
-
-
- // create HFiles for different column families
- Path dir = util.getDataTestDir(table);
- Path bulk1 = new Path(dir, "normalBulkload");
- FileSystem fs = util.getTestFileSystem();
- buildHFiles(fs, bulk1, 1);
- HTable t = new HTable(util.getConfiguration(), Bytes.toBytes(table));
- lih.doBulkLoad(bulk1, t);
+ populateTable(table,1);
assertExpectedTable(table, ROWCOUNT, 1);
- // Now let's cause trouble
- final AtomicInteger attmptedCalls = new AtomicInteger();
+ // Now let's cause trouble. This will occur after checks and cause bulk
+ // files to fail when attempt to atomically import. This is recoverable.
+ final AtomicInteger attemptedCalls = new AtomicInteger();
LoadIncrementalHFiles lih2 = new LoadIncrementalHFiles(
util.getConfiguration()) {
- protected List<LoadQueueItem> tryAtomicRegionLoad(final HConnection conn,
- byte[] tableName, final byte[] first, Collection<LoadQueueItem> lqis) {
- int i = attmptedCalls.incrementAndGet();
+ protected void bulkLoadPhase(final HTable htable, final HConnection conn,
+ ExecutorService pool, Deque<LoadQueueItem> queue,
+ final Multimap<ByteBuffer, LoadQueueItem> regionGroups) throws IOException {
+ int i = attemptedCalls.incrementAndGet();
if (i == 1) {
// On first attempt force a split.
- try {
- // need to call regions server to by synchronous but isn't visible.
-
- HRegionServer hrs = util.getRSForFirstRegionInTable(Bytes
- .toBytes(table));
-
- HRegionInfo region = null;
- for (HRegionInfo hri : hrs.getOnlineRegions()) {
- if (Bytes.equals(hri.getTableName(), Bytes.toBytes(table))) {
- // splitRegion doesn't work if startkey/endkey are null
- hrs.splitRegion(hri, rowkey(ROWCOUNT / 2)); // hard code split
- }
- }
-
- int regions;
- do {
- regions = 0;
- for (HRegionInfo hri : hrs.getOnlineRegions()) {
- if (Bytes.equals(hri.getTableName(), Bytes.toBytes(table))) {
- regions++;
- }
- }
- if (regions != 2) {
- LOG.info("Taking some time to complete split...");
- Thread.sleep(250);
- }
- } while (regions != 2);
- } catch (IOException e) {
- e.printStackTrace();
- } catch (InterruptedException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
+ forceSplit(table);
}
- return super.tryAtomicRegionLoad(conn, tableName, first, lqis);
+ super.bulkLoadPhase(htable, conn, pool, queue, regionGroups);
}
};
// create HFiles for different column families
- Path bulk2 = new Path(dir, "bulkload2");
- buildHFiles(fs, bulk2, 2); // all values are '2'
- lih2.doBulkLoad(bulk2, t);
+ HTable t = new HTable(util.getConfiguration(), Bytes.toBytes(table));
+ Path bulk = buildBulkFiles(table, 2);
+ lih2.doBulkLoad(bulk, t);
// check that data was loaded
-
// The three expected attempts are 1) failure because need to split, 2)
// load of split top 3) load of split bottom
- assertEquals(attmptedCalls.get(), 3);
+ assertEquals(attemptedCalls.get(), 3);
assertExpectedTable(table, ROWCOUNT, 2);
}
+ /**
+ * This test splits a table and attempts to bulk load. The bulk import files
+ * should be split before atomically importing.
+ */
+ @Test
+ public void testGroupOrSplitPresplit() throws Exception {
+ final String table = "groupOrSplitPresplit";
+ setupTable(table, 10);
+ populateTable(table, 1);
+ assertExpectedTable(table, ROWCOUNT, 1);
+ forceSplit(table);
+
+ final AtomicInteger countedLqis= new AtomicInteger();
+ LoadIncrementalHFiles lih = new LoadIncrementalHFiles(
+ util.getConfiguration()) {
+ protected List<LoadQueueItem> groupOrSplit(
+ Multimap<ByteBuffer, LoadQueueItem> regionGroups,
+ final LoadQueueItem item, final HTable htable,
+ final Pair<byte[][], byte[][]> startEndKeys) throws IOException {
+ List<LoadQueueItem> lqis = super.groupOrSplit(regionGroups, item, htable, startEndKeys);
+ if (lqis != null) {
+ countedLqis.addAndGet(lqis.size());
+ }
+ return lqis;
+ }
+ };
+
+ // create HFiles for different column families
+ Path bulk = buildBulkFiles(table, 2);
+ HTable ht = new HTable(util.getConfiguration(), Bytes.toBytes(table));
+ lih.doBulkLoad(bulk, ht);
+
+ assertExpectedTable(table, ROWCOUNT, 2);
+ assertEquals(20, countedLqis.get());
+ }
+
+ /**
+ * This simulates an remote exception which should cause LIHF to exit with an
+ * exception.
+ */
@Test(expected = IOException.class)
public void testGroupOrSplitFailure() throws Exception {
- String table = "groupOrSplitStoreFail";
+ String table = "groupOrSplitFailure";
setupTable(table, 10);
- try {
- LoadIncrementalHFiles lih = new LoadIncrementalHFiles(
- util.getConfiguration()) {
- int i = 0;
-
- protected List<LoadQueueItem> groupOrSplit(
- Multimap<ByteBuffer, LoadQueueItem> regionGroups,
- final LoadQueueItem item, final HTable table,
- final Pair<byte[][], byte[][]> startEndKeys) throws IOException {
- i++;
-
- if (i == 5) {
- throw new IOException("failure");
- }
- return super.groupOrSplit(regionGroups, item, table, startEndKeys);
- }
- };
+ LoadIncrementalHFiles lih = new LoadIncrementalHFiles(
+ util.getConfiguration()) {
+ int i = 0;
- Path dir = util.getDataTestDir(table);
+ protected List<LoadQueueItem> groupOrSplit(
+ Multimap<ByteBuffer, LoadQueueItem> regionGroups,
+ final LoadQueueItem item, final HTable table,
+ final Pair<byte[][], byte[][]> startEndKeys) throws IOException {
+ i++;
- // create HFiles for different column families
- FileSystem fs = util.getTestFileSystem();
- buildHFiles(fs, dir, 1);
- HTable t = new HTable(util.getConfiguration(), Bytes.toBytes(table));
- lih.doBulkLoad(dir, t);
+ if (i == 5) {
+ throw new IOException("failure");
+ }
+ return super.groupOrSplit(regionGroups, item, table, startEndKeys);
+ }
+ };
- // check that data was loaded
- assertExpectedTable(table, ROWCOUNT, 1);
+ // create HFiles for different column families
+ Path dir = buildBulkFiles(table,1);
+ HTable t = new HTable(util.getConfiguration(), Bytes.toBytes(table));
+ lih.doBulkLoad(dir, t);
- } finally {
- util.shutdownMiniCluster();
- }
+ fail("doBulkLoad should have thrown an exception");
}
}