You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by en...@apache.org on 2014/02/26 00:31:31 UTC
svn commit: r1571865 - in /hbase/branches/hbase-10070:
hbase-common/src/main/resources/
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/
hbase-server/src/main/java/org/apache/hadoop/hbase/util/
hbase-server/src/test/java/org/apache/hado...
Author: enis
Date: Tue Feb 25 23:31:30 2014
New Revision: 1571865
URL: http://svn.apache.org/r1571865
Log:
HBASE-10352 Region and RegionServer changes for opening region replicas, and refreshing store files
Added:
hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StorefileRefresherChore.java
hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/util/ServerRegionReplicaUtil.java
hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionReplicas.java
hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreFileRefresherChore.java
Modified:
hbase/branches/hbase-10070/hbase-common/src/main/resources/hbase-default.xml
hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java
hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java
hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileInfo.java
hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/HBaseTestingUtility.java
hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStore.java
Modified: hbase/branches/hbase-10070/hbase-common/src/main/resources/hbase-default.xml
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-10070/hbase-common/src/main/resources/hbase-default.xml?rev=1571865&r1=1571864&r2=1571865&view=diff
==============================================================================
--- hbase/branches/hbase-10070/hbase-common/src/main/resources/hbase-default.xml (original)
+++ hbase/branches/hbase-10070/hbase-common/src/main/resources/hbase-default.xml Tue Feb 25 23:31:30 2014
@@ -1150,4 +1150,17 @@ possible configurations would overwhelm
procedure. After implementing your own MasterProcedureManager, just put it in HBase's classpath
and add the fully qualified class name here.</description>
</property>
+ <property>
+ <name>hbase.regionserver.storefile.refresh.period</name>
+ <value>0</value>
+ <description>
+ The period (in milliseconds) for refreshing the store files for the secondary regions. 0
+ means this feature is disabled. Secondary regions sees new files (from flushes and
+ compactions) from primary once the secondary region refreshes the list of files in the
+ region (there is no notification mechanism). But too frequent refreshes might cause
+ extra Namenode pressure. If the files cannot be refreshed for longer than HFile TTL
+ (hbase.master.hfilecleaner.ttl) the requests are rejected. Configuring HFile TTL to a larger
+ value is also recommended with this setting.
+ </description>
+ </property>
</configuration>
Modified: hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java?rev=1571865&r1=1571864&r2=1571865&view=diff
==============================================================================
--- hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java (original)
+++ hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java Tue Feb 25 23:31:30 2014
@@ -137,6 +137,7 @@ import org.apache.hadoop.hbase.util.Envi
import org.apache.hadoop.hbase.util.FSUtils;
import org.apache.hadoop.hbase.util.HashedBytes;
import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.hbase.util.ServerRegionReplicaUtil;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.io.MultipleIOException;
import org.apache.hadoop.util.StringUtils;
@@ -195,7 +196,7 @@ public class HRegion implements HeapSize
public static final String LOAD_CFS_ON_DEMAND_CONFIG_KEY =
"hbase.hregion.scan.loadColumnFamiliesOnDemand";
-
+
/**
* This is the global default value for durability. All tables/mutations not
* defining a durability or using USE_DEFAULT will default to this value.
@@ -366,6 +367,9 @@ public class HRegion implements HeapSize
volatile boolean writesEnabled = true;
// Set if region is read-only
volatile boolean readOnly = false;
+ // whether the reads are enabled. This is different than readOnly, because readOnly is
+ // static in the lifetime of the region, while readsEnabled is dynamic
+ volatile boolean readsEnabled = true;
/**
* Set flags that make this region read-only.
@@ -385,6 +389,10 @@ public class HRegion implements HeapSize
return this.flushRequested;
}
+ void setReadsEnabled(boolean readsEnabled) {
+ this.readsEnabled = readsEnabled;
+ }
+
static final long HEAP_SIZE = ClassSize.align(
ClassSize.OBJECT + 5 * Bytes.SIZEOF_BOOLEAN);
}
@@ -399,7 +407,7 @@ public class HRegion implements HeapSize
private RegionServerAccounting rsAccounting;
private List<Pair<Long, Long>> recentFlushes = new ArrayList<Pair<Long,Long>>();
private long flushCheckInterval;
- // flushPerChanges is to prevent too many changes in memstore
+ // flushPerChanges is to prevent too many changes in memstore
private long flushPerChanges;
private long blockingMemStoreSize;
final long threadWakeFrequency;
@@ -499,7 +507,7 @@ public class HRegion implements HeapSize
throw new IllegalArgumentException(MEMSTORE_FLUSH_PER_CHANGES + " can not exceed "
+ MAX_FLUSH_PER_CHANGES);
}
-
+
this.rowLockWaitDuration = conf.getInt("hbase.rowlock.wait.duration",
DEFAULT_ROWLOCK_WAIT_DURATION);
@@ -644,7 +652,7 @@ public class HRegion implements HeapSize
fs.cleanupAnySplitDetritus();
fs.cleanupMergesDir();
- this.writestate.setReadOnly(this.htableDescriptor.isReadOnly());
+ this.writestate.setReadOnly(ServerRegionReplicaUtil.isReadOnly(this));
this.writestate.flushRequested = false;
this.writestate.compacting = 0;
@@ -737,7 +745,7 @@ public class HRegion implements HeapSize
for (Store store : this.stores.values()) {
try {
store.close();
- } catch (IOException e) {
+ } catch (IOException e) {
LOG.warn(e.getMessage());
}
}
@@ -2018,6 +2026,7 @@ public class HRegion implements HeapSize
this.nonce = nonce;
}
+ @Override
public Mutation getMutation(int index) {
return this.operations[index];
}
@@ -2776,6 +2785,16 @@ public class HRegion implements HeapSize
}
}
+ protected void checkReadsEnabled() throws IOException {
+ if (!this.writestate.readsEnabled) {
+ throw new IOException ("The region's reads are disabled. Cannot serve the request");
+ }
+ }
+
+ public void setReadsEnabled(boolean readsEnabled) {
+ this.writestate.setReadsEnabled(readsEnabled);
+ }
+
/**
* Add updates first to the hlog and then add values to memstore.
* Warning: Assumption is caller has lock on passed in row.
@@ -3885,7 +3904,7 @@ public class HRegion implements HeapSize
if (filter != null && filter.hasFilterRow()) {
filter.filterRowCells(results);
}
-
+
if (isEmptyRow || filterRow()) {
results.clear();
boolean moreRows = nextRow(currentRow, offset, length);
@@ -3953,7 +3972,7 @@ public class HRegion implements HeapSize
return filter != null && (!filter.hasFilterRow())
&& filter.filterRow();
}
-
+
private boolean filterRowKey(byte[] row, int offset, short length) throws IOException {
return filter != null
&& filter.filterRowKey(row, offset, length);
@@ -5625,7 +5644,7 @@ public class HRegion implements HeapSize
* modifies data. It has to be called just before a try.
* #closeRegionOperation needs to be called in the try's finally block
* Acquires a read lock and checks if the region is closing or closed.
- * @throws IOException
+ * @throws IOException
*/
public void startRegionOperation() throws IOException {
startRegionOperation(Operation.ANY);
@@ -5633,14 +5652,15 @@ public class HRegion implements HeapSize
/**
* @param op The operation is about to be taken on the region
- * @throws IOException
+ * @throws IOException
*/
protected void startRegionOperation(Operation op) throws IOException {
switch (op) {
- case INCREMENT:
- case APPEND:
- case GET:
+ case GET: // read operations
case SCAN:
+ checkReadsEnabled();
+ case INCREMENT: // write operations
+ case APPEND:
case SPLIT_REGION:
case MERGE_REGION:
case PUT:
@@ -5683,7 +5703,7 @@ public class HRegion implements HeapSize
/**
* Closes the lock. This needs to be called in the finally block corresponding
* to the try block of #startRegionOperation
- * @throws IOException
+ * @throws IOException
*/
public void closeRegionOperation() throws IOException {
closeRegionOperation(Operation.ANY);
Modified: hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java?rev=1571865&r1=1571864&r2=1571865&view=diff
==============================================================================
--- hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java (original)
+++ hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionFileSystem.java Tue Feb 25 23:31:30 2014
@@ -49,6 +49,7 @@ import org.apache.hadoop.hbase.fs.HFileS
import org.apache.hadoop.hbase.io.Reference;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.FSUtils;
+import org.apache.hadoop.hbase.util.ServerRegionReplicaUtil;
import org.apache.hadoop.hbase.util.Threads;
/**
@@ -72,10 +73,12 @@ public class HRegionFileSystem {
private static final String REGION_TEMP_DIR = ".tmp";
private final HRegionInfo regionInfo;
+ //regionInfo for interacting with FS (getting encodedName, etc)
+ private final HRegionInfo regionInfoForFs;
private final Configuration conf;
private final Path tableDir;
private final FileSystem fs;
-
+
/**
* In order to handle NN connectivity hiccups, one need to retry non-idempotent operation at the
* client level.
@@ -98,6 +101,7 @@ public class HRegionFileSystem {
this.conf = conf;
this.tableDir = tableDir;
this.regionInfo = regionInfo;
+ this.regionInfoForFs = ServerRegionReplicaUtil.getRegionInfoForFs(regionInfo);
this.hdfsClientRetriesNumber = conf.getInt("hdfs.client.retries.number",
DEFAULT_HDFS_CLIENT_RETRIES_NUMBER);
this.baseSleepBeforeRetries = conf.getInt("hdfs.client.sleep.before.retries",
@@ -121,7 +125,7 @@ public class HRegionFileSystem {
/** @return {@link Path} to the region directory. */
public Path getRegionDir() {
- return new Path(this.tableDir, this.regionInfo.getEncodedName());
+ return new Path(this.tableDir, this.regionInfoForFs.getEncodedName());
}
// ===========================================================================
@@ -203,6 +207,7 @@ public class HRegionFileSystem {
public boolean hasReferences(final String familyName) throws IOException {
FileStatus[] files = FSUtils.listStatus(fs, getStoreDir(familyName),
new PathFilter () {
+ @Override
public boolean accept(Path path) {
return StoreFileInfo.isReference(path);
}
@@ -249,14 +254,14 @@ public class HRegionFileSystem {
*/
public void deleteFamily(final String familyName) throws IOException {
// archive family store files
- HFileArchiver.archiveFamily(fs, conf, regionInfo, tableDir, Bytes.toBytes(familyName));
+ HFileArchiver.archiveFamily(fs, conf, regionInfoForFs, tableDir, Bytes.toBytes(familyName));
// delete the family folder
Path familyDir = getStoreDir(familyName);
if(fs.exists(familyDir) && !deleteDir(familyDir))
throw new IOException("Could not delete family " + familyName
- + " from FileSystem for region " + regionInfo.getRegionNameAsString() + "("
- + regionInfo.getEncodedName() + ")");
+ + " from FileSystem for region " + regionInfoForFs.getRegionNameAsString() + "("
+ + regionInfoForFs.getEncodedName() + ")");
}
/**
@@ -326,7 +331,7 @@ public class HRegionFileSystem {
Path storeDir = getStoreDir(familyName);
if(!fs.exists(storeDir) && !createDir(storeDir))
throw new IOException("Failed creating " + storeDir);
-
+
String name = buildPath.getName();
if (generateNewName) {
name = generateUniqueName((seqNum < 0) ? null : "_SeqId_" + seqNum + "_");
@@ -366,7 +371,7 @@ public class HRegionFileSystem {
*/
public void removeStoreFile(final String familyName, final Path filePath)
throws IOException {
- HFileArchiver.archiveStoreFile(this.conf, this.fs, this.regionInfo,
+ HFileArchiver.archiveStoreFile(this.conf, this.fs, this.regionInfoForFs,
this.tableDir, Bytes.toBytes(familyName), filePath);
}
@@ -378,7 +383,7 @@ public class HRegionFileSystem {
*/
public void removeStoreFiles(final String familyName, final Collection<StoreFile> storeFiles)
throws IOException {
- HFileArchiver.archiveStoreFiles(this.conf, this.fs, this.regionInfo,
+ HFileArchiver.archiveStoreFiles(this.conf, this.fs, this.regionInfoForFs,
this.tableDir, Bytes.toBytes(familyName), storeFiles);
}
@@ -528,16 +533,16 @@ public class HRegionFileSystem {
*/
Path splitStoreFile(final HRegionInfo hri, final String familyName,
final StoreFile f, final byte[] splitRow, final boolean top) throws IOException {
-
+
// Check whether the split row lies in the range of the store file
// If it is outside the range, return directly.
if (top) {
//check if larger than last key.
KeyValue splitKey = KeyValue.createFirstOnRow(splitRow);
- byte[] lastKey = f.createReader().getLastKey();
+ byte[] lastKey = f.createReader().getLastKey();
// If lastKey is null means storefile is empty.
if (lastKey == null) return null;
- if (f.getReader().getComparator().compareFlatKey(splitKey.getBuffer(),
+ if (f.getReader().getComparator().compareFlatKey(splitKey.getBuffer(),
splitKey.getKeyOffset(), splitKey.getKeyLength(), lastKey, 0, lastKey.length) > 0) {
return null;
}
@@ -547,14 +552,14 @@ public class HRegionFileSystem {
byte[] firstKey = f.createReader().getFirstKey();
// If firstKey is null means storefile is empty.
if (firstKey == null) return null;
- if (f.getReader().getComparator().compareFlatKey(splitKey.getBuffer(),
+ if (f.getReader().getComparator().compareFlatKey(splitKey.getBuffer(),
splitKey.getKeyOffset(), splitKey.getKeyLength(), firstKey, 0, firstKey.length) < 0) {
return null;
- }
+ }
}
-
+
f.getReader().close(true);
-
+
Path splitDir = new Path(getSplitsDir(hri), familyName);
// A reference to the bottom half of the hsf store file.
Reference r =
@@ -563,7 +568,7 @@ public class HRegionFileSystem {
// See REF_NAME_REGEX regex above. The referred-to regions name is
// up in the path of the passed in <code>f</code> -- parentdir is family,
// then the directory above is the region name.
- String parentRegionName = regionInfo.getEncodedName();
+ String parentRegionName = regionInfoForFs.getEncodedName();
// Write reference with same file id only with the other region name as
// suffix and into the new region location (under same family).
Path p = new Path(splitDir, f.getPath().getName() + "." + parentRegionName);
@@ -636,12 +641,12 @@ public class HRegionFileSystem {
Path referenceDir = new Path(new Path(mergedDir,
mergedRegion.getEncodedName()), familyName);
// A whole reference to the store file.
- Reference r = Reference.createTopReference(regionInfo.getStartKey());
+ Reference r = Reference.createTopReference(regionInfoForFs.getStartKey());
// Add the referred-to regions name as a dot separated suffix.
// See REF_NAME_REGEX regex above. The referred-to regions name is
// up in the path of the passed in <code>f</code> -- parentdir is family,
// then the directory above is the region name.
- String mergingRegionName = regionInfo.getEncodedName();
+ String mergingRegionName = regionInfoForFs.getEncodedName();
// Write reference with same file id only with the other region name as
// suffix and into the new region location (under same family).
Path p = new Path(referenceDir, f.getPath().getName() + "."
@@ -653,7 +658,7 @@ public class HRegionFileSystem {
* Commit a merged region, moving it from the merges temporary directory to
* the proper location in the filesystem.
* @param mergedRegionInfo merged region {@link HRegionInfo}
- * @throws IOException
+ * @throws IOException
*/
void commitMergedRegion(final HRegionInfo mergedRegionInfo) throws IOException {
Path regionDir = new Path(this.tableDir, mergedRegionInfo.getEncodedName());
@@ -731,7 +736,7 @@ public class HRegionFileSystem {
// pb version is much shorter -- we write now w/o the toString version -- so checking length
// only should be sufficient. I don't want to read the file every time to check if it pb
// serialized.
- byte[] content = getRegionInfoFileContent(regionInfo);
+ byte[] content = getRegionInfoFileContent(regionInfoForFs);
try {
Path regionInfoFile = new Path(getRegionDir(), REGION_INFO_FILE);
@@ -747,7 +752,7 @@ public class HRegionFileSystem {
throw new IOException("Unable to remove existing " + regionInfoFile);
}
} catch (FileNotFoundException e) {
- LOG.warn(REGION_INFO_FILE + " file not found for region: " + regionInfo.getEncodedName());
+ LOG.warn(REGION_INFO_FILE + " file not found for region: " + regionInfoForFs.getEncodedName());
}
// Write HRI to a file in case we need to recover hbase:meta
@@ -759,7 +764,7 @@ public class HRegionFileSystem {
* @param useTempDir indicate whether or not using the region .tmp dir for a safer file creation.
*/
private void writeRegionInfoOnFilesystem(boolean useTempDir) throws IOException {
- byte[] content = getRegionInfoFileContent(regionInfo);
+ byte[] content = getRegionInfoFileContent(regionInfoForFs);
writeRegionInfoOnFilesystem(content, useTempDir);
}
Modified: hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java?rev=1571865&r1=1571864&r2=1571865&view=diff
==============================================================================
--- hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java (original)
+++ hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java Tue Feb 25 23:31:30 2014
@@ -50,7 +50,6 @@ import java.util.concurrent.locks.Reentr
import javax.management.ObjectName;
-import com.google.protobuf.HBaseZeroCopyByteString;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
@@ -235,6 +234,7 @@ import org.cliffc.high_scale_lib.Counter
import com.google.protobuf.BlockingRpcChannel;
import com.google.protobuf.ByteString;
+import com.google.protobuf.HBaseZeroCopyByteString;
import com.google.protobuf.Message;
import com.google.protobuf.RpcController;
import com.google.protobuf.ServiceException;
@@ -482,6 +482,9 @@ public class HRegionServer implements Cl
*/
private final int scannerLeaseTimeoutPeriod;
+ // chore for refreshing store files for secondary regions
+ private StorefileRefresherChore storefileRefresher;
+
/**
* The reference to the priority extraction function
*/
@@ -821,6 +824,12 @@ public class HRegionServer implements Cl
this.isa.getAddress(), 0));
this.pauseMonitor = new JvmPauseMonitor(conf);
pauseMonitor.start();
+
+ int storefileRefreshPeriod = conf.getInt(StorefileRefresherChore.REGIONSERVER_STOREFILE_REFRESH_PERIOD
+ , StorefileRefresherChore.DEFAULT_REGIONSERVER_STOREFILE_REFRESH_PERIOD);
+ if (storefileRefreshPeriod > 0) {
+ this.storefileRefresher = new StorefileRefresherChore(storefileRefreshPeriod, this, this);
+ }
}
/**
@@ -946,6 +955,9 @@ public class HRegionServer implements Cl
if (this.nonceManagerChore != null) {
this.nonceManagerChore.interrupt();
}
+ if (this.storefileRefresher != null) {
+ this.storefileRefresher.interrupt();
+ }
// Stop the snapshot and other procedure handlers, forcefully killing all running tasks
rspmHost.stop(this.abortRequested || this.killed);
@@ -1608,6 +1620,10 @@ public class HRegionServer implements Cl
Threads.setDaemonThreadRunning(this.nonceManagerChore.getThread(), n + ".nonceCleaner",
uncaughtExceptionHandler);
}
+ if (this.storefileRefresher != null) {
+ Threads.setDaemonThreadRunning(this.storefileRefresher.getThread(), n + ".storefileRefresher",
+ uncaughtExceptionHandler);
+ }
// Leases is not a Thread. Internally it runs a daemon thread. If it gets
// an unhandled exception, it will just exit.
@@ -1894,6 +1910,9 @@ public class HRegionServer implements Cl
this.replicationSinkHandler.stopReplicationService();
}
}
+ if (this.storefileRefresher != null) {
+ Threads.shutdown(this.storefileRefresher.getThread());
+ }
}
/**
Modified: hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java?rev=1571865&r1=1571864&r2=1571865&view=diff
==============================================================================
--- hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java (original)
+++ hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java Tue Feb 25 23:31:30 2014
@@ -26,6 +26,8 @@ import java.security.KeyException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.NavigableSet;
@@ -90,6 +92,7 @@ import com.google.common.base.Preconditi
import com.google.common.collect.ImmutableCollection;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
/**
* A Store holds a column family in a Region. Its a memstore and a set of zero
@@ -474,10 +477,13 @@ public class HStore implements Store {
*/
private List<StoreFile> loadStoreFiles() throws IOException {
Collection<StoreFileInfo> files = fs.getStoreFiles(getColumnFamilyName());
+ return openStoreFiles(files);
+ }
+
+ private List<StoreFile> openStoreFiles(Collection<StoreFileInfo> files) throws IOException {
if (files == null || files.size() == 0) {
return new ArrayList<StoreFile>();
}
-
// initialize the thread pool for opening store files in parallel..
ThreadPoolExecutor storeFileOpenerThreadPool =
this.region.getStoreFileOpenAndCloseThreadPool("StoreFileOpenerThread-" +
@@ -527,7 +533,7 @@ public class HStore implements Store {
for (StoreFile file : results) {
try {
if (file != null) file.closeReader(true);
- } catch (IOException e) {
+ } catch (IOException e) {
LOG.warn(e.getMessage());
}
}
@@ -537,6 +543,53 @@ public class HStore implements Store {
return results;
}
+ /**
+ * Checks the underlying store files, and opens the files that have not
+ * been opened, and removes the store file readers for store files no longer
+ * available. Mainly used by secondary region replicas to keep up to date with
+ * the primary region files.
+ * @throws IOException
+ */
+ @Override
+ public void refreshStoreFiles() throws IOException {
+ StoreFileManager sfm = storeEngine.getStoreFileManager();
+ Collection<StoreFile> currentFiles = sfm.getStorefiles();
+ if (currentFiles == null) currentFiles = new ArrayList<StoreFile>(0);
+
+ Collection<StoreFileInfo> newFiles = fs.getStoreFiles(getColumnFamilyName());
+ if (newFiles == null) newFiles = new ArrayList<StoreFileInfo>(0);
+
+ HashMap<StoreFileInfo, StoreFile> currentFilesSet = new HashMap<StoreFileInfo, StoreFile>(currentFiles.size());
+ for (StoreFile sf : currentFiles) {
+ currentFilesSet.put(sf.getFileInfo(), sf);
+ }
+ HashSet<StoreFileInfo> newFilesSet = new HashSet<StoreFileInfo>(newFiles);
+
+ Set<StoreFileInfo> toBeAddedFiles = Sets.difference(newFilesSet, currentFilesSet.keySet());
+ Set<StoreFileInfo> toBeRemovedFiles = Sets.difference(currentFilesSet.keySet(), newFilesSet);
+
+ if (toBeAddedFiles.isEmpty() && toBeRemovedFiles.isEmpty()) {
+ return;
+ }
+
+ LOG.info("Refreshing store files for region " + this.getRegionInfo().getRegionNameAsString()
+ + " files to add: " + toBeAddedFiles + " files to remove: " + toBeRemovedFiles);
+
+ Set<StoreFile> toBeRemovedStoreFiles = new HashSet<StoreFile>(toBeRemovedFiles.size());
+ for (StoreFileInfo sfi : toBeRemovedFiles) {
+ toBeRemovedStoreFiles.add(currentFilesSet.get(sfi));
+ }
+
+ // try to open the files
+ List<StoreFile> openedFiles = openStoreFiles(toBeAddedFiles);
+
+ // propogate the file changes to the underlying store file manager
+ replaceStoreFiles(toBeRemovedStoreFiles, openedFiles); //won't throw an exception
+
+ // notify scanners, close file readers, and recompute store size
+ completeCompaction(toBeRemovedStoreFiles, false);
+ }
+
private StoreFile createStoreFileAndReader(final Path p) throws IOException {
StoreFileInfo info = new StoreFileInfo(conf, this.getFileSystem(), p);
info.setRegionCoprocessorHost(this.region.getCoprocessorHost());
@@ -1094,7 +1147,7 @@ public class HStore implements Store {
writeCompactionWalRecord(filesToCompact, sfs);
replaceStoreFiles(filesToCompact, sfs);
// At this point the store will use new files for all new scanners.
- completeCompaction(filesToCompact); // Archive old files & update store size.
+ completeCompaction(filesToCompact, true); // Archive old files & update store size.
} finally {
finishCompactionRequest(cr);
}
@@ -1148,7 +1201,8 @@ public class HStore implements Store {
this.region.getRegionInfo(), compactionDescriptor, this.region.getSequenceId());
}
- private void replaceStoreFiles(final Collection<StoreFile> compactedFiles,
+ @VisibleForTesting
+ void replaceStoreFiles(final Collection<StoreFile> compactedFiles,
final Collection<StoreFile> result) throws IOException {
this.lock.writeLock().lock();
try {
@@ -1301,7 +1355,7 @@ public class HStore implements Store {
this.getCoprocessorHost().postCompact(this, sf, null);
}
replaceStoreFiles(filesToCompact, Lists.newArrayList(sf));
- completeCompaction(filesToCompact);
+ completeCompaction(filesToCompact, true);
}
} finally {
synchronized (filesCompacting) {
@@ -1466,7 +1520,7 @@ public class HStore implements Store {
}
}
- /*
+ /**
* <p>It works by processing a compaction that's been written to disk.
*
* <p>It is usually invoked at the end of a compaction, but might also be
@@ -1483,6 +1537,28 @@ public class HStore implements Store {
*/
@VisibleForTesting
protected void completeCompaction(final Collection<StoreFile> compactedFiles)
+ throws IOException {
+ completeCompaction(compactedFiles, true);
+ }
+
+
+ /**
+ * <p>It works by processing a compaction that's been written to disk.
+ *
+ * <p>It is usually invoked at the end of a compaction, but might also be
+ * invoked at HStore startup, if the prior execution died midway through.
+ *
+ * <p>Moving the compacted TreeMap into place means:
+ * <pre>
+ * 1) Unload all replaced StoreFile, close and collect list to delete.
+ * 2) Compute new store size
+ * </pre>
+ *
+ * @param compactedFiles list of files that were compacted
+ * @param newFile StoreFile that is the result of the compaction
+ */
+ @VisibleForTesting
+ protected void completeCompaction(final Collection<StoreFile> compactedFiles, boolean removeFiles)
throws IOException {
try {
// Do not delete old store files until we have sent out notification of
@@ -1497,7 +1573,9 @@ public class HStore implements Store {
for (StoreFile compactedFile : compactedFiles) {
compactedFile.closeReader(true);
}
- this.fs.removeStoreFiles(this.getColumnFamilyName(), compactedFiles);
+ if (removeFiles) {
+ this.fs.removeStoreFiles(this.getColumnFamilyName(), compactedFiles);
+ }
} catch (IOException e) {
e = RemoteExceptionHandler.checkIOException(e);
LOG.error("Failed removing compacted files in " + this +
Modified: hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java?rev=1571865&r1=1571864&r2=1571865&view=diff
==============================================================================
--- hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java (original)
+++ hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Store.java Tue Feb 25 23:31:30 2014
@@ -27,10 +27,10 @@ import org.apache.hadoop.classification.
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.HeapSize;
import org.apache.hadoop.hbase.io.compress.Compression;
@@ -349,4 +349,13 @@ public interface Store extends HeapSize,
* @return Whether this store has too many store files.
*/
boolean hasTooManyStoreFiles();
+
+ /**
+ * Checks the underlying store files, and opens the files that have not
+ * been opened, and removes the store file readers for store files no longer
+ * available. Mainly used by secondary region replicas to keep up to date with
+ * the primary region files.
+ * @throws IOException
+ */
+ void refreshStoreFiles() throws IOException;
}
Modified: hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileInfo.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileInfo.java?rev=1571865&r1=1571864&r2=1571865&view=diff
==============================================================================
--- hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileInfo.java (original)
+++ hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreFileInfo.java Tue Feb 25 23:31:30 2014
@@ -43,7 +43,7 @@ import org.apache.hadoop.hbase.util.FSUt
* Describe a StoreFile (hfile, reference, link)
*/
@InterfaceAudience.Private
-public class StoreFileInfo {
+public class StoreFileInfo implements Comparable<StoreFileInfo> {
public static final Log LOG = LogFactory.getLog(StoreFileInfo.class);
/**
@@ -390,4 +390,27 @@ public class StoreFileInfo {
}
return FSUtils.computeHDFSBlocksDistribution(fs, status, start, length);
}
+
+ @Override
+ public boolean equals(Object that) {
+ if (that == null) {
+ return false;
+ }
+
+ if (that instanceof StoreFileInfo) {
+ return this.compareTo((StoreFileInfo)that) == 0;
+ }
+
+ return false;
+ };
+
+ @Override
+ public int compareTo(StoreFileInfo o) {
+ return this.fileStatus.compareTo(o.fileStatus);
+ }
+
+ @Override
+ public int hashCode() {
+ return this.fileStatus.hashCode();
+ }
}
Added: hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StorefileRefresherChore.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StorefileRefresherChore.java?rev=1571865&view=auto
==============================================================================
--- hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StorefileRefresherChore.java (added)
+++ hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StorefileRefresherChore.java Tue Feb 25 23:31:30 2014
@@ -0,0 +1,118 @@
+/**
+ * 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.regionserver;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.Chore;
+import org.apache.hadoop.hbase.Stoppable;
+import org.apache.hadoop.hbase.master.cleaner.TimeToLiveHFileCleaner;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.util.StringUtils;
+
+/**
+ * A chore for refreshing the store files for secondary regions hosted in the region server.
+ *
+ * This chore should run periodically with a shorter interval than HFile TTL
+ * ("hbase.master.hfilecleaner.ttl", default 5 minutes).
+ * It ensures that if we cannot refresh files longer than that amount, the region
+ * will stop serving read requests because the referenced files might have been deleted (by the
+ * primary region).
+ */
+public class StorefileRefresherChore extends Chore {
+
+ private static final Log LOG = LogFactory.getLog(StorefileRefresherChore.class);
+
+ /**
+ * The period (in milliseconds) for refreshing the store files for the secondary regions.
+ */
+ static final String REGIONSERVER_STOREFILE_REFRESH_PERIOD
+ = "hbase.regionserver.storefile.refresh.period";
+ static final int DEFAULT_REGIONSERVER_STOREFILE_REFRESH_PERIOD = 0; //disabled by default
+
+ private HRegionServer regionServer;
+ private long hfileTtl;
+ private int period;
+
+ //ts of last time regions store files are refreshed
+ private Map<String, Long> lastRefreshTimes; // encodedName -> long
+
+ public StorefileRefresherChore(int period, HRegionServer regionServer, Stoppable stoppable) {
+ super("StorefileRefresherChore", period, stoppable);
+ this.period = period;
+ this.regionServer = regionServer;
+ this.hfileTtl = this.regionServer.getConfiguration().getLong(
+ TimeToLiveHFileCleaner.TTL_CONF_KEY, TimeToLiveHFileCleaner.DEFAULT_TTL);
+ if (period > hfileTtl / 2) {
+ throw new RuntimeException(REGIONSERVER_STOREFILE_REFRESH_PERIOD +
+ " should be set smaller than half of " + TimeToLiveHFileCleaner.TTL_CONF_KEY);
+ }
+ lastRefreshTimes = new HashMap<String, Long>();
+ }
+
+ @Override
+ protected void chore() {
+ for (HRegion r : regionServer.getOnlineRegionsLocalContext()) {
+ if (!r.writestate.isReadOnly()) {
+ // skip checking for this region if it can accept writes
+ continue;
+ }
+ String encodedName = r.getRegionInfo().getEncodedName();
+ long time = EnvironmentEdgeManager.currentTimeMillis();
+ if (!lastRefreshTimes.containsKey(encodedName)) {
+ lastRefreshTimes.put(encodedName, time);
+ }
+ try {
+ for (Store store : r.getStores().values()) {
+ // TODO: some stores might see new data from flush, while others do not which
+ // MIGHT break atomic edits across column families. We can fix this with setting
+ // mvcc read numbers that we know every store has seen
+ store.refreshStoreFiles();
+ }
+ } catch (IOException ex) {
+ LOG.warn("Exception while trying to refresh store files for region:" + r.getRegionInfo()
+ + ", exception:" + StringUtils.stringifyException(ex));
+
+ // Store files have a TTL in the archive directory. If we fail to refresh for that long, we stop serving reads
+ if (isRegionStale(encodedName, time)) {
+ r.setReadsEnabled(false); // stop serving reads
+ }
+ continue;
+ }
+ lastRefreshTimes.put(encodedName, time);
+ r.setReadsEnabled(true); // restart serving reads
+ }
+
+ // remove closed regions
+ for (String encodedName : lastRefreshTimes.keySet()) {
+ if (regionServer.getFromOnlineRegions(encodedName) == null) {
+ lastRefreshTimes.remove(encodedName);
+ }
+ }
+ }
+
+ protected boolean isRegionStale(String encodedName, long time) {
+ long lastRefreshTime = lastRefreshTimes.get(encodedName);
+ return time - lastRefreshTime > hfileTtl - period;
+ }
+}
Added: hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/util/ServerRegionReplicaUtil.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/util/ServerRegionReplicaUtil.java?rev=1571865&view=auto
==============================================================================
--- hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/util/ServerRegionReplicaUtil.java (added)
+++ hbase/branches/hbase-10070/hbase-server/src/main/java/org/apache/hadoop/hbase/util/ServerRegionReplicaUtil.java Tue Feb 25 23:31:30 2014
@@ -0,0 +1,52 @@
+/**
+ * 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.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.client.RegionReplicaUtil;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+
+/**
+ * Similar to {@link RegionReplicaUtil} but for the server side
+ */
+public class ServerRegionReplicaUtil extends RegionReplicaUtil {
+
+ /**
+ * Returns the regionInfo object to use for interacting with the file system.
+ * @return An HRegionInfo object to interact with the filesystem
+ */
+ public static HRegionInfo getRegionInfoForFs(HRegionInfo regionInfo) {
+ if (regionInfo == null) {
+ return null;
+ }
+ return RegionReplicaUtil.getRegionInfoForDefaultReplica(regionInfo);
+ }
+
+ /**
+ * Returns whether this region replica can accept writes.
+ * @param region the HRegion object
+ * @return whether the replica is read only
+ */
+ public static boolean isReadOnly(HRegion region) {
+ return region.getTableDesc().isReadOnly()
+ || !isDefaultReplica(region.getRegionInfo());
+ }
+
+
+}
Modified: hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/HBaseTestingUtility.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/HBaseTestingUtility.java?rev=1571865&r1=1571864&r2=1571865&view=diff
==============================================================================
--- hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/HBaseTestingUtility.java (original)
+++ hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/HBaseTestingUtility.java Tue Feb 25 23:31:30 2014
@@ -1770,6 +1770,15 @@ public class HBaseTestingUtility extends
}
}
+ public void deleteNumericRows(final HTable t, final byte[] f, int startRow, int endRow) throws IOException {
+ for (int i = startRow; i < endRow; i++) {
+ byte[] data = Bytes.toBytes(String.valueOf(i));
+ Delete delete = new Delete(data);
+ delete.deleteFamily(f);
+ t.delete(delete);
+ }
+ }
+
/**
* Return the number of rows in the given table.
*/
Modified: hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java?rev=1571865&r1=1571864&r2=1571865&view=diff
==============================================================================
--- hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java (original)
+++ hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegion.java Tue Feb 25 23:31:30 2014
@@ -136,7 +136,7 @@ import com.google.common.collect.Lists;
/**
* Basic stand-alone testing of HRegion.
- *
+ *
* A lot of the meta information for an HRegion now lives inside other HRegions
* or in the HBaseMaster, so only basic testing is possible.
*/
@@ -151,7 +151,7 @@ public class TestHRegion {
private static final String COLUMN_FAMILY = "MyCF";
HRegion region = null;
- private static HBaseTestingUtility TEST_UTIL; // do not run unit tests in parallel
+ private static HBaseTestingUtility TEST_UTIL; // do not run unit tests in parallel
public static Configuration conf ;
private String DIR;
private static FileSystem fs;
@@ -191,7 +191,7 @@ public class TestHRegion {
String getName() {
return name.getMethodName();
}
-
+
// ////////////////////////////////////////////////////////////////////////////
// New tests that doesn't spin up a mini cluster but rather just test the
// individual code pieces in the HRegion. Putting files locally in
@@ -1971,7 +1971,7 @@ public class TestHRegion {
/**
* This method tests https://issues.apache.org/jira/browse/HBASE-2516.
- *
+ *
* @throws IOException
*/
@Test
@@ -2540,7 +2540,7 @@ public class TestHRegion {
/**
* Added for HBASE-5416
- *
+ *
* Here we test scan optimization when only subset of CFs are used in filter
* conditions.
*/
@@ -2609,7 +2609,7 @@ public class TestHRegion {
/**
* HBASE-5416
- *
+ *
* Test case when scan limits amount of KVs returned on each next() call.
*/
@Test
@@ -2703,7 +2703,7 @@ public class TestHRegion {
// ////////////////////////////////////////////////////////////////////////////
/**
* Splits twice and verifies getting from each of the split regions.
- *
+ *
* @throws Exception
*/
@Test
@@ -2835,7 +2835,7 @@ public class TestHRegion {
* Flushes the cache in a thread while scanning. The tests verify that the
* scan is coherent - e.g. the returned results are always of the same or
* later update as the previous results.
- *
+ *
* @throws IOException
* scan / compact
* @throws InterruptedException
@@ -2957,7 +2957,7 @@ public class TestHRegion {
/**
* Writes very wide records and scans for the latest every time.. Flushes and
* compacts the region every now and then to keep things realistic.
- *
+ *
* @throws IOException
* by flush / scan / compaction
* @throws InterruptedException
@@ -3115,7 +3115,7 @@ public class TestHRegion {
/**
* Writes very wide records and gets the latest row every time.. Flushes and
* compacts the region aggressivly to catch issues.
- *
+ *
* @throws IOException
* by flush / scan / compaction
* @throws InterruptedException
@@ -3500,7 +3500,7 @@ public class TestHRegion {
/**
* Testcase to check state of region initialization task set to ABORTED or not
* if any exceptions during initialization
- *
+ *
* @throws Exception
*/
@Test
@@ -3923,7 +3923,116 @@ public class TestHRegion {
region.close();
}
+ @Test
+ public void testRegionReplicaSecondary() throws IOException {
+ // create a primary region, load some data and flush
+ // create a secondary region, and do a get against that
+ Path rootDir = new Path(DIR + "testRegionReplicaSecondary");
+
+ byte[][] families = new byte[][] {
+ Bytes.toBytes("cf1"), Bytes.toBytes("cf2"), Bytes.toBytes("cf3")
+ };
+ byte[] cq = Bytes.toBytes("cq");
+ HTableDescriptor htd = new HTableDescriptor(TableName.valueOf("testRegionReplicaSecondary"));
+ for (byte[] family : families) {
+ htd.addFamily(new HColumnDescriptor(family));
+ }
+
+ long time = System.currentTimeMillis();
+ HRegionInfo primaryHri = new HRegionInfo(htd.getTableName(),
+ HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW,
+ false, time, 0);
+ HRegionInfo secondaryHri = new HRegionInfo(htd.getTableName(),
+ HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW,
+ false, time, 1);
+
+ HRegion primaryRegion = null, secondaryRegion = null;
+
+ try {
+ primaryRegion = HRegion.createHRegion(primaryHri,
+ rootDir, TEST_UTIL.getConfiguration(), htd);
+
+ // load some data
+ putData(primaryRegion, 0, 1000, cq, families);
+
+ // flush region
+ primaryRegion.flushcache();
+
+ // open secondary region
+ secondaryRegion = HRegion.openHRegion(rootDir, secondaryHri, htd, null, conf);
+
+ verifyData(secondaryRegion, 0, 1000, cq, families);
+ } finally {
+ if (primaryRegion != null) {
+ HRegion.closeHRegion(primaryRegion);
+ }
+ if (secondaryRegion != null) {
+ HRegion.closeHRegion(secondaryRegion);
+ }
+ }
+ }
+
+ @Test
+ public void testRegionReplicaSecondaryIsReadOnly() throws IOException {
+ // create a primary region, load some data and flush
+ // create a secondary region, and do a put against that
+ Path rootDir = new Path(DIR + "testRegionReplicaSecondary");
+
+ byte[][] families = new byte[][] {
+ Bytes.toBytes("cf1"), Bytes.toBytes("cf2"), Bytes.toBytes("cf3")
+ };
+ byte[] cq = Bytes.toBytes("cq");
+ HTableDescriptor htd = new HTableDescriptor(TableName.valueOf("testRegionReplicaSecondary"));
+ for (byte[] family : families) {
+ htd.addFamily(new HColumnDescriptor(family));
+ }
+
+ long time = System.currentTimeMillis();
+ HRegionInfo primaryHri = new HRegionInfo(htd.getTableName(),
+ HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW,
+ false, time, 0);
+ HRegionInfo secondaryHri = new HRegionInfo(htd.getTableName(),
+ HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW,
+ false, time, 1);
+
+ HRegion primaryRegion = null, secondaryRegion = null;
+
+ try {
+ primaryRegion = HRegion.createHRegion(primaryHri,
+ rootDir, TEST_UTIL.getConfiguration(), htd);
+
+ // load some data
+ putData(primaryRegion, 0, 1000, cq, families);
+
+ // flush region
+ primaryRegion.flushcache();
+
+ // open secondary region
+ secondaryRegion = HRegion.openHRegion(rootDir, secondaryHri, htd, null, conf);
+
+ try {
+ putData(secondaryRegion, 0, 1000, cq, families);
+ fail("Should have thrown exception");
+ } catch (IOException ex) {
+ // expected
+ }
+ } finally {
+ if (primaryRegion != null) {
+ HRegion.closeHRegion(primaryRegion);
+ }
+ if (secondaryRegion != null) {
+ HRegion.closeHRegion(secondaryRegion);
+ }
+ }
+
+ }
+
private void putData(int startRow, int numRows, byte[] qf, byte[]... families) throws IOException {
+ putData(this.region, startRow, numRows, qf, families);
+ }
+
+ private void putData(HRegion region,
+ int startRow, int numRows, byte[] qf, byte[]... families) throws IOException {
for (int i = startRow; i < startRow + numRows; i++) {
Put put = new Put(Bytes.toBytes("" + i));
put.setDurability(Durability.SKIP_WAL);
@@ -3966,13 +4075,13 @@ public class TestHRegion {
/*
* Assert first value in the passed region is <code>firstValue</code>.
- *
+ *
* @param r
- *
+ *
* @param fs
- *
+ *
* @param firstValue
- *
+ *
* @throws IOException
*/
private void assertScan(final HRegion r, final byte[] fs, final byte[] firstValue)
Added: hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionReplicas.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionReplicas.java?rev=1571865&view=auto
==============================================================================
--- hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionReplicas.java (added)
+++ hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionReplicas.java Tue Feb 25 23:31:30 2014
@@ -0,0 +1,299 @@
+/**
+ * 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.regionserver;
+
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.MediumTests;
+import org.apache.hadoop.hbase.NotServingRegionException;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.catalog.TestMetaReaderEditor;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.protobuf.RequestConverter;
+import org.apache.hadoop.hbase.protobuf.generated.AdminProtos;
+import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hadoop.hbase.zookeeper.ZKAssign;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import com.google.protobuf.ServiceException;
+
+/**
+ * Tests for region replicas. Sad that we cannot isolate these without bringing up a whole
+ * cluster. See {@link TestRegionServerNoMaster}.
+ */
+@Category(MediumTests.class)
+public class TestRegionReplicas {
+ private static final int NB_SERVERS = 1;
+ private static HTable table;
+ private static final byte[] row = "TestRegionReplicas".getBytes();
+
+ private static HRegionInfo hriPrimary;
+ private static HRegionInfo hriSecondary;
+
+ private static final HBaseTestingUtility HTU = new HBaseTestingUtility();
+ private static final byte[] f = HConstants.CATALOG_FAMILY;
+
+ @BeforeClass
+ public static void before() throws Exception {
+ HTU.startMiniCluster(NB_SERVERS);
+ final byte[] tableName = Bytes.toBytes(TestRegionReplicas.class.getSimpleName());
+
+ // Create table then get the single region for our new table.
+ table = HTU.createTable(tableName, f);
+
+ hriPrimary = table.getRegionLocation(row, false).getRegionInfo();
+
+ // mock a secondary region info to open
+ hriSecondary = new HRegionInfo(hriPrimary.getTable(), hriPrimary.getStartKey(),
+ hriPrimary.getEndKey(), hriPrimary.isSplit(), hriPrimary.getRegionId(), 1);
+
+ // No master
+ HTU.getHBaseCluster().getMaster().stopMaster();
+ }
+
+ @AfterClass
+ public static void afterClass() throws Exception {
+ table.close();
+ HTU.shutdownMiniCluster();
+ }
+
+ @After
+ public void after() throws Exception {
+ // Clean the state if the test failed before cleaning the znode
+ // It does not manage all bad failures, so if there are multiple failures, only
+ // the first one should be looked at.
+ ZKAssign.deleteNodeFailSilent(HTU.getZooKeeperWatcher(), hriPrimary);
+ }
+
+ private HRegionServer getRS() {
+ return HTU.getMiniHBaseCluster().getRegionServer(0);
+ }
+
+ private void openRegion(HRegionInfo hri) throws Exception {
+ ZKAssign.createNodeOffline(HTU.getZooKeeperWatcher(), hri, getRS().getServerName());
+ // first version is '0'
+ AdminProtos.OpenRegionRequest orr = RequestConverter.buildOpenRegionRequest(hri, 0, null);
+ AdminProtos.OpenRegionResponse responseOpen = getRS().openRegion(null, orr);
+ Assert.assertTrue(responseOpen.getOpeningStateCount() == 1);
+ Assert.assertTrue(responseOpen.getOpeningState(0).
+ equals(AdminProtos.OpenRegionResponse.RegionOpeningState.OPENED));
+ checkRegionIsOpened(hri.getEncodedName());
+ }
+
+ private void closeRegion(HRegionInfo hri) throws Exception {
+ ZKAssign.createNodeClosing(HTU.getZooKeeperWatcher(), hri, getRS().getServerName());
+
+ AdminProtos.CloseRegionRequest crr = RequestConverter.buildCloseRegionRequest(
+ hri.getEncodedName(), true);
+ AdminProtos.CloseRegionResponse responseClose = getRS().closeRegion(null, crr);
+ Assert.assertTrue(responseClose.getClosed());
+
+ checkRegionIsClosed(hri.getEncodedName());
+
+ ZKAssign.deleteClosedNode(HTU.getZooKeeperWatcher(), hri.getEncodedName(), getRS().getServerName());
+ }
+
+ private void checkRegionIsOpened(String encodedRegionName) throws Exception {
+
+ while (!getRS().getRegionsInTransitionInRS().isEmpty()) {
+ Thread.sleep(1);
+ }
+
+ Assert.assertTrue(getRS().getRegionByEncodedName(encodedRegionName).isAvailable());
+
+ Assert.assertTrue(
+ ZKAssign.deleteOpenedNode(HTU.getZooKeeperWatcher(), encodedRegionName, getRS().getServerName()));
+ }
+
+
+ private void checkRegionIsClosed(String encodedRegionName) throws Exception {
+
+ while (!getRS().getRegionsInTransitionInRS().isEmpty()) {
+ Thread.sleep(1);
+ }
+
+ try {
+ Assert.assertFalse(getRS().getRegionByEncodedName(encodedRegionName).isAvailable());
+ } catch (NotServingRegionException expected) {
+ // That's how it work: if the region is closed we have an exception.
+ }
+
+ // We don't delete the znode here, because there is not always a znode.
+ }
+
+ @Test(timeout = 60000)
+ public void testOpenRegionReplica() throws Exception {
+ openRegion(hriSecondary);
+ try {
+ //load some data to primary
+ HTU.loadNumericRows(table, f, 0, 1000);
+
+ // assert that we can read back from primary
+ Assert.assertEquals(1000, HTU.countRows(table));
+ } finally {
+ HTU.deleteNumericRows(table, f, 0, 1000);
+ closeRegion(hriSecondary);
+ }
+ }
+
+ /** Tests that the meta location is saved for secondary regions */
+ @Test(timeout = 60000)
+ public void testRegionReplicaUpdatesMetaLocation() throws Exception {
+ openRegion(hriSecondary);
+ HTable meta = null;
+ try {
+ meta = new HTable(HTU.getConfiguration(), TableName.META_TABLE_NAME);
+ TestMetaReaderEditor.assertMetaLocation(meta, hriPrimary.getRegionName()
+ , getRS().getServerName(), -1, 1, false);
+ } finally {
+ if (meta != null ) meta.close();
+ closeRegion(hriSecondary);
+ }
+ }
+
+ @Test(timeout = 60000)
+ public void testRegionReplicaGets() throws Exception {
+ try {
+ //load some data to primary
+ HTU.loadNumericRows(table, f, 0, 1000);
+ // assert that we can read back from primary
+ Assert.assertEquals(1000, HTU.countRows(table));
+ // flush so that region replica can read
+ HTU.getHBaseAdmin().flush(table.getTableName());
+
+ openRegion(hriSecondary);
+
+ // first try directly against region
+ HRegion region = getRS().getFromOnlineRegions(hriSecondary.getEncodedName());
+ assertGet(region, 42, true);
+
+ assertGetRpc(hriSecondary, 42, true);
+
+ } finally {
+ HTU.deleteNumericRows(table, HConstants.CATALOG_FAMILY, 0, 1000);
+ closeRegion(hriSecondary);
+ }
+ }
+
+ private void assertGet(HRegion region, int value, boolean expect) throws IOException {
+ byte[] row = Bytes.toBytes(String.valueOf(value));
+ Get get = new Get(row);
+ Result result = region.get(get);
+ if (expect) {
+ Assert.assertArrayEquals(row, result.getValue(f, null));
+ } else {
+ result.isEmpty();
+ }
+ }
+
+ // build a mock rpc
+ private void assertGetRpc(HRegionInfo info, int value, boolean expect) throws IOException, ServiceException {
+ byte[] row = Bytes.toBytes(String.valueOf(value));
+ Get get = new Get(row);
+ ClientProtos.GetRequest getReq = RequestConverter.buildGetRequest(info.getRegionName(), get);
+ ClientProtos.GetResponse getResp = getRS().get(null, getReq);
+ Result result = ProtobufUtil.toResult(getResp.getResult());
+ if (expect) {
+ Assert.assertArrayEquals(row, result.getValue(f, null));
+ } else {
+ result.isEmpty();
+ }
+ }
+
+ private void restartRegionServer() throws Exception {
+ afterClass();
+ before();
+ }
+
+ @Test(timeout = 300000)
+ public void testRefreshStoreFiles() throws Exception {
+ // enable store file refreshing
+ final int refreshPeriod = 2000; // 2 sec
+ HTU.getConfiguration().setInt("hbase.hstore.compactionThreshold", 100);
+ HTU.getConfiguration().setInt(StorefileRefresherChore.REGIONSERVER_STOREFILE_REFRESH_PERIOD, refreshPeriod);
+ // restart the region server so that it starts the refresher chore
+ restartRegionServer();
+
+ try {
+ openRegion(hriSecondary);
+
+ //load some data to primary
+ HTU.loadNumericRows(table, f, 0, 1000);
+ // assert that we can read back from primary
+ Assert.assertEquals(1000, HTU.countRows(table));
+ // flush so that region replica can read
+ HTU.getHBaseAdmin().flush(table.getTableName());
+
+ // ensure that chore is run
+ Threads.sleep(4 * refreshPeriod);
+
+ assertGetRpc(hriSecondary, 42, true);
+ assertGetRpc(hriSecondary, 1042, false);
+
+ //load some data to primary
+ HTU.loadNumericRows(table, f, 1000, 1100);
+ HTU.getHBaseAdmin().flush(table.getTableName());
+
+ HTU.loadNumericRows(table, f, 2000, 2100);
+ HTU.getHBaseAdmin().flush(table.getTableName());
+
+ // ensure that chore is run
+ Threads.sleep(4 * refreshPeriod);
+
+ assertGetRpc(hriSecondary, 42, true);
+ assertGetRpc(hriSecondary, 1042, true);
+ assertGetRpc(hriSecondary, 2042, true);
+
+ // ensure that we are see the 3 store files
+ HRegion secondaryRegion = getRS().getFromOnlineRegions(hriSecondary.getEncodedName());
+ Assert.assertEquals(3, secondaryRegion.getStore(f).getStorefilesCount());
+
+ // force compaction
+ HTU.compact(table.getName(), true);
+
+ long wakeUpTime = System.currentTimeMillis() + 4 * refreshPeriod;
+ while (System.currentTimeMillis() < wakeUpTime) {
+ assertGetRpc(hriSecondary, 42, true);
+ assertGetRpc(hriSecondary, 1042, true);
+ assertGetRpc(hriSecondary, 2042, true);
+ Threads.sleep(10);
+ }
+
+ // ensure that we see the compacted file only
+ Assert.assertEquals(1, secondaryRegion.getStore(f).getStorefilesCount());
+
+ } finally {
+ HTU.deleteNumericRows(table, HConstants.CATALOG_FAMILY, 0, 1000);
+ closeRegion(hriSecondary);
+ }
+ }
+}
Modified: hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStore.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStore.java?rev=1571865&r1=1571864&r2=1571865&view=diff
==============================================================================
--- hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStore.java (original)
+++ hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStore.java Tue Feb 25 23:31:30 2014
@@ -19,6 +19,11 @@
package org.apache.hadoop.hbase.regionserver;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
import java.io.IOException;
import java.lang.ref.SoftReference;
import java.security.PrivilegedExceptionAction;
@@ -78,6 +83,8 @@ import org.apache.hadoop.util.Progressab
import org.junit.experimental.categories.Category;
import org.mockito.Mockito;
+import com.google.common.collect.Lists;
+
/**
* Test class for the Store
*/
@@ -130,7 +137,7 @@ public class TestStore extends TestCase
}
private void init(String methodName) throws IOException {
- init(methodName, HBaseConfiguration.create());
+ init(methodName, TEST_UTIL.getConfiguration());
}
private void init(String methodName, Configuration conf)
@@ -203,7 +210,7 @@ public class TestStore extends TestCase
int ttl = 4;
IncrementingEnvironmentEdge edge = new IncrementingEnvironmentEdge();
EnvironmentEdgeManagerTestHelper.injectEdge(edge);
-
+
Configuration conf = HBaseConfiguration.create();
// Enable the expired store file deletion
conf.setBoolean("hbase.store.delete.expired.storefile", true);
@@ -258,7 +265,7 @@ public class TestStore extends TestCase
FileSystem fs = FileSystem.get(conf);
// Initialize region
init(getName(), conf);
-
+
int storeFileNum = 4;
for (int i = 1; i <= storeFileNum; i++) {
LOG.info("Adding some data for the store file #"+i);
@@ -278,12 +285,12 @@ public class TestStore extends TestCase
lowestTimeStampFromFS = getLowestTimeStampFromFS(fs, store.getStorefiles());
assertEquals(lowestTimeStampFromManager, lowestTimeStampFromFS);
}
-
- private static long getLowestTimeStampFromFS(FileSystem fs,
+
+ private static long getLowestTimeStampFromFS(FileSystem fs,
final Collection<StoreFile> candidates) throws IOException {
long minTs = Long.MAX_VALUE;
if (candidates.isEmpty()) {
- return minTs;
+ return minTs;
}
Path[] p = new Path[candidates.size()];
int i = 0;
@@ -291,7 +298,7 @@ public class TestStore extends TestCase
p[i] = sf.getPath();
++i;
}
-
+
FileStatus[] stats = fs.listStatus(p);
if (stats == null || stats.length == 0) {
return minTs;
@@ -645,6 +652,7 @@ public class TestStore extends TestCase
conf.setClass("fs.file.impl", FaultyFileSystem.class,
FileSystem.class);
user.runAs(new PrivilegedExceptionAction<Object>() {
+ @Override
public Object run() throws Exception {
// Make sure it worked (above is sensitive to caching details in hadoop core)
FileSystem fs = FileSystem.get(conf);
@@ -705,6 +713,7 @@ public class TestStore extends TestCase
overwrite, bufferSize, replication, blockSize, progress), faultPos);
}
+ @Override
public FSDataOutputStream createNonRecursive(Path f, boolean overwrite,
int bufferSize, short replication, long blockSize, Progressable progress)
throws IOException {
@@ -880,5 +889,103 @@ public class TestStore extends TestCase
init(this.getName(), conf);
assertEquals(DummyStoreEngine.lastCreatedCompactor, this.store.storeEngine.getCompactor());
}
+
+ private void addStoreFile() throws IOException {
+ StoreFile f = this.store.getStorefiles().iterator().next();
+ Path storedir = f.getPath().getParent();
+ long seqid = this.store.getMaxSequenceId(true);
+ Configuration c = TEST_UTIL.getConfiguration();
+ FileSystem fs = FileSystem.get(c);
+ HFileContext fileContext = new HFileContextBuilder().withBlockSize(BLOCKSIZE_SMALL).build();
+ StoreFile.Writer w = new StoreFile.WriterBuilder(c, new CacheConfig(c),
+ fs)
+ .withOutputDir(storedir)
+ .withFileContext(fileContext)
+ .build();
+ w.appendMetadata(seqid + 1, false);
+ w.close();
+ LOG.info("Added store file:" + w.getPath());
+ }
+
+ private void archiveStoreFile(int index) throws IOException {
+ Collection<StoreFile> files = this.store.getStorefiles();
+ StoreFile sf = null;
+ Iterator<StoreFile> it = files.iterator();
+ for (int i = 0; i <= index; i++) {
+ sf = it.next();
+ }
+ store.getRegionFileSystem().removeStoreFiles(store.getColumnFamilyName(), Lists.newArrayList(sf));
+ }
+
+ public void testRefreshStoreFiles() throws Exception {
+ init(this.getName());
+
+ assertEquals(0, this.store.getStorefilesCount());
+
+ // add some data, flush
+ this.store.add(new KeyValue(row, family, qf1, 1, (byte[])null));
+ flush(1);
+ assertEquals(1, this.store.getStorefilesCount());
+
+ // add one more file
+ addStoreFile();
+
+ assertEquals(1, this.store.getStorefilesCount());
+ store.refreshStoreFiles();
+ assertEquals(2, this.store.getStorefilesCount());
+
+ // add three more files
+ addStoreFile();
+ addStoreFile();
+ addStoreFile();
+
+ assertEquals(2, this.store.getStorefilesCount());
+ store.refreshStoreFiles();
+ assertEquals(5, this.store.getStorefilesCount());
+
+ archiveStoreFile(0);
+
+ assertEquals(5, this.store.getStorefilesCount());
+ store.refreshStoreFiles();
+ assertEquals(4, this.store.getStorefilesCount());
+
+ archiveStoreFile(0);
+ archiveStoreFile(1);
+ archiveStoreFile(2);
+
+ assertEquals(4, this.store.getStorefilesCount());
+ store.refreshStoreFiles();
+ assertEquals(1, this.store.getStorefilesCount());
+
+ archiveStoreFile(0);
+ store.refreshStoreFiles();
+ assertEquals(0, this.store.getStorefilesCount());
+ }
+
+ @SuppressWarnings("unchecked")
+ public void testRefreshStoreFilesNotChanged() throws IOException {
+ init(this.getName());
+
+ assertEquals(0, this.store.getStorefilesCount());
+
+ // add some data, flush
+ this.store.add(new KeyValue(row, family, qf1, 1, (byte[])null));
+ flush(1);
+ // add one more file
+ addStoreFile();
+
+ HStore spiedStore = spy(store);
+
+ // call first time after files changed
+ spiedStore.refreshStoreFiles();
+ assertEquals(2, this.store.getStorefilesCount());
+ verify(spiedStore, times(1)).replaceStoreFiles(any(Collection.class), any(Collection.class));
+
+ // call second time
+ spiedStore.refreshStoreFiles();
+
+ //ensure that replaceStoreFiles is not called if files are not refreshed
+ verify(spiedStore, times(0)).replaceStoreFiles(null, null);
+ }
}
Added: hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreFileRefresherChore.java
URL: http://svn.apache.org/viewvc/hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreFileRefresherChore.java?rev=1571865&view=auto
==============================================================================
--- hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreFileRefresherChore.java (added)
+++ hbase/branches/hbase-10070/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestStoreFileRefresherChore.java Tue Feb 25 23:31:30 2014
@@ -0,0 +1,209 @@
+/**
+ * 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.regionserver;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.SmallTests;
+import org.apache.hadoop.hbase.Stoppable;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Durability;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.StoppableImplementation;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+@Category(SmallTests.class)
+public class TestStoreFileRefresherChore {
+
+ private HBaseTestingUtility TEST_UTIL;
+ private Path testDir;
+
+ @Before
+ public void setUp() {
+ TEST_UTIL = new HBaseTestingUtility();
+ testDir = TEST_UTIL.getDataTestDir("TestStoreFileRefresherChore");
+ }
+
+ private HTableDescriptor getTableDesc(TableName tableName, byte[]... families) {
+ HTableDescriptor htd = new HTableDescriptor(tableName);
+ for (byte[] family : families) {
+ HColumnDescriptor hcd = new HColumnDescriptor(family);
+ // Set default to be three versions.
+ hcd.setMaxVersions(Integer.MAX_VALUE);
+ htd.addFamily(hcd);
+ }
+ return htd;
+ }
+
+ static class FailingHRegionFileSystem extends HRegionFileSystem {
+ boolean fail = false;
+ FailingHRegionFileSystem(Configuration conf, FileSystem fs, Path tableDir, HRegionInfo regionInfo) {
+ super(conf, fs, tableDir, regionInfo);
+ }
+
+ @Override
+ public Collection<StoreFileInfo> getStoreFiles(String familyName) throws IOException {
+ if (fail) {
+ throw new IOException("simulating FS failure");
+ }
+ return super.getStoreFiles(familyName);
+ }
+ }
+
+ private HRegion initHRegion(HTableDescriptor htd, byte[] startKey, byte[] stopKey, int replicaId) throws IOException {
+ Configuration conf = TEST_UTIL.getConfiguration();
+ Path tableDir = new Path(testDir, htd.getTableName().getNameAsString());
+
+ HRegionInfo info = new HRegionInfo(htd.getTableName(), startKey, stopKey, false, 0, replicaId);
+
+ HRegionFileSystem fs = new FailingHRegionFileSystem(conf, tableDir.getFileSystem(conf), tableDir, info);
+ HRegion region = new HRegion(fs, null, conf, htd, null);
+
+ region.initialize();
+
+ return region;
+ }
+
+ private void putData(HRegion region, int startRow, int numRows, byte[] qf, byte[]... families) throws IOException {
+ for (int i = startRow; i < startRow + numRows; i++) {
+ Put put = new Put(Bytes.toBytes("" + i));
+ put.setDurability(Durability.SKIP_WAL);
+ for (byte[] family : families) {
+ put.add(family, qf, null);
+ }
+ region.put(put);
+ }
+ }
+
+ private void verifyData(HRegion newReg, int startRow, int numRows, byte[] qf, byte[]... families)
+ throws IOException {
+ for (int i = startRow; i < startRow + numRows; i++) {
+ byte[] row = Bytes.toBytes("" + i);
+ Get get = new Get(row);
+ for (byte[] family : families) {
+ get.addColumn(family, qf);
+ }
+ Result result = newReg.get(get);
+ Cell[] raw = result.rawCells();
+ assertEquals(families.length, result.size());
+ for (int j = 0; j < families.length; j++) {
+ assertTrue(CellUtil.matchingRow(raw[j], row));
+ assertTrue(CellUtil.matchingFamily(raw[j], families[j]));
+ assertTrue(CellUtil.matchingQualifier(raw[j], qf));
+ }
+ }
+ }
+
+ static class StaleStorefileRefresherChore extends StorefileRefresherChore {
+ boolean isStale = false;
+ public StaleStorefileRefresherChore(int period, HRegionServer regionServer,
+ Stoppable stoppable) {
+ super(period, regionServer, stoppable);
+ }
+ @Override
+ protected boolean isRegionStale(String encodedName, long time) {
+ return isStale;
+ }
+ }
+
+ @Test (timeout = 60000)
+ public void testIsStale() throws IOException {
+ int period = 0;
+ byte[][] families = new byte[][] {Bytes.toBytes("cf")};
+ byte[] qf = Bytes.toBytes("cq");
+
+ HRegionServer regionServer = mock(HRegionServer.class);
+ List<HRegion> regions = new ArrayList<HRegion>();
+ when(regionServer.getOnlineRegionsLocalContext()).thenReturn(regions);
+ when(regionServer.getConfiguration()).thenReturn(TEST_UTIL.getConfiguration());
+
+ HTableDescriptor htd = getTableDesc(TableName.valueOf("testIsStale"), families);
+ HRegion primary = initHRegion(htd, HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW, 0);
+ HRegion replica1 = initHRegion(htd, HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW, 1);
+ regions.add(primary);
+ regions.add(replica1);
+
+ StaleStorefileRefresherChore chore = new StaleStorefileRefresherChore(period, regionServer, new StoppableImplementation());
+
+ // write some data to primary and flush
+ putData(primary, 0, 100, qf, families);
+ primary.flushcache();
+ verifyData(primary, 0, 100, qf, families);
+
+ try {
+ verifyData(replica1, 0, 100, qf, families);
+ Assert.fail("should have failed");
+ } catch(AssertionError ex) {
+ // expected
+ }
+ chore.chore();
+ verifyData(replica1, 0, 100, qf, families);
+
+ // simulate an fs failure where we cannot refresh the store files for the replica
+ ((FailingHRegionFileSystem)replica1.getRegionFileSystem()).fail = true;
+
+ // write some more data to primary and flush
+ putData(primary, 100, 100, qf, families);
+ primary.flushcache();
+ verifyData(primary, 0, 200, qf, families);
+
+ chore.chore(); // should not throw ex, but we cannot refresh the store files
+
+ verifyData(replica1, 0, 100, qf, families);
+ try {
+ verifyData(replica1, 100, 100, qf, families);
+ Assert.fail("should have failed");
+ } catch(AssertionError ex) {
+ // expected
+ }
+
+ chore.isStale = true;
+ chore.chore(); //now after this, we cannot read back any value
+ try {
+ verifyData(replica1, 0, 100, qf, families);
+ Assert.fail("should have failed with IOException");
+ } catch(IOException ex) {
+ // expected
+ }
+ }
+}