hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From zhang...@apache.org
Subject hbase git commit: HBASE-15537 Make multi WAL work with WALs other than FSHLog
Date Fri, 08 Apr 2016 02:44:52 GMT
Repository: hbase
Updated Branches:
  refs/heads/master 2dcd08bc3 -> 394b89d15


HBASE-15537 Make multi WAL work with WALs other than FSHLog


Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/394b89d1
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/394b89d1
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/394b89d1

Branch: refs/heads/master
Commit: 394b89d153a9bef67a84633f4ff68aff26d53832
Parents: 2dcd08b
Author: zhangduo <zhangduo@apache.org>
Authored: Wed Apr 6 17:04:28 2016 +0800
Committer: zhangduo <zhangduo@apache.org>
Committed: Fri Apr 8 10:36:16 2016 +0800

----------------------------------------------------------------------
 .../hbase/regionserver/wal/AbstractFSWAL.java   |   4 +
 .../hadoop/hbase/wal/AbstractFSWALProvider.java |   4 +-
 .../hbase/wal/RegionGroupingProvider.java       | 138 ++++++++-----------
 .../org/apache/hadoop/hbase/wal/WALFactory.java |  30 ++--
 ...ReplicationEndpointWithMultipleAsyncWAL.java |  36 +++++
 .../TestReplicationEndpointWithMultipleWAL.java |   2 +
 ...lMasterRSCompressedWithMultipleAsyncWAL.java |  37 +++++
 ...onKillMasterRSCompressedWithMultipleWAL.java |   2 +
 ...plicationSyncUpToolWithMultipleAsyncWAL.java |  37 +++++
 ...estReplicationSyncUpToolWithMultipleWAL.java |   2 +
 .../wal/TestBoundedRegionGroupingStrategy.java  | 131 ++++++++++--------
 11 files changed, 273 insertions(+), 150 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/394b89d1/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java
index b89488a..e4c4eb3 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/AbstractFSWAL.java
@@ -678,6 +678,10 @@ public abstract class AbstractFSWAL<W> implements WAL {
         // NewPath could be equal to oldPath if replaceWriter fails.
         newPath = replaceWriter(oldPath, newPath, nextWriter);
         tellListenersAboutPostLogRoll(oldPath, newPath);
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Create new " + getClass().getSimpleName() + " writer with pipeline:
"
+              + Arrays.toString(getPipeline()));
+        }
         // Can we delete any of the old log files?
         if (getNumRolledLogFiles() > 0) {
           cleanOldLogs();

http://git-wip-us.apache.org/repos/asf/hbase/blob/394b89d1/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractFSWALProvider.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractFSWALProvider.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractFSWALProvider.java
index 2f5c299..e495e99 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractFSWALProvider.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractFSWALProvider.java
@@ -82,7 +82,7 @@ public abstract class AbstractFSWALProvider<T extends AbstractFSWAL<?>>
implemen
    * @param factory factory that made us, identity used for FS layout. may not be null
    * @param conf may not be null
    * @param listeners may be null
-   * @param providerId differentiate between providers from one facotry, used for FS layout.
may be
+   * @param providerId differentiate between providers from one factory, used for FS layout.
may be
    *          null
    */
   @Override
@@ -109,7 +109,7 @@ public abstract class AbstractFSWALProvider<T extends AbstractFSWAL<?>>
implemen
   }
 
   @Override
-  public WAL getWAL(byte[] identifier, byte[] namespace) throws IOException {
+  public T getWAL(byte[] identifier, byte[] namespace) throws IOException {
     T walCopy = wal;
     if (walCopy == null) {
       // only lock when need to create wal, and need to lock since

http://git-wip-us.apache.org/repos/asf/hbase/blob/394b89d1/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/RegionGroupingProvider.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/RegionGroupingProvider.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/RegionGroupingProvider.java
index 0aeaccf..b447e94 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/RegionGroupingProvider.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/RegionGroupingProvider.java
@@ -18,34 +18,31 @@
  */
 package org.apache.hadoop.hbase.wal;
 
-import static org.apache.hadoop.hbase.wal.DefaultWALProvider.META_WAL_PROVIDER_ID;
-import static org.apache.hadoop.hbase.wal.DefaultWALProvider.WAL_FILE_NAME_DELIMITER;
+import static org.apache.hadoop.hbase.wal.AbstractFSWALProvider.META_WAL_PROVIDER_ID;
+import static org.apache.hadoop.hbase.wal.AbstractFSWALProvider.WAL_FILE_NAME_DELIMITER;
 
 import java.io.IOException;
 import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.locks.Lock;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.hbase.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.regionserver.wal.FSHLog;
-import org.apache.hadoop.hbase.regionserver.wal.MetricsWAL;
-
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
 // imports for classes still in regionserver.wal
 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
 import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.hbase.util.FSUtils;
+import org.apache.hadoop.hbase.util.IdReadWriteLock;
 
 /**
  * A WAL Provider that returns a WAL per group of regions.
  *
+ * This provider follows the decorator pattern and mainly holds the logic for WAL grouping.
+ * WAL creation/roll/close is delegated to {@link #DELEGATE_PROVIDER}
+ *
  * Region grouping is handled via {@link RegionGroupingStrategy} and can be configured via
the
  * property "hbase.wal.regiongrouping.strategy". Current strategy choices are
  * <ul>
@@ -57,7 +54,7 @@ import org.apache.hadoop.hbase.util.FSUtils;
  * Optionally, a FQCN to a custom implementation may be given.
  */
 @InterfaceAudience.Private
-class RegionGroupingProvider implements WALProvider {
+public class RegionGroupingProvider implements WALProvider {
   private static final Log LOG = LogFactory.getLog(RegionGroupingProvider.class);
 
   /**
@@ -124,22 +121,23 @@ class RegionGroupingProvider implements WALProvider {
   public static final String REGION_GROUPING_STRATEGY = "hbase.wal.regiongrouping.strategy";
   public static final String DEFAULT_REGION_GROUPING_STRATEGY = Strategies.defaultStrategy.name();
 
+  /** delegate provider for WAL creation/roll/close */
+  public static final String DELEGATE_PROVIDER = "hbase.wal.regiongrouping.delegate.provider";
+  public static final String DEFAULT_DELEGATE_PROVIDER = WALFactory.Providers.defaultProvider
+      .name();
+
   private static final String META_WAL_GROUP_NAME = "meta";
 
-  /** A group-wal mapping, recommended to make sure one-one rather than many-one mapping
*/
-  protected final Map<String, FSHLog> cached = new HashMap<String, FSHLog>();
-  /** Stores unique wals generated by this RegionGroupingProvider */
-  private final Set<FSHLog> logs = Collections.synchronizedSet(new HashSet<FSHLog>());
+  /** A group-provider mapping, make sure one-one rather than many-one mapping */
+  private final ConcurrentMap<String, WALProvider> cached = new ConcurrentHashMap<>();
 
-  /**
-   * we synchronize on walCacheLock to prevent wal recreation in different threads
-   */
-  final Object walCacheLock = new Object();
+  private final IdReadWriteLock createLock = new IdReadWriteLock();
 
-  protected RegionGroupingStrategy strategy = null;
+  private RegionGroupingStrategy strategy = null;
+  private WALFactory factory = null;
   private List<WALActionsListener> listeners = null;
   private String providerId = null;
-  private Configuration conf = null;
+  private Class<? extends WALProvider> providerClass;
 
   @Override
   public void init(final WALFactory factory, final Configuration conf,
@@ -147,6 +145,7 @@ class RegionGroupingProvider implements WALProvider {
     if (null != strategy) {
       throw new IllegalStateException("WALProvider.init should only be called once.");
     }
+    this.factory = factory;
     this.listeners = null == listeners ? null : Collections.unmodifiableList(listeners);
     StringBuilder sb = new StringBuilder().append(factory.factoryId);
     if (providerId != null) {
@@ -158,45 +157,33 @@ class RegionGroupingProvider implements WALProvider {
     }
     this.providerId = sb.toString();
     this.strategy = getStrategy(conf, REGION_GROUPING_STRATEGY, DEFAULT_REGION_GROUPING_STRATEGY);
-    this.conf = conf;
+    this.providerClass = factory.getProviderClass(DELEGATE_PROVIDER, DEFAULT_DELEGATE_PROVIDER);
   }
 
-  /**
-   * Populate the cache for this group.
-   */
-  FSHLog populateCache(String groupName) throws IOException {
-    boolean isMeta = META_WAL_PROVIDER_ID.equals(providerId);
-    String hlogPrefix;
-    List<WALActionsListener> listeners;
-    if (isMeta) {
-      hlogPrefix = this.providerId;
-      // don't watch log roll for meta
-      listeners = Collections.<WALActionsListener> singletonList(new MetricsWAL());
+  private WALProvider createProvider(String group) throws IOException {
+    if (META_WAL_PROVIDER_ID.equals(providerId)) {
+      return factory.createProvider(providerClass, listeners, META_WAL_PROVIDER_ID);
     } else {
-      hlogPrefix = groupName;
-      listeners = this.listeners;
+      return factory.createProvider(providerClass, listeners, group);
     }
-    FSHLog log = new FSHLog(FileSystem.get(conf), FSUtils.getRootDir(conf),
-        DefaultWALProvider.getWALDirectoryName(providerId), HConstants.HREGION_OLDLOGDIR_NAME,
-        conf, listeners, true, hlogPrefix, isMeta ? META_WAL_PROVIDER_ID : null);
-    cached.put(groupName, log);
-    logs.add(log);
-    return log;
   }
 
   private WAL getWAL(final String group) throws IOException {
-    WAL log = cached.get(group);
-    if (null == log) {
-      // only lock when need to create wal, and need to lock since
-      // creating hlog on fs is time consuming
-      synchronized (this.walCacheLock) {
-        log = cached.get(group);// check again
-        if (null == log) {
-          log = populateCache(group);
+    WALProvider provider = cached.get(group);
+    if (provider == null) {
+      Lock lock = createLock.getLock(group.hashCode()).writeLock();
+      lock.lock();
+      try {
+        provider = cached.get(group);
+        if (provider == null) {
+          provider = createProvider(group);
+          cached.put(group, provider);
         }
+      } finally {
+        lock.unlock();
       }
     }
-    return log;
+    return provider.getWAL(null, null);
   }
 
   @Override
@@ -214,15 +201,15 @@ class RegionGroupingProvider implements WALProvider {
   public void shutdown() throws IOException {
     // save the last exception and rethrow
     IOException failure = null;
-    synchronized (logs) {
-      for (FSHLog wal : logs) {
-        try {
-          wal.shutdown();
-        } catch (IOException exception) {
-          LOG.error("Problem shutting down log '" + wal + "': " + exception.getMessage());
-          LOG.debug("Details of problem shutting down log '" + wal + "'", exception);
-          failure = exception;
+    for (WALProvider provider: cached.values()) {
+      try {
+        provider.shutdown();
+      } catch (IOException e) {
+        LOG.error("Problem shutting down wal provider '" + provider + "': " + e.getMessage());
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Details of problem shutting down wal provider '" + provider + "'", e);
         }
+        failure = e;
       }
     }
     if (failure != null) {
@@ -234,15 +221,15 @@ class RegionGroupingProvider implements WALProvider {
   public void close() throws IOException {
     // save the last exception and rethrow
     IOException failure = null;
-    synchronized (logs) {
-      for (FSHLog wal : logs) {
-        try {
-          wal.close();
-        } catch (IOException exception) {
-          LOG.error("Problem closing log '" + wal + "': " + exception.getMessage());
-          LOG.debug("Details of problem closing wal '" + wal + "'", exception);
-          failure = exception;
+    for (WALProvider provider : cached.values()) {
+      try {
+        provider.close();
+      } catch (IOException e) {
+        LOG.error("Problem closing wal provider '" + provider + "': " + e.getMessage());
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Details of problem closing wal provider '" + provider + "'", e);
         }
+        failure = e;
       }
     }
     if (failure != null) {
@@ -262,10 +249,8 @@ class RegionGroupingProvider implements WALProvider {
   @Override
   public long getNumLogFiles() {
     long numLogFiles = 0;
-    synchronized (logs) {
-      for (FSHLog wal : logs) {
-        numLogFiles += wal.getNumLogFiles();
-      }
+    for (WALProvider provider : cached.values()) {
+      numLogFiles += provider.getNumLogFiles();
     }
     return numLogFiles;
   }
@@ -273,12 +258,9 @@ class RegionGroupingProvider implements WALProvider {
   @Override
   public long getLogFileSize() {
     long logFileSize = 0;
-    synchronized (logs) {
-      for (FSHLog wal : logs) {
-        logFileSize += wal.getLogFileSize();
-      }
+    for (WALProvider provider : cached.values()) {
+      logFileSize += provider.getLogFileSize();
     }
     return logFileSize;
   }
-
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/394b89d1/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALFactory.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALFactory.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALFactory.java
index a2761df..a9c17b5 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALFactory.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALFactory.java
@@ -127,38 +127,46 @@ public class WALFactory {
     factoryId = SINGLETON_ID;
   }
 
-  /**
-   * instantiate a provider from a config property.
-   * requires conf to have already been set (as well as anything the provider might need
to read).
-   */
-  WALProvider getProvider(final String key, final String defaultValue,
-      final List<WALActionsListener> listeners, final String providerId) throws IOException
{
-    Class<? extends WALProvider> clazz;
+  Class<? extends WALProvider> getProviderClass(String key, String defaultValue) {
     try {
-      clazz = Providers.valueOf(conf.get(key, defaultValue)).clazz;
+      return Providers.valueOf(conf.get(key, defaultValue)).clazz;
     } catch (IllegalArgumentException exception) {
       // Fall back to them specifying a class name
       // Note that the passed default class shouldn't actually be used, since the above only
fails
       // when there is a config value present.
-      clazz = conf.getClass(key, DefaultWALProvider.class, WALProvider.class);
+      return conf.getClass(key, DefaultWALProvider.class, WALProvider.class);
     }
+  }
+
+  WALProvider createProvider(Class<? extends WALProvider> clazz,
+      List<WALActionsListener> listeners, String providerId) throws IOException {
     LOG.info("Instantiating WALProvider of type " + clazz);
     try {
       final WALProvider result = clazz.newInstance();
       result.init(this, conf, listeners, providerId);
       return result;
     } catch (InstantiationException exception) {
-      LOG.error("couldn't set up WALProvider, check config key " + key);
+      LOG.error("couldn't set up WALProvider, the configured class is " + clazz);
       LOG.debug("Exception details for failure to load WALProvider.", exception);
       throw new IOException("couldn't set up WALProvider", exception);
     } catch (IllegalAccessException exception) {
-      LOG.error("couldn't set up WALProvider, check config key " + key);
+      LOG.error("couldn't set up WALProvider, the configured class is " + clazz);
       LOG.debug("Exception details for failure to load WALProvider.", exception);
       throw new IOException("couldn't set up WALProvider", exception);
     }
   }
 
   /**
+   * instantiate a provider from a config property.
+   * requires conf to have already been set (as well as anything the provider might need
to read).
+   */
+  WALProvider getProvider(final String key, final String defaultValue,
+      final List<WALActionsListener> listeners, final String providerId) throws IOException
{
+    Class<? extends WALProvider> clazz = getProviderClass(key, defaultValue);
+    return createProvider(clazz, listeners, providerId);
+  }
+
+  /**
    * @param conf must not be null, will keep a reference to read params in later reader/writer
    *     instances.
    * @param listeners may be null. will be given to all created wals (and not meta-wals)

http://git-wip-us.apache.org/repos/asf/hbase/blob/394b89d1/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationEndpointWithMultipleAsyncWAL.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationEndpointWithMultipleAsyncWAL.java
b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationEndpointWithMultipleAsyncWAL.java
new file mode 100644
index 0000000..debe8a1
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationEndpointWithMultipleAsyncWAL.java
@@ -0,0 +1,36 @@
+/**
+ * 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.replication.multiwal;
+
+import org.apache.hadoop.hbase.replication.TestReplicationEndpoint;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.testclassification.ReplicationTests;
+import org.apache.hadoop.hbase.wal.RegionGroupingProvider;
+import org.apache.hadoop.hbase.wal.WALFactory;
+import org.junit.BeforeClass;
+import org.junit.experimental.categories.Category;
+
+@Category({ ReplicationTests.class, MediumTests.class })
+public class TestReplicationEndpointWithMultipleAsyncWAL extends TestReplicationEndpoint
{
+  @BeforeClass
+  public static void setUpBeforeClass() throws Exception {
+    conf1.set(WALFactory.WAL_PROVIDER, "multiwal");
+    conf1.set(RegionGroupingProvider.DELEGATE_PROVIDER, "asyncfs");
+    TestReplicationEndpoint.setUpBeforeClass();
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/394b89d1/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationEndpointWithMultipleWAL.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationEndpointWithMultipleWAL.java
b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationEndpointWithMultipleWAL.java
index cd7873a..8f350f2 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationEndpointWithMultipleWAL.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationEndpointWithMultipleWAL.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hbase.replication.multiwal;
 import org.apache.hadoop.hbase.replication.TestReplicationEndpoint;
 import org.apache.hadoop.hbase.testclassification.MediumTests;
 import org.apache.hadoop.hbase.testclassification.ReplicationTests;
+import org.apache.hadoop.hbase.wal.RegionGroupingProvider;
 import org.apache.hadoop.hbase.wal.WALFactory;
 import org.junit.BeforeClass;
 import org.junit.experimental.categories.Category;
@@ -29,6 +30,7 @@ public class TestReplicationEndpointWithMultipleWAL extends TestReplicationEndpo
   @BeforeClass
   public static void setUpBeforeClass() throws Exception {
     conf1.set(WALFactory.WAL_PROVIDER, "multiwal");
+    conf1.set(RegionGroupingProvider.DELEGATE_PROVIDER, "filesystem");
     TestReplicationEndpoint.setUpBeforeClass();
   }
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/394b89d1/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationKillMasterRSCompressedWithMultipleAsyncWAL.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationKillMasterRSCompressedWithMultipleAsyncWAL.java
b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationKillMasterRSCompressedWithMultipleAsyncWAL.java
new file mode 100644
index 0000000..416cda7
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationKillMasterRSCompressedWithMultipleAsyncWAL.java
@@ -0,0 +1,37 @@
+/**
+ * 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.replication.multiwal;
+
+import org.apache.hadoop.hbase.replication.TestReplicationKillMasterRSCompressed;
+import org.apache.hadoop.hbase.testclassification.LargeTests;
+import org.apache.hadoop.hbase.testclassification.ReplicationTests;
+import org.apache.hadoop.hbase.wal.RegionGroupingProvider;
+import org.apache.hadoop.hbase.wal.WALFactory;
+import org.junit.BeforeClass;
+import org.junit.experimental.categories.Category;
+
+@Category({ReplicationTests.class, LargeTests.class})
+public class TestReplicationKillMasterRSCompressedWithMultipleAsyncWAL extends
+    TestReplicationKillMasterRSCompressed {
+  @BeforeClass
+  public static void setUpBeforeClass() throws Exception {
+    conf1.set(WALFactory.WAL_PROVIDER, "multiwal");
+    conf1.set(RegionGroupingProvider.DELEGATE_PROVIDER, "asyncfs");
+    TestReplicationKillMasterRSCompressed.setUpBeforeClass();
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/394b89d1/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationKillMasterRSCompressedWithMultipleWAL.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationKillMasterRSCompressedWithMultipleWAL.java
b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationKillMasterRSCompressedWithMultipleWAL.java
index 0b2c2cc..1806776 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationKillMasterRSCompressedWithMultipleWAL.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationKillMasterRSCompressedWithMultipleWAL.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hbase.replication.multiwal;
 import org.apache.hadoop.hbase.replication.TestReplicationKillMasterRSCompressed;
 import org.apache.hadoop.hbase.testclassification.LargeTests;
 import org.apache.hadoop.hbase.testclassification.ReplicationTests;
+import org.apache.hadoop.hbase.wal.RegionGroupingProvider;
 import org.apache.hadoop.hbase.wal.WALFactory;
 import org.junit.BeforeClass;
 import org.junit.experimental.categories.Category;
@@ -30,6 +31,7 @@ public class TestReplicationKillMasterRSCompressedWithMultipleWAL extends
   @BeforeClass
   public static void setUpBeforeClass() throws Exception {
     conf1.set(WALFactory.WAL_PROVIDER, "multiwal");
+    conf1.set(RegionGroupingProvider.DELEGATE_PROVIDER, "filesystem");
     TestReplicationKillMasterRSCompressed.setUpBeforeClass();
   }
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/394b89d1/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationSyncUpToolWithMultipleAsyncWAL.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationSyncUpToolWithMultipleAsyncWAL.java
b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationSyncUpToolWithMultipleAsyncWAL.java
new file mode 100644
index 0000000..bb91aaf
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationSyncUpToolWithMultipleAsyncWAL.java
@@ -0,0 +1,37 @@
+/**
+ * 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.replication.multiwal;
+
+import org.apache.hadoop.hbase.replication.TestReplicationBase;
+import org.apache.hadoop.hbase.replication.TestReplicationSyncUpTool;
+import org.apache.hadoop.hbase.testclassification.LargeTests;
+import org.apache.hadoop.hbase.testclassification.ReplicationTests;
+import org.apache.hadoop.hbase.wal.RegionGroupingProvider;
+import org.apache.hadoop.hbase.wal.WALFactory;
+import org.junit.BeforeClass;
+import org.junit.experimental.categories.Category;
+
+@Category({ ReplicationTests.class, LargeTests.class })
+public class TestReplicationSyncUpToolWithMultipleAsyncWAL extends TestReplicationSyncUpTool
{
+  @BeforeClass
+  public static void setUpBeforeClass() throws Exception {
+    conf1.set(WALFactory.WAL_PROVIDER, "multiwal");
+    conf1.set(RegionGroupingProvider.DELEGATE_PROVIDER, "asyncfs");
+    TestReplicationBase.setUpBeforeClass();
+  }
+}

http://git-wip-us.apache.org/repos/asf/hbase/blob/394b89d1/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationSyncUpToolWithMultipleWAL.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationSyncUpToolWithMultipleWAL.java
b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationSyncUpToolWithMultipleWAL.java
index a8e455c..7329a90 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationSyncUpToolWithMultipleWAL.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/multiwal/TestReplicationSyncUpToolWithMultipleWAL.java
@@ -21,6 +21,7 @@ import org.apache.hadoop.hbase.replication.TestReplicationBase;
 import org.apache.hadoop.hbase.replication.TestReplicationSyncUpTool;
 import org.apache.hadoop.hbase.testclassification.LargeTests;
 import org.apache.hadoop.hbase.testclassification.ReplicationTests;
+import org.apache.hadoop.hbase.wal.RegionGroupingProvider;
 import org.apache.hadoop.hbase.wal.WALFactory;
 import org.junit.BeforeClass;
 import org.junit.experimental.categories.Category;
@@ -30,6 +31,7 @@ public class TestReplicationSyncUpToolWithMultipleWAL extends TestReplicationSyn
   @BeforeClass
   public static void setUpBeforeClass() throws Exception {
     conf1.set(WALFactory.WAL_PROVIDER, "multiwal");
+    conf1.set(RegionGroupingProvider.DELEGATE_PROVIDER, "filesystem");
     TestReplicationBase.setUpBeforeClass();
   }
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/394b89d1/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestBoundedRegionGroupingStrategy.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestBoundedRegionGroupingStrategy.java
b/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestBoundedRegionGroupingStrategy.java
index 2044f82..8523e69 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestBoundedRegionGroupingStrategy.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestBoundedRegionGroupingStrategy.java
@@ -18,80 +18,92 @@
  */
 package org.apache.hadoop.hbase.wal;
 
+import static org.apache.hadoop.hbase.wal.BoundedGroupingStrategy.DEFAULT_NUM_REGION_GROUPS;
+import static org.apache.hadoop.hbase.wal.BoundedGroupingStrategy.NUM_REGION_GROUPS;
+import static org.apache.hadoop.hbase.wal.RegionGroupingProvider.*;
+import static org.apache.hadoop.hbase.wal.WALFactory.WAL_PROVIDER;
+import static org.junit.Assert.assertEquals;
+
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Random;
 import java.util.Set;
 
-import static org.junit.Assert.assertEquals;
-import static org.apache.hadoop.hbase.wal.BoundedGroupingStrategy.NUM_REGION_GROUPS;
-import static org.apache.hadoop.hbase.wal.BoundedGroupingStrategy.DEFAULT_NUM_REGION_GROUPS;
-import static org.apache.hadoop.hbase.wal.WALFactory.WAL_PROVIDER;
-import static org.apache.hadoop.hbase.wal.RegionGroupingProvider.REGION_GROUPING_STRATEGY;
-
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.HBaseTestingUtility;
 import org.apache.hadoop.hbase.testclassification.LargeTests;
 import org.apache.hadoop.hbase.testclassification.RegionServerTests;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.FSUtils;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
-import org.junit.Rule;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
-import org.junit.rules.TestName;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameter;
+import org.junit.runners.Parameterized.Parameters;
 
-@Category({RegionServerTests.class, LargeTests.class})
+@RunWith(Parameterized.class)
+@Category({ RegionServerTests.class, LargeTests.class })
 public class TestBoundedRegionGroupingStrategy {
-  protected static final Log LOG = LogFactory.getLog(TestBoundedRegionGroupingStrategy.class);
+  private static final Log LOG = LogFactory.getLog(TestBoundedRegionGroupingStrategy.class);
+
+  private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
+
+  private static Configuration CONF;
+  private static DistributedFileSystem FS;
 
-  @Rule
-  public TestName currentTest = new TestName();
-  protected static Configuration conf;
-  protected static FileSystem fs;
-  protected final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
+  @Parameter
+  public String walProvider;
+
+  @Parameters(name = "{index}: delegate-provider={0}")
+  public static Iterable<Object[]> data() {
+    return Arrays.asList(new Object[] { "defaultProvider" }, new Object[] { "asyncfs" });
+  }
 
   @Before
   public void setUp() throws Exception {
-    FileStatus[] entries = fs.listStatus(new Path("/"));
-    for (FileStatus dir : entries) {
-      fs.delete(dir.getPath(), true);
-    }
+    CONF.set(DELEGATE_PROVIDER, walProvider);
   }
 
   @After
   public void tearDown() throws Exception {
+    FileStatus[] entries = FS.listStatus(new Path("/"));
+    for (FileStatus dir : entries) {
+      FS.delete(dir.getPath(), true);
+    }
   }
 
   @BeforeClass
   public static void setUpBeforeClass() throws Exception {
-    conf = TEST_UTIL.getConfiguration();
+    CONF = TEST_UTIL.getConfiguration();
     // Make block sizes small.
-    conf.setInt("dfs.blocksize", 1024 * 1024);
+    CONF.setInt("dfs.blocksize", 1024 * 1024);
     // quicker heartbeat interval for faster DN death notification
-    conf.setInt("dfs.namenode.heartbeat.recheck-interval", 5000);
-    conf.setInt("dfs.heartbeat.interval", 1);
-    conf.setInt("dfs.client.socket-timeout", 5000);
+    CONF.setInt("dfs.namenode.heartbeat.recheck-interval", 5000);
+    CONF.setInt("dfs.heartbeat.interval", 1);
+    CONF.setInt("dfs.client.socket-timeout", 5000);
 
     // faster failover with cluster.shutdown();fs.close() idiom
-    conf.setInt("hbase.ipc.client.connect.max.retries", 1);
-    conf.setInt("dfs.client.block.recovery.retries", 1);
-    conf.setInt("hbase.ipc.client.connection.maxidletime", 500);
+    CONF.setInt("hbase.ipc.client.connect.max.retries", 1);
+    CONF.setInt("dfs.client.block.recovery.retries", 1);
+    CONF.setInt("hbase.ipc.client.connection.maxidletime", 500);
 
-    conf.setClass(WAL_PROVIDER, RegionGroupingProvider.class, WALProvider.class);
-    conf.set(REGION_GROUPING_STRATEGY, RegionGroupingProvider.Strategies.bounded.name());
+    CONF.setClass(WAL_PROVIDER, RegionGroupingProvider.class, WALProvider.class);
+    CONF.set(REGION_GROUPING_STRATEGY, RegionGroupingProvider.Strategies.bounded.name());
 
     TEST_UTIL.startMiniDFSCluster(3);
 
-    fs = TEST_UTIL.getDFSCluster().getFileSystem();
+    FS = TEST_UTIL.getDFSCluster().getFileSystem();
   }
 
   @AfterClass
@@ -106,8 +118,8 @@ public class TestBoundedRegionGroupingStrategy {
   public void testConcurrentWrites() throws Exception {
     // Run the WPE tool with three threads writing 3000 edits each concurrently.
     // When done, verify that all edits were written.
-    int errCode = WALPerformanceEvaluation.innerMain(new Configuration(conf),
-        new String [] {"-threads", "3", "-verify", "-noclosefs", "-iterations", "3000"});
+    int errCode = WALPerformanceEvaluation.innerMain(new Configuration(CONF),
+      new String[] { "-threads", "3", "-verify", "-noclosefs", "-iterations", "3000" });
     assertEquals(0, errCode);
   }
 
@@ -117,39 +129,39 @@ public class TestBoundedRegionGroupingStrategy {
   @Test
   public void testMoreRegionsThanBound() throws Exception {
     final String parallelism = Integer.toString(DEFAULT_NUM_REGION_GROUPS * 2);
-    int errCode = WALPerformanceEvaluation.innerMain(new Configuration(conf),
-        new String [] {"-threads", parallelism, "-verify", "-noclosefs", "-iterations", "3000",
-            "-regions", parallelism});
+    int errCode = WALPerformanceEvaluation.innerMain(new Configuration(CONF),
+      new String[] { "-threads", parallelism, "-verify", "-noclosefs", "-iterations", "3000",
+          "-regions", parallelism });
     assertEquals(0, errCode);
   }
 
   @Test
   public void testBoundsGreaterThanDefault() throws Exception {
-    final int temp = conf.getInt(NUM_REGION_GROUPS, DEFAULT_NUM_REGION_GROUPS);
+    final int temp = CONF.getInt(NUM_REGION_GROUPS, DEFAULT_NUM_REGION_GROUPS);
     try {
-      conf.setInt(NUM_REGION_GROUPS, temp*4);
-      final String parallelism = Integer.toString(temp*4);
-      int errCode = WALPerformanceEvaluation.innerMain(new Configuration(conf),
-          new String [] {"-threads", parallelism, "-verify", "-noclosefs", "-iterations",
"3000",
-              "-regions", parallelism});
+      CONF.setInt(NUM_REGION_GROUPS, temp * 4);
+      final String parallelism = Integer.toString(temp * 4);
+      int errCode = WALPerformanceEvaluation.innerMain(new Configuration(CONF),
+        new String[] { "-threads", parallelism, "-verify", "-noclosefs", "-iterations", "3000",
+            "-regions", parallelism });
       assertEquals(0, errCode);
     } finally {
-      conf.setInt(NUM_REGION_GROUPS, temp);
+      CONF.setInt(NUM_REGION_GROUPS, temp);
     }
   }
 
   @Test
   public void testMoreRegionsThanBoundWithBoundsGreaterThanDefault() throws Exception {
-    final int temp = conf.getInt(NUM_REGION_GROUPS, DEFAULT_NUM_REGION_GROUPS);
+    final int temp = CONF.getInt(NUM_REGION_GROUPS, DEFAULT_NUM_REGION_GROUPS);
     try {
-      conf.setInt(NUM_REGION_GROUPS, temp*4);
-      final String parallelism = Integer.toString(temp*4*2);
-      int errCode = WALPerformanceEvaluation.innerMain(new Configuration(conf),
-          new String [] {"-threads", parallelism, "-verify", "-noclosefs", "-iterations",
"3000",
-              "-regions", parallelism});
+      CONF.setInt(NUM_REGION_GROUPS, temp * 4);
+      final String parallelism = Integer.toString(temp * 4 * 2);
+      int errCode = WALPerformanceEvaluation.innerMain(new Configuration(CONF),
+        new String[] { "-threads", parallelism, "-verify", "-noclosefs", "-iterations", "3000",
+            "-regions", parallelism });
       assertEquals(0, errCode);
     } finally {
-      conf.setInt(NUM_REGION_GROUPS, temp);
+      CONF.setInt(NUM_REGION_GROUPS, temp);
     }
   }
 
@@ -158,32 +170,33 @@ public class TestBoundedRegionGroupingStrategy {
    */
   @Test
   public void setMembershipDedups() throws IOException {
-    final int temp = conf.getInt(NUM_REGION_GROUPS, DEFAULT_NUM_REGION_GROUPS);
+    final int temp = CONF.getInt(NUM_REGION_GROUPS, DEFAULT_NUM_REGION_GROUPS);
     WALFactory wals = null;
     try {
-      conf.setInt(NUM_REGION_GROUPS, temp*4);
+      CONF.setInt(NUM_REGION_GROUPS, temp * 4);
       // Set HDFS root directory for storing WAL
-      FSUtils.setRootDir(conf, TEST_UTIL.getDataTestDirOnTestFS());
+      FSUtils.setRootDir(CONF, TEST_UTIL.getDataTestDirOnTestFS());
 
-      wals = new WALFactory(conf, null, currentTest.getMethodName());
-      final Set<WAL> seen = new HashSet<WAL>(temp*4);
+      wals = new WALFactory(CONF, null, "setMembershipDedups");
+      final Set<WAL> seen = new HashSet<WAL>(temp * 4);
       final Random random = new Random();
       int count = 0;
       // we know that this should see one of the wals more than once
-      for (int i = 0; i < temp*8; i++) {
+      for (int i = 0; i < temp * 8; i++) {
         final WAL maybeNewWAL = wals.getWAL(Bytes.toBytes(random.nextInt()), null);
         LOG.info("Iteration " + i + ", checking wal " + maybeNewWAL);
         if (seen.add(maybeNewWAL)) {
           count++;
         }
       }
-      assertEquals("received back a different number of WALs that are not equal() to each
other " +
-          "than the bound we placed.", temp*4, count);
+      assertEquals("received back a different number of WALs that are not equal() to each
other "
+          + "than the bound we placed.",
+        temp * 4, count);
     } finally {
       if (wals != null) {
         wals.close();
       }
-      conf.setInt(NUM_REGION_GROUPS, temp);
+      CONF.setInt(NUM_REGION_GROUPS, temp);
     }
   }
 }


Mime
View raw message