You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by zh...@apache.org on 2019/04/25 14:45:11 UTC
[hbase] branch HBASE-21512 updated (1465841 -> e2c5674)
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a change to branch HBASE-21512
in repository https://gitbox.apache.org/repos/asf/hbase.git.
omit 1465841 HBASE-22303 Fix TestReplicationDroppedTables
omit 4be510b HBASE-22302 Fix TestHbck
omit 0d9e9b9 HBASE-22297 Fix TestRegionMergeTransitionOnCluster and TestSplitTransactionOnCluster
omit c2df8e3 HBASE-22295 Fix TestClientOperationTimeout
omit 2e99be0 HBASE-22281 Fix failed shell UTs
omit 71a775c HBASE-22223 Implement RegionLocator based on AsyncTableRegionLocator
omit 46fcb78 HBASE-22238 Fix TestRpcControllerFactory
omit 85abd56 HBASE-21725 Implement BufferedMutator Based on AsyncBufferedMutator
omit 7b74bb7 HBASE-22239 Should translate the exception before returning it in AsyncRegionServerAdmin
new 9c0dfab HBASE-21725 Implement BufferedMutator Based on AsyncBufferedMutator
new 681a23c HBASE-22238 Fix TestRpcControllerFactory
new 9a611b4 HBASE-22223 Implement RegionLocator based on AsyncTableRegionLocator
new 602c075 HBASE-22281 Fix failed shell UTs
new 677aba5 HBASE-22295 Fix TestClientOperationTimeout
new 1e78cf8 HBASE-22297 Fix TestRegionMergeTransitionOnCluster and TestSplitTransactionOnCluster
new 4a0b702 HBASE-22302 Fix TestHbck
new e2c5674 HBASE-22239 Also catch RemoteException in SyncReplicationTestBase.verifyReplicationRequestRejection
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (1465841)
\
N -- N -- N refs/heads/HBASE-21512 (e2c5674)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 8 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/hadoop/hbase/client/AsyncRegionServerAdmin.java | 14 ++------------
.../hadoop/hbase/replication/SyncReplicationTestBase.java | 14 ++++++++++++--
2 files changed, 14 insertions(+), 14 deletions(-)
[hbase] 04/08: HBASE-22281 Fix failed shell UTs
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a commit to branch HBASE-21512
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit 602c075da5f3a4aca2c2a236cc5aa2f92d1a568f
Author: zhangduo <zh...@apache.org>
AuthorDate: Mon Apr 22 22:00:43 2019 +0800
HBASE-22281 Fix failed shell UTs
---
hbase-shell/src/main/ruby/hbase/admin.rb | 44 ++++++++++++++++++++++++--------
hbase-shell/src/main/ruby/hbase/hbase.rb | 23 +++++++++++------
hbase-shell/src/main/ruby/hbase/table.rb | 2 +-
3 files changed, 50 insertions(+), 19 deletions(-)
diff --git a/hbase-shell/src/main/ruby/hbase/admin.rb b/hbase-shell/src/main/ruby/hbase/admin.rb
index acfe178..ab08fdf 100644
--- a/hbase-shell/src/main/ruby/hbase/admin.rb
+++ b/hbase-shell/src/main/ruby/hbase/admin.rb
@@ -54,7 +54,7 @@ module Hbase
# Requests a table or region or region server flush
def flush(name)
@admin.flushRegion(name.to_java_bytes)
- rescue java.lang.IllegalArgumentException
+ rescue java.lang.IllegalArgumentException, org.apache.hadoop.hbase.UnknownRegionException
# Unknown region. Try table.
begin
@admin.flush(TableName.valueOf(name))
@@ -79,9 +79,17 @@ module Hbase
end
begin
- @admin.compactRegion(table_or_region_name.to_java_bytes, family_bytes)
- rescue java.lang.IllegalArgumentException => e
- @admin.compact(TableName.valueOf(table_or_region_name), family_bytes, compact_type)
+ if family_bytes.nil?
+ @admin.compactRegion(table_or_region_name.to_java_bytes)
+ else
+ @admin.compactRegion(table_or_region_name.to_java_bytes, family_bytes)
+ end
+ rescue java.lang.IllegalArgumentException, org.apache.hadoop.hbase.UnknownRegionException
+ if family_bytes.nil?
+ @admin.compact(TableName.valueOf(table_or_region_name), compact_type)
+ else
+ @admin.compact(TableName.valueOf(table_or_region_name), family_bytes, compact_type)
+ end
end
end
@@ -124,9 +132,17 @@ module Hbase
end
begin
- @admin.majorCompactRegion(table_or_region_name.to_java_bytes, family_bytes)
- rescue java.lang.IllegalArgumentException => e
- @admin.majorCompact(TableName.valueOf(table_or_region_name), family_bytes, compact_type)
+ if family_bytes.nil?
+ @admin.majorCompactRegion(table_or_region_name.to_java_bytes)
+ else
+ @admin.majorCompactRegion(table_or_region_name.to_java_bytes, family_bytes)
+ end
+ rescue java.lang.IllegalArgumentException, org.apache.hadoop.hbase.UnknownRegionException
+ if family_bytes.nil?
+ @admin.majorCompact(TableName.valueOf(table_or_region_name), compact_type)
+ else
+ @admin.majorCompact(TableName.valueOf(table_or_region_name), family_bytes, compact_type)
+ end
end
end
@@ -144,9 +160,17 @@ module Hbase
split_point_bytes = nil
split_point_bytes = split_point.to_java_bytes unless split_point.nil?
begin
- @admin.splitRegionAsync(table_or_region_name.to_java_bytes, split_point_bytes).get
- rescue java.lang.IllegalArgumentException, org.apache.hadoop.hbase.UnknownRegionException => e
- @admin.split(TableName.valueOf(table_or_region_name), split_point_bytes)
+ if split_point_bytes.nil?
+ org.apache.hadoop.hbase.util.FutureUtils.get(@admin.splitRegionAsync(table_or_region_name.to_java_bytes))
+ else
+ org.apache.hadoop.hbase.util.FutureUtils.get(@admin.splitRegionAsync(table_or_region_name.to_java_bytes, split_point_bytes))
+ end
+ rescue java.lang.IllegalArgumentException, org.apache.hadoop.hbase.UnknownRegionException
+ if split_point_bytes.nil?
+ @admin.split(TableName.valueOf(table_or_region_name))
+ else
+ @admin.split(TableName.valueOf(table_or_region_name), split_point_bytes)
+ end
end
end
diff --git a/hbase-shell/src/main/ruby/hbase/hbase.rb b/hbase-shell/src/main/ruby/hbase/hbase.rb
index 1f37f99..a9b35ed 100644
--- a/hbase-shell/src/main/ruby/hbase/hbase.rb
+++ b/hbase-shell/src/main/ruby/hbase/hbase.rb
@@ -42,16 +42,21 @@ module Hbase
configuration.setInt('hbase.client.retries.number', 7)
configuration.setInt('hbase.ipc.client.connect.max.retries', 3)
end
- @connection = ConnectionFactory.createConnection(configuration)
end
+ def connection
+ if @connection.nil?
+ @connection = ConnectionFactory.createConnection(configuration)
+ end
+ @connection
+ end
# Returns ruby's Admin class from admin.rb
def admin
- ::Hbase::Admin.new(@connection)
+ ::Hbase::Admin.new(self.connection)
end
def rsgroup_admin
- ::Hbase::RSGroupAdmin.new(@connection)
+ ::Hbase::RSGroupAdmin.new(self.connection)
end
def taskmonitor
@@ -60,7 +65,7 @@ module Hbase
# Create new one each time
def table(table, shell)
- ::Hbase::Table.new(@connection.getTable(TableName.valueOf(table)), shell)
+ ::Hbase::Table.new(self.connection.getTable(TableName.valueOf(table)), shell)
end
def replication_admin
@@ -68,19 +73,21 @@ module Hbase
end
def security_admin
- ::Hbase::SecurityAdmin.new(@connection.getAdmin)
+ ::Hbase::SecurityAdmin.new(self.connection.getAdmin)
end
def visibility_labels_admin
- ::Hbase::VisibilityLabelsAdmin.new(@connection.getAdmin)
+ ::Hbase::VisibilityLabelsAdmin.new(self.connection.getAdmin)
end
def quotas_admin
- ::Hbase::QuotasAdmin.new(@connection.getAdmin)
+ ::Hbase::QuotasAdmin.new(self.connection.getAdmin)
end
def shutdown
- @connection.close
+ if @connection != nil
+ @connection.close
+ end
end
end
end
diff --git a/hbase-shell/src/main/ruby/hbase/table.rb b/hbase-shell/src/main/ruby/hbase/table.rb
index 26bb84f..fe8ec26 100644
--- a/hbase-shell/src/main/ruby/hbase/table.rb
+++ b/hbase-shell/src/main/ruby/hbase/table.rb
@@ -717,7 +717,7 @@ EOF
# Returns a list of column names in the table
def get_all_columns
- @table.table_descriptor.getFamilies.map do |family|
+ @table.descriptor.getColumnFamilies.map do |family|
"#{family.getNameAsString}:"
end
end
[hbase] 02/08: HBASE-22238 Fix TestRpcControllerFactory
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a commit to branch HBASE-21512
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit 681a23cef42c058e29d7ff942301df4acaf30684
Author: zhangduo <zh...@apache.org>
AuthorDate: Sun Apr 14 21:30:34 2019 +0800
HBASE-22238 Fix TestRpcControllerFactory
---
.../hbase/client/TestRpcControllerFactory.java | 171 ++++++++++-----------
1 file changed, 77 insertions(+), 94 deletions(-)
diff --git a/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/client/TestRpcControllerFactory.java b/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/client/TestRpcControllerFactory.java
index 2d60733..bdda4e8 100644
--- a/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/client/TestRpcControllerFactory.java
+++ b/hbase-endpoint/src/test/java/org/apache/hadoop/hbase/client/TestRpcControllerFactory.java
@@ -32,6 +32,7 @@ import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.HBaseTestingUtility;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Scan.ReadType;
import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
import org.apache.hadoop.hbase.coprocessor.ProtobufCoprocessorService;
import org.apache.hadoop.hbase.ipc.DelegatingHBaseRpcController;
@@ -52,12 +53,12 @@ import org.apache.hbase.thirdparty.com.google.common.collect.ConcurrentHashMulti
import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
import org.apache.hbase.thirdparty.com.google.common.collect.Multiset;
-@Category({MediumTests.class, ClientTests.class})
+@Category({ MediumTests.class, ClientTests.class })
public class TestRpcControllerFactory {
@ClassRule
public static final HBaseClassTestRule CLASS_RULE =
- HBaseClassTestRule.forClass(TestRpcControllerFactory.class);
+ HBaseClassTestRule.forClass(TestRpcControllerFactory.class);
public static class StaticRpcControllerFactory extends RpcControllerFactory {
@@ -85,7 +86,6 @@ public class TestRpcControllerFactory {
private static Multiset<Integer> GROUPED_PRIORITY = ConcurrentHashMultiset.create();
private static AtomicInteger INT_PRIORITY = new AtomicInteger();
- private static AtomicInteger TABLE_PRIORITY = new AtomicInteger();
public CountingRpcController(HBaseRpcController delegate) {
super(delegate);
@@ -93,24 +93,8 @@ public class TestRpcControllerFactory {
@Override
public void setPriority(int priority) {
- int oldPriority = getPriority();
- super.setPriority(priority);
- int newPriority = getPriority();
- if (newPriority != oldPriority) {
- INT_PRIORITY.incrementAndGet();
- GROUPED_PRIORITY.add(priority);
- }
- }
-
- @Override
- public void setPriority(TableName tn) {
- super.setPriority(tn);
- // ignore counts for system tables - it could change and we really only want to check on what
- // the client should change
- if (tn != null && !tn.isSystemTable()) {
- TABLE_PRIORITY.incrementAndGet();
- }
-
+ INT_PRIORITY.incrementAndGet();
+ GROUPED_PRIORITY.add(priority);
}
}
@@ -120,7 +104,7 @@ public class TestRpcControllerFactory {
public TestName name = new TestName();
@BeforeClass
- public static void setup() throws Exception {
+ public static void setUp() throws Exception {
// load an endpoint so we have an endpoint to test - it doesn't matter which one, but
// this is already in tests, so we can just use it.
Configuration conf = UTIL.getConfiguration();
@@ -131,7 +115,7 @@ public class TestRpcControllerFactory {
}
@AfterClass
- public static void teardown() throws Exception {
+ public static void tearDown() throws Exception {
UTIL.shutdownMiniCluster();
}
@@ -154,84 +138,83 @@ public class TestRpcControllerFactory {
// change one of the connection properties so we get a new Connection with our configuration
conf.setInt(HConstants.HBASE_RPC_TIMEOUT_KEY, HConstants.DEFAULT_HBASE_RPC_TIMEOUT + 1);
- Connection connection = ConnectionFactory.createConnection(conf);
- Table table = connection.getTable(tableName);
- byte[] row = Bytes.toBytes("row");
- Put p = new Put(row);
- p.addColumn(fam1, fam1, Bytes.toBytes("val0"));
- table.put(p);
-
- Integer counter = 1;
- counter = verifyCount(counter);
-
- Delete d = new Delete(row);
- d.addColumn(fam1, fam1);
- table.delete(d);
- counter = verifyCount(counter);
-
- Put p2 = new Put(row);
- p2.addColumn(fam1, Bytes.toBytes("qual"), Bytes.toBytes("val1"));
- table.batch(Lists.newArrayList(p, p2), null);
- // this only goes to a single server, so we don't need to change the count here
- counter = verifyCount(counter);
-
- Append append = new Append(row);
- append.addColumn(fam1, fam1, Bytes.toBytes("val2"));
- table.append(append);
- counter = verifyCount(counter);
-
- // and check the major lookup calls as well
- Get g = new Get(row);
- table.get(g);
- counter = verifyCount(counter);
-
- ResultScanner scan = table.getScanner(fam1);
- scan.next();
- scan.close();
- counter = verifyCount(counter + 1);
-
- Get g2 = new Get(row);
- table.get(Lists.newArrayList(g, g2));
- // same server, so same as above for not changing count
- counter = verifyCount(counter);
-
- // make sure all the scanner types are covered
- Scan scanInfo = new Scan(row);
- // regular small
- scanInfo.setSmall(true);
- counter = doScan(table, scanInfo, counter);
-
- // reversed, small
- scanInfo.setReversed(true);
- counter = doScan(table, scanInfo, counter);
-
- // reversed, regular
- scanInfo.setSmall(false);
- counter = doScan(table, scanInfo, counter + 1);
-
- // make sure we have no priority count
- verifyPriorityGroupCount(HConstants.ADMIN_QOS, 0);
- // lets set a custom priority on a get
- Get get = new Get(row);
- get.setPriority(HConstants.ADMIN_QOS);
- table.get(get);
- verifyPriorityGroupCount(HConstants.ADMIN_QOS, 1);
-
- table.close();
- connection.close();
+ try (Connection connection = ConnectionFactory.createConnection(conf);
+ Table table = connection.getTable(tableName)) {
+ byte[] row = Bytes.toBytes("row");
+ Put p = new Put(row);
+ p.addColumn(fam1, fam1, Bytes.toBytes("val0"));
+ table.put(p);
+
+ Integer counter = 1;
+ counter = verifyCount(counter);
+
+ Delete d = new Delete(row);
+ d.addColumn(fam1, fam1);
+ table.delete(d);
+ counter = verifyCount(counter);
+
+ Put p2 = new Put(row);
+ p2.addColumn(fam1, Bytes.toBytes("qual"), Bytes.toBytes("val1"));
+ table.batch(Lists.newArrayList(p, p2), null);
+ // this only goes to a single server, so we don't need to change the count here
+ counter = verifyCount(counter);
+
+ Append append = new Append(row);
+ append.addColumn(fam1, fam1, Bytes.toBytes("val2"));
+ table.append(append);
+ counter = verifyCount(counter);
+
+ // and check the major lookup calls as well
+ Get g = new Get(row);
+ table.get(g);
+ counter = verifyCount(counter);
+
+ ResultScanner scan = table.getScanner(fam1);
+ scan.next();
+ scan.close();
+ counter = verifyCount(counter + 1);
+
+ Get g2 = new Get(row);
+ table.get(Lists.newArrayList(g, g2));
+ // same server, so same as above for not changing count
+ counter = verifyCount(counter);
+
+ // make sure all the scanner types are covered
+ Scan scanInfo = new Scan().withStartRow(row);
+ // regular small
+ scanInfo.setReadType(ReadType.PREAD);
+ counter = doScan(table, scanInfo, counter);
+
+ // reversed, small
+ scanInfo.setReversed(true);
+ counter = doScan(table, scanInfo, counter);
+
+ // reversed, regular
+ scanInfo.setReadType(ReadType.STREAM);
+ counter = doScan(table, scanInfo, counter + 1);
+
+ // make sure we have no priority count
+ verifyPriorityGroupCount(HConstants.ADMIN_QOS, 0);
+ // lets set a custom priority on a get
+ Get get = new Get(row);
+ get.setPriority(HConstants.ADMIN_QOS);
+ table.get(get);
+ // we will reset the controller for setting the call timeout so it will lead to an extra
+ // setPriority
+ verifyPriorityGroupCount(HConstants.ADMIN_QOS, 2);
+ }
}
int doScan(Table table, Scan scan, int expectedCount) throws IOException {
- ResultScanner results = table.getScanner(scan);
- results.next();
- results.close();
+ try (ResultScanner results = table.getScanner(scan)) {
+ results.next();
+ }
return verifyCount(expectedCount);
}
int verifyCount(Integer counter) {
- assertTrue(CountingRpcController.TABLE_PRIORITY.get() >= counter);
- assertEquals(0, CountingRpcController.INT_PRIORITY.get());
- return CountingRpcController.TABLE_PRIORITY.get() + 1;
+ assertTrue(CountingRpcController.INT_PRIORITY.get() >= counter);
+ return CountingRpcController.GROUPED_PRIORITY.count(HConstants.NORMAL_QOS) + 1;
}
void verifyPriorityGroupCount(int priorityLevel, int count) {
[hbase] 06/08: HBASE-22297 Fix TestRegionMergeTransitionOnCluster
and TestSplitTransactionOnCluster
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a commit to branch HBASE-21512
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit 1e78cf88eb4c0c21687d83688924063046dda101
Author: zhangduo <zh...@apache.org>
AuthorDate: Tue Apr 23 22:22:39 2019 +0800
HBASE-22297 Fix TestRegionMergeTransitionOnCluster and TestSplitTransactionOnCluster
Signed-off-by: Michael Stack <st...@apache.org>
---
.../TestRegionMergeTransactionOnCluster.java | 19 +++++++++----------
.../regionserver/TestSplitTransactionOnCluster.java | 3 ++-
2 files changed, 11 insertions(+), 11 deletions(-)
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionMergeTransactionOnCluster.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionMergeTransactionOnCluster.java
index ea93468..7b54ffb 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionMergeTransactionOnCluster.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestRegionMergeTransactionOnCluster.java
@@ -26,7 +26,6 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.lang3.RandomUtils;
import org.apache.hadoop.conf.Configuration;
@@ -63,6 +62,7 @@ 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.hbase.util.FutureUtils;
import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.PairOfSameType;
@@ -312,7 +312,6 @@ public class TestRegionMergeTransactionOnCluster {
LOG.info("Starting " + name.getMethodName());
final TableName tableName = TableName.valueOf(name.getMethodName());
final Admin admin = TEST_UTIL.getAdmin();
- final int syncWaitTimeout = 10 * 60000; // 10min
try {
// Create table and load data.
@@ -326,8 +325,8 @@ public class TestRegionMergeTransactionOnCluster {
am.offlineRegion(b);
try {
// Merge offline region. Region a is offline here
- admin.mergeRegionsAsync(a.getEncodedNameAsBytes(), b.getEncodedNameAsBytes(), false)
- .get(syncWaitTimeout, TimeUnit.MILLISECONDS);
+ FutureUtils.get(
+ admin.mergeRegionsAsync(a.getEncodedNameAsBytes(), b.getEncodedNameAsBytes(), false));
fail("Offline regions should not be able to merge");
} catch (DoNotRetryRegionException ie) {
System.out.println(ie);
@@ -336,21 +335,21 @@ public class TestRegionMergeTransactionOnCluster {
try {
// Merge the same region: b and b.
- admin.mergeRegionsAsync(b.getEncodedNameAsBytes(), b.getEncodedNameAsBytes(), true);
+ FutureUtils
+ .get(admin.mergeRegionsAsync(b.getEncodedNameAsBytes(), b.getEncodedNameAsBytes(), true));
fail("A region should not be able to merge with itself, even forcifully");
} catch (IOException ie) {
assertTrue("Exception should mention regions not online",
- StringUtils.stringifyException(ie).contains("region to itself")
- && ie instanceof MergeRegionException);
+ StringUtils.stringifyException(ie).contains("region to itself") &&
+ ie instanceof MergeRegionException);
}
try {
// Merge unknown regions
- admin.mergeRegionsAsync(Bytes.toBytes("-f1"), Bytes.toBytes("-f2"), true);
+ FutureUtils.get(admin.mergeRegionsAsync(Bytes.toBytes("-f1"), Bytes.toBytes("-f2"), true));
fail("Unknown region could not be merged");
} catch (IOException ie) {
- assertTrue("UnknownRegionException should be thrown",
- ie instanceof UnknownRegionException);
+ assertTrue("UnknownRegionException should be thrown", ie instanceof UnknownRegionException);
}
table.close();
} finally {
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestSplitTransactionOnCluster.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestSplitTransactionOnCluster.java
index 405819e..21d8297 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestSplitTransactionOnCluster.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestSplitTransactionOnCluster.java
@@ -89,6 +89,7 @@ import org.apache.hadoop.hbase.testclassification.RegionServerTests;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.FSUtils;
+import org.apache.hadoop.hbase.util.FutureUtils;
import org.apache.hadoop.hbase.util.HBaseFsck;
import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread;
import org.apache.hadoop.hbase.util.RetryCounter;
@@ -328,7 +329,7 @@ public class TestSplitTransactionOnCluster {
// We don't roll back here anymore. Instead we fail-fast on construction of the
// split transaction. Catch the exception instead.
try {
- this.admin.splitRegionAsync(hri.getRegionName());
+ FutureUtils.get(this.admin.splitRegionAsync(hri.getRegionName()));
fail();
} catch (DoNotRetryRegionException e) {
// Expected
[hbase] 08/08: HBASE-22239 Also catch RemoteException in
SyncReplicationTestBase.verifyReplicationRequestRejection
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a commit to branch HBASE-21512
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit e2c5674f5974452bcd1a39f315c0bf335d7e1d2a
Author: zhangduo <zh...@apache.org>
AuthorDate: Thu Apr 25 22:43:51 2019 +0800
HBASE-22239 Also catch RemoteException in SyncReplicationTestBase.verifyReplicationRequestRejection
---
.../hadoop/hbase/replication/SyncReplicationTestBase.java | 14 ++++++++++++--
1 file changed, 12 insertions(+), 2 deletions(-)
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/SyncReplicationTestBase.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/SyncReplicationTestBase.java
index e0d112d..fd8df32 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/SyncReplicationTestBase.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/SyncReplicationTestBase.java
@@ -17,8 +17,10 @@
*/
package org.apache.hadoop.hbase.replication;
+import static org.hamcrest.CoreMatchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -51,6 +53,7 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.wal.WAL.Entry;
import org.apache.hadoop.hbase.wal.WALEdit;
import org.apache.hadoop.hbase.wal.WALKeyImpl;
+import org.apache.hadoop.ipc.RemoteException;
import org.junit.AfterClass;
import org.junit.BeforeClass;
@@ -247,6 +250,12 @@ public class SyncReplicationTestBase {
}
}
+ private void assertRejection(Throwable error) {
+ assertThat(error, instanceOf(DoNotRetryIOException.class));
+ assertTrue(error.getMessage().contains("Reject to apply to sink cluster"));
+ assertTrue(error.getMessage().contains(TABLE_NAME.toString()));
+ }
+
protected final void verifyReplicationRequestRejection(HBaseTestingUtility utility,
boolean expectedRejection) throws Exception {
HRegionServer regionServer = utility.getRSForFirstRegionInTable(TABLE_NAME);
@@ -264,9 +273,10 @@ public class SyncReplicationTestBase {
ReplicationProtbufUtil.replicateWALEntry(
connection.getRegionServerAdmin(regionServer.getServerName()), entries, null, null, null);
fail("Should throw IOException when sync-replication state is in A or DA");
+ } catch (RemoteException e) {
+ assertRejection(e.unwrapRemoteException());
} catch (DoNotRetryIOException e) {
- assertTrue(e.getMessage().contains("Reject to apply to sink cluster"));
- assertTrue(e.getMessage().contains(TABLE_NAME.toString()));
+ assertRejection(e);
}
}
}
[hbase] 03/08: HBASE-22223 Implement RegionLocator based on
AsyncTableRegionLocator
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a commit to branch HBASE-21512
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit 9a611b4bd7ff44f6ea6c154f830a708c2b4fcd37
Author: Duo Zhang <zh...@apache.org>
AuthorDate: Tue Apr 16 16:52:54 2019 +0800
HBASE-22223 Implement RegionLocator based on AsyncTableRegionLocator
---
.../hadoop/hbase/client/AsyncConnectionImpl.java | 12 +---
.../client/ConnectionOverAsyncConnection.java | 16 ++---
.../RegionLocatorOverAsyncTableRegionLocator.java | 70 ++++++++++++++++++++++
3 files changed, 76 insertions(+), 22 deletions(-)
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java
index 4a00412..84e1da6 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java
@@ -28,7 +28,6 @@ import static org.apache.hadoop.hbase.client.NonceGenerator.CLIENT_NONCES_ENABLE
import static org.apache.hadoop.hbase.util.FutureUtils.addListener;
import java.io.IOException;
-import java.io.UncheckedIOException;
import java.net.SocketAddress;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
@@ -208,7 +207,7 @@ class AsyncConnectionImpl implements AsyncConnection {
metrics.ifPresent(MetricsConnection::shutdown);
ConnectionOverAsyncConnection c = this.conn;
if (c != null) {
- c.closeConnImpl();
+ c.closePool();
}
closed = true;
}
@@ -362,14 +361,7 @@ class AsyncConnectionImpl implements AsyncConnection {
if (c != null) {
return c;
}
- try {
- c = new ConnectionOverAsyncConnection(this,
- ConnectionFactory.createConnectionImpl(conf, null, user));
- } catch (IOException e) {
- // TODO: finally we will not rely on ConnectionImplementation anymore and there will no
- // IOException here.
- throw new UncheckedIOException(e);
- }
+ c = new ConnectionOverAsyncConnection(this);
this.conn = c;
}
return c;
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionOverAsyncConnection.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionOverAsyncConnection.java
index 8ec7ab8..861aab0 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionOverAsyncConnection.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionOverAsyncConnection.java
@@ -43,20 +43,12 @@ class ConnectionOverAsyncConnection implements Connection {
private volatile ExecutorService batchPool = null;
- protected final AsyncConnectionImpl conn;
-
- /**
- * @deprecated we can not implement all the related stuffs at once so keep it here for now, will
- * remove it after we implement all the stuffs, like Admin, RegionLocator, etc.
- */
- @Deprecated
- private final ConnectionImplementation oldConn;
+ private final AsyncConnectionImpl conn;
private final ConnectionConfiguration connConf;
- ConnectionOverAsyncConnection(AsyncConnectionImpl conn, ConnectionImplementation oldConn) {
+ ConnectionOverAsyncConnection(AsyncConnectionImpl conn) {
this.conn = conn;
- this.oldConn = oldConn;
this.connConf = new ConnectionConfiguration(conn.getConfiguration());
}
@@ -109,7 +101,7 @@ class ConnectionOverAsyncConnection implements Connection {
@Override
public RegionLocator getRegionLocator(TableName tableName) throws IOException {
- return oldConn.getRegionLocator(tableName);
+ return new RegionLocatorOverAsyncTableRegionLocator(conn.getRegionLocator(tableName));
}
@Override
@@ -129,7 +121,7 @@ class ConnectionOverAsyncConnection implements Connection {
// will be called from AsyncConnection, to avoid infinite loop as in the above method we will call
// AsyncConnection.close.
- void closeConnImpl() {
+ void closePool() {
ExecutorService batchPool = this.batchPool;
if (batchPool != null) {
ConnectionUtils.shutdownPool(batchPool);
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RegionLocatorOverAsyncTableRegionLocator.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RegionLocatorOverAsyncTableRegionLocator.java
new file mode 100644
index 0000000..5e21e3b
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RegionLocatorOverAsyncTableRegionLocator.java
@@ -0,0 +1,70 @@
+/**
+ * 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.client;
+
+import static org.apache.hadoop.hbase.util.FutureUtils.get;
+
+import java.io.IOException;
+import java.util.List;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * The {@link RegionLocator} implementation based on {@link AsyncTableRegionLocator}.
+ */
+@InterfaceAudience.Private
+class RegionLocatorOverAsyncTableRegionLocator implements RegionLocator {
+
+ private final AsyncTableRegionLocator locator;
+
+ RegionLocatorOverAsyncTableRegionLocator(AsyncTableRegionLocator locator) {
+ this.locator = locator;
+ }
+
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public HRegionLocation getRegionLocation(byte[] row, int replicaId, boolean reload)
+ throws IOException {
+ return get(locator.getRegionLocation(row, replicaId, reload));
+ }
+
+ @Override
+ public List<HRegionLocation> getRegionLocations(byte[] row, boolean reload) throws IOException {
+ return get(locator.getRegionLocations(row, reload));
+ }
+
+ @Override
+ public void clearRegionLocationCache() {
+ locator.clearRegionLocationCache();
+ }
+
+ @Override
+ public List<HRegionLocation> getAllRegionLocations() throws IOException {
+ return get(locator.getAllRegionLocations());
+ }
+
+ @Override
+ public TableName getName() {
+ return locator.getName();
+ }
+
+}
[hbase] 07/08: HBASE-22302 Fix TestHbck
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a commit to branch HBASE-21512
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit 4a0b70224eabc9b437c5121469665605a23f1f18
Author: zhangduo <zh...@apache.org>
AuthorDate: Wed Apr 24 22:30:02 2019 +0800
HBASE-22302 Fix TestHbck
---
.../src/test/java/org/apache/hadoop/hbase/client/TestHbck.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestHbck.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestHbck.java
index d9a7ca9..ee277d9 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestHbck.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestHbck.java
@@ -29,7 +29,6 @@ import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
-
import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.HBaseClassTestRule;
@@ -273,7 +272,8 @@ public class TestHbck {
FailingSplitAfterMetaUpdatedMasterObserver observer = master.getMasterCoprocessorHost()
.findCoprocessor(FailingSplitAfterMetaUpdatedMasterObserver.class);
assertNotNull(observer);
- try (Admin admin = TEST_UTIL.getConnection().getAdmin()) {
+ try {
+ AsyncAdmin admin = TEST_UTIL.getAsyncConnection().getAdmin();
byte[] splitKey = Bytes.toBytes("bcd");
admin.split(TableName.valueOf(testTable), splitKey);
observer.latch.await(5000, TimeUnit.MILLISECONDS);
[hbase] 01/08: HBASE-21725 Implement BufferedMutator Based on
AsyncBufferedMutator
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a commit to branch HBASE-21512
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit 9c0dfab95594e655e0bf3d16d712deacf24b77a9
Author: zhangduo <zh...@apache.org>
AuthorDate: Sun Apr 14 20:32:38 2019 +0800
HBASE-21725 Implement BufferedMutator Based on AsyncBufferedMutator
---
.../hadoop/hbase/client/BufferedMutator.java | 10 ++
.../BufferedMutatorOverAsyncBufferedMutator.java | 175 +++++++++++++++++++++
.../hadoop/hbase/client/BufferedMutatorParams.java | 23 ++-
.../client/ConnectionOverAsyncConnection.java | 19 ++-
.../hadoop/hbase/client/TestBufferedMutator.java | 82 ----------
.../hadoop/hbase/client/TestBufferedMutator.java | 90 +++++++++++
6 files changed, 309 insertions(+), 90 deletions(-)
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutator.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutator.java
index 7805f77..8ad6a79 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutator.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutator.java
@@ -62,7 +62,11 @@ import org.apache.yetus.audience.InterfaceAudience;
public interface BufferedMutator extends Closeable {
/**
* Key to use setting non-default BufferedMutator implementation in Configuration.
+ * <p/>
+ * @deprecated Since 3.0.0, will be removed in 4.0.0. For internal test use only, do not use it
+ * any more.
*/
+ @Deprecated
String CLASSNAME_KEY = "hbase.client.bufferedmutator.classname";
/**
@@ -179,12 +183,18 @@ public interface BufferedMutator extends Closeable {
/**
* Set rpc timeout for this mutator instance
+ * @deprecated Since 3.0.0, will be removed in 4.0.0. Please set this through the
+ * {@link BufferedMutatorParams}.
*/
+ @Deprecated
void setRpcTimeout(int timeout);
/**
* Set operation timeout for this mutator instance
+ * @deprecated Since 3.0.0, will be removed in 4.0.0. Please set this through the
+ * {@link BufferedMutatorParams}.
*/
+ @Deprecated
void setOperationTimeout(int timeout);
/**
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutatorOverAsyncBufferedMutator.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutatorOverAsyncBufferedMutator.java
new file mode 100644
index 0000000..a7d4595
--- /dev/null
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutatorOverAsyncBufferedMutator.java
@@ -0,0 +1,175 @@
+/**
+ * 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.client;
+
+import static org.apache.hadoop.hbase.util.FutureUtils.addListener;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * {@link BufferedMutator} implementation based on {@link AsyncBufferedMutator}.
+ */
+@InterfaceAudience.Private
+class BufferedMutatorOverAsyncBufferedMutator implements BufferedMutator {
+
+ private final AsyncBufferedMutator mutator;
+
+ private final ExceptionListener listener;
+
+ private List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ private final ConcurrentLinkedQueue<Pair<Mutation, Throwable>> errors =
+ new ConcurrentLinkedQueue<>();
+
+ private final static int BUFFERED_FUTURES_THRESHOLD = 1024;
+
+ BufferedMutatorOverAsyncBufferedMutator(AsyncBufferedMutator mutator,
+ ExceptionListener listener) {
+ this.mutator = mutator;
+ this.listener = listener;
+ }
+
+ @Override
+ public TableName getName() {
+ return mutator.getName();
+ }
+
+ @Override
+ public Configuration getConfiguration() {
+ return mutator.getConfiguration();
+ }
+
+ @Override
+ public void mutate(Mutation mutation) throws IOException {
+ mutate(Collections.singletonList(mutation));
+ }
+
+ private static final Pattern ADDR_MSG_MATCHER = Pattern.compile("Call to (\\S+) failed");
+
+ // not always work, so may return an empty string
+ private String getHostnameAndPort(Throwable error) {
+ Matcher matcher = ADDR_MSG_MATCHER.matcher(error.getMessage());
+ if (matcher.matches()) {
+ return matcher.group(1);
+ } else {
+ return "";
+ }
+ }
+
+ private RetriesExhaustedWithDetailsException makeError() {
+ List<Row> rows = new ArrayList<>();
+ List<Throwable> throwables = new ArrayList<>();
+ List<String> hostnameAndPorts = new ArrayList<>();
+ for (;;) {
+ Pair<Mutation, Throwable> pair = errors.poll();
+ if (pair == null) {
+ break;
+ }
+ rows.add(pair.getFirst());
+ throwables.add(pair.getSecond());
+ hostnameAndPorts.add(getHostnameAndPort(pair.getSecond()));
+ }
+ return new RetriesExhaustedWithDetailsException(throwables, rows, hostnameAndPorts);
+ }
+
+ @Override
+ public void mutate(List<? extends Mutation> mutations) throws IOException {
+ List<CompletableFuture<Void>> toBuffered = new ArrayList<>();
+ List<CompletableFuture<Void>> fs = mutator.mutate(mutations);
+ for (int i = 0, n = fs.size(); i < n; i++) {
+ CompletableFuture<Void> toComplete = new CompletableFuture<>();
+ final int index = i;
+ addListener(fs.get(index), (r, e) -> {
+ if (e != null) {
+ errors.add(Pair.newPair(mutations.get(index), e));
+ toComplete.completeExceptionally(e);
+ } else {
+ toComplete.complete(r);
+ }
+ });
+ toBuffered.add(toComplete);
+ }
+ synchronized (this) {
+ futures.addAll(toBuffered);
+ if (futures.size() > BUFFERED_FUTURES_THRESHOLD) {
+ tryCompleteFuture();
+ }
+ if (!errors.isEmpty()) {
+ RetriesExhaustedWithDetailsException error = makeError();
+ listener.onException(error, this);
+ }
+ }
+ }
+
+ private void tryCompleteFuture() {
+ futures = futures.stream().filter(f -> !f.isDone()).collect(Collectors.toList());
+ }
+
+ @Override
+ public void close() throws IOException {
+ flush();
+ mutator.close();
+ }
+
+ @Override
+ public void flush() throws IOException {
+ mutator.flush();
+ synchronized (this) {
+ List<CompletableFuture<Void>> toComplete = this.futures;
+ this.futures = new ArrayList<>();
+ try {
+ CompletableFuture.allOf(toComplete.toArray(new CompletableFuture<?>[toComplete.size()]))
+ .join();
+ } catch (CompletionException e) {
+ // just ignore, we will record the actual error in the errors field
+ }
+ if (!errors.isEmpty()) {
+ RetriesExhaustedWithDetailsException error = makeError();
+ listener.onException(error, this);
+ }
+ }
+ }
+
+ @Override
+ public long getWriteBufferSize() {
+ return mutator.getWriteBufferSize();
+ }
+
+ @Override
+ public void setRpcTimeout(int timeout) {
+ // no effect
+ }
+
+ @Override
+ public void setOperationTimeout(int timeout) {
+ // no effect
+ }
+}
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutatorParams.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutatorParams.java
index 3f6c565..49fb77b 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutatorParams.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/BufferedMutatorParams.java
@@ -101,13 +101,21 @@ public class BufferedMutatorParams implements Cloneable {
return this;
}
+ /**
+ * @deprecated Since 3.0.0, will be removed in 4.0.0. We use a common timer in the whole client
+ * implementation so you can not set it any more.
+ */
+ @Deprecated
public long getWriteBufferPeriodicFlushTimerTickMs() {
return writeBufferPeriodicFlushTimerTickMs;
}
/**
* Set the TimerTick how often the buffer timeout if checked.
+ * @deprecated Since 3.0.0, will be removed in 4.0.0. We use a common timer in the whole client
+ * implementation so you can not set it any more.
*/
+ @Deprecated
public BufferedMutatorParams setWriteBufferPeriodicFlushTimerTickMs(long timerTickMs) {
this.writeBufferPeriodicFlushTimerTickMs = timerTickMs;
return this;
@@ -141,9 +149,12 @@ public class BufferedMutatorParams implements Cloneable {
}
/**
- * @return Name of the class we will use when we construct a
- * {@link BufferedMutator} instance or null if default implementation.
+ * @return Name of the class we will use when we construct a {@link BufferedMutator} instance or
+ * null if default implementation.
+ * @deprecated Since 3.0.0, will be removed in 4.0.0. You can not set it any more as the
+ * implementation has to use too many internal stuffs in HBase.
*/
+ @Deprecated
public String getImplementationClassName() {
return this.implementationClassName;
}
@@ -151,7 +162,10 @@ public class BufferedMutatorParams implements Cloneable {
/**
* Specify a BufferedMutator implementation other than the default.
* @param implementationClassName Name of the BufferedMutator implementation class
+ * @deprecated Since 3.0.0, will be removed in 4.0.0. You can not set it any more as the
+ * implementation has to use too many internal stuffs in HBase.
*/
+ @Deprecated
public BufferedMutatorParams implementationClassName(String implementationClassName) {
this.implementationClassName = implementationClassName;
return this;
@@ -169,11 +183,6 @@ public class BufferedMutatorParams implements Cloneable {
return this;
}
- /*
- * (non-Javadoc)
- *
- * @see java.lang.Object#clone()
- */
@edu.umd.cs.findbugs.annotations.SuppressWarnings(value="CN_IDIOM_NO_SUPER_CALL",
justification="The clone below is complete")
@Override
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionOverAsyncConnection.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionOverAsyncConnection.java
index dfe7d8f..8ec7ab8 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionOverAsyncConnection.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionOverAsyncConnection.java
@@ -87,7 +87,24 @@ class ConnectionOverAsyncConnection implements Connection {
@Override
public BufferedMutator getBufferedMutator(BufferedMutatorParams params) throws IOException {
- return oldConn.getBufferedMutator(params);
+ AsyncBufferedMutatorBuilder builder = conn.getBufferedMutatorBuilder(params.getTableName());
+ if (params.getRpcTimeout() != BufferedMutatorParams.UNSET) {
+ builder.setRpcTimeout(params.getRpcTimeout(), TimeUnit.MILLISECONDS);
+ }
+ if (params.getOperationTimeout() != BufferedMutatorParams.UNSET) {
+ builder.setOperationTimeout(params.getOperationTimeout(), TimeUnit.MILLISECONDS);
+ }
+ if (params.getWriteBufferSize() != BufferedMutatorParams.UNSET) {
+ builder.setWriteBufferSize(params.getWriteBufferSize());
+ }
+ if (params.getWriteBufferPeriodicFlushTimeoutMs() != BufferedMutatorParams.UNSET) {
+ builder.setWriteBufferPeriodicFlush(params.getWriteBufferPeriodicFlushTimeoutMs(),
+ TimeUnit.MILLISECONDS);
+ }
+ if (params.getMaxKeyValueSize() != BufferedMutatorParams.UNSET) {
+ builder.setMaxKeyValueSize(params.getMaxKeyValueSize());
+ }
+ return new BufferedMutatorOverAsyncBufferedMutator(builder.build(), params.getListener());
}
@Override
diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestBufferedMutator.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestBufferedMutator.java
deleted file mode 100644
index 96bb846..0000000
--- a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestBufferedMutator.java
+++ /dev/null
@@ -1,82 +0,0 @@
-/**
- * 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.client;
-
-import static org.junit.Assert.assertTrue;
-
-import java.io.IOException;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.HBaseClassTestRule;
-import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
-import org.apache.hadoop.hbase.security.UserProvider;
-import org.apache.hadoop.hbase.testclassification.ClientTests;
-import org.apache.hadoop.hbase.testclassification.SmallTests;
-import org.junit.ClassRule;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.junit.rules.TestName;
-
-@Category({ SmallTests.class, ClientTests.class })
-public class TestBufferedMutator {
-
- @ClassRule
- public static final HBaseClassTestRule CLASS_RULE =
- HBaseClassTestRule.forClass(TestBufferedMutator.class);
-
- @Rule
- public TestName name = new TestName();
-
- /**
- * My BufferedMutator. Just to prove that I can insert a BM other than default.
- */
- public static class MyBufferedMutator extends BufferedMutatorImpl {
- MyBufferedMutator(ConnectionImplementation conn, RpcRetryingCallerFactory rpcCallerFactory,
- RpcControllerFactory rpcFactory, BufferedMutatorParams params) {
- super(conn, rpcCallerFactory, rpcFactory, params);
- }
- }
-
- @Test
- public void testAlternateBufferedMutatorImpl() throws IOException {
- BufferedMutatorParams params =
- new BufferedMutatorParams(TableName.valueOf(name.getMethodName()));
- Configuration conf = HBaseConfiguration.create();
- conf.set(AsyncRegistryFactory.REGISTRY_IMPL_CONF_KEY, DoNothingAsyncRegistry.class.getName());
- try (ConnectionImplementation connection = ConnectionFactory.createConnectionImpl(conf, null,
- UserProvider.instantiate(conf).getCurrent())) {
- BufferedMutator bm = connection.getBufferedMutator(params);
- // Assert we get default BM if nothing specified.
- assertTrue(bm instanceof BufferedMutatorImpl);
- // Now try and set my own BM implementation.
- params.implementationClassName(MyBufferedMutator.class.getName());
- bm = connection.getBufferedMutator(params);
- assertTrue(bm instanceof MyBufferedMutator);
- }
- // Now try creating a Connection after setting an alterate BufferedMutator into
- // the configuration and confirm we get what was expected.
- conf.set(BufferedMutator.CLASSNAME_KEY, MyBufferedMutator.class.getName());
- try (Connection connection = ConnectionFactory.createConnectionImpl(conf, null,
- UserProvider.instantiate(conf).getCurrent())) {
- BufferedMutator bm = connection.getBufferedMutator(params);
- assertTrue(bm instanceof MyBufferedMutator);
- }
- }
-}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestBufferedMutator.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestBufferedMutator.java
new file mode 100644
index 0000000..23e69ee
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestBufferedMutator.java
@@ -0,0 +1,90 @@
+/**
+ * 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.client;
+
+import static org.junit.Assert.assertArrayEquals;
+
+import java.io.IOException;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.testclassification.ClientTests;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+@Category({ MediumTests.class, ClientTests.class })
+public class TestBufferedMutator {
+
+ @ClassRule
+ public static final HBaseClassTestRule CLASS_RULE =
+ HBaseClassTestRule.forClass(TestBufferedMutator.class);
+
+ private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
+
+ private static TableName TABLE_NAME = TableName.valueOf("test");
+
+ private static byte[] CF = Bytes.toBytes("cf");
+
+ private static byte[] CQ = Bytes.toBytes("cq");
+
+ private static int COUNT = 1024;
+
+ private static byte[] VALUE = new byte[1024];
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ TEST_UTIL.startMiniCluster(1);
+ TEST_UTIL.createTable(TABLE_NAME, CF);
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ TEST_UTIL.shutdownMiniCluster();
+ }
+
+ @Test
+ public void test() throws Exception {
+ try (BufferedMutator mutator = TEST_UTIL.getConnection().getBufferedMutator(TABLE_NAME)) {
+ mutator.mutate(IntStream.range(0, COUNT / 2)
+ .mapToObj(i -> new Put(Bytes.toBytes(i)).addColumn(CF, CQ, VALUE))
+ .collect(Collectors.toList()));
+ mutator.flush();
+ mutator.mutate(IntStream.range(COUNT / 2, COUNT)
+ .mapToObj(i -> new Put(Bytes.toBytes(i)).addColumn(CF, CQ, VALUE))
+ .collect(Collectors.toList()));
+ mutator.close();
+ verifyData();
+ }
+ }
+
+ private void verifyData() throws IOException {
+ try (Table table = TEST_UTIL.getConnection().getTable(TABLE_NAME)) {
+ for (int i = 0; i < COUNT; i++) {
+ Result r = table.get(new Get(Bytes.toBytes(i)));
+ assertArrayEquals(VALUE, ((Result) r).getValue(CF, CQ));
+ }
+ }
+ }
+}
[hbase] 05/08: HBASE-22295 Fix TestClientOperationTimeout
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a commit to branch HBASE-21512
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit 677aba5fab4fcb3d4d8f666242d7fc57e1424de8
Author: zhangduo <zh...@apache.org>
AuthorDate: Tue Apr 23 21:54:31 2019 +0800
HBASE-22295 Fix TestClientOperationTimeout
Signed-off-by: Michael Stack <st...@apache.org>
---
.../test/java/org/apache/hadoop/hbase/TestClientOperationTimeout.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/TestClientOperationTimeout.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/TestClientOperationTimeout.java
index 6c46e50..4dd614b 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/TestClientOperationTimeout.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/TestClientOperationTimeout.java
@@ -115,7 +115,7 @@ public class TestClientOperationTimeout {
* Tests that a get on a table throws {@link SocketTimeoutException} when the operation takes
* longer than 'hbase.client.operation.timeout'.
*/
- @Test(expected = SocketTimeoutException.class)
+ @Test(expected = RetriesExhaustedException.class)
public void testGetTimeout() throws Exception {
DELAY_GET = 600;
table.get(new Get(ROW));
@@ -125,7 +125,7 @@ public class TestClientOperationTimeout {
* Tests that a put on a table throws {@link SocketTimeoutException} when the operation takes
* longer than 'hbase.client.operation.timeout'.
*/
- @Test(expected = SocketTimeoutException.class)
+ @Test(expected = RetriesExhaustedException.class)
public void testPutTimeout() throws Exception {
DELAY_MUTATE = 600;