Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-9899 for idempotent operation dups, return the result instead of throwing conflict exception (Guanghao Zhang)

This commit is contained in:
stack 2016-08-08 21:44:37 -07:00
parent 0a1a0a6883
commit 25c4ff5404
10 changed files with 352 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ProtobufUtil.java
View File

@ -851,6 +851,62 @@ public final class ProtobufUtil {
return increment;
}
/**
* Convert a protocol buffer Mutate to a Get.
* @param proto the protocol buffer Mutate to convert.
* @param cellScanner
* @return the converted client get.
* @throws IOException
*/
public static Get toGet(final MutationProto proto, final CellScanner cellScanner)
throws IOException {
MutationType type = proto.getMutateType();
assert type == MutationType.INCREMENT || type == MutationType.APPEND : type.name();
byte[] row = proto.hasRow() ? proto.getRow().toByteArray() : null;
Get get = null;
int cellCount = proto.hasAssociatedCellCount() ? proto.getAssociatedCellCount() : 0;
if (cellCount > 0) {
// The proto has metadata only and the data is separate to be found in the cellScanner.
if (cellScanner == null) {
throw new DoNotRetryIOException("Cell count of " + cellCount + " but no cellScanner: "
+ TextFormat.shortDebugString(proto));
}
for (int i = 0; i < cellCount; i++) {
if (!cellScanner.advance()) {
throw new DoNotRetryIOException("Cell count of " + cellCount + " but at index " + i
+ " no cell returned: " + TextFormat.shortDebugString(proto));
}
Cell cell = cellScanner.current();
if (get == null) {
get = new Get(Bytes.copy(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength()));
}
get.addColumn(Bytes.copy(cell.getFamilyArray(), cell.getFamilyOffset(),
cell.getFamilyLength()), Bytes.copy(cell.getQualifierArray(), cell.getQualifierOffset(),
cell.getQualifierLength()));
}
} else {
get = new Get(row);
for (ColumnValue column : proto.getColumnValueList()) {
byte[] family = column.getFamily().toByteArray();
for (QualifierValue qv : column.getQualifierValueList()) {
byte[] qualifier = qv.getQualifier().toByteArray();
if (!qv.hasValue()) {
throw new DoNotRetryIOException("Missing required field: qualifier value");
}
get.addColumn(family, qualifier);
}
}
}
if (proto.hasTimeRange()) {
TimeRange timeRange = protoToTimeRange(proto.getTimeRange());
get.setTimeRange(timeRange.getMin(), timeRange.getMax());
}
for (NameBytesPair attribute : proto.getAttributeList()) {
get.setAttribute(attribute.getName(), attribute.getValue().toByteArray());
}
return get;
}
/**
* Convert a client Scan to a protocol buffer Scan
*

70
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
View File

@ -2624,8 +2624,13 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
}
@Override
public RegionScanner getScanner(Scan scan,
List<KeyValueScanner> additionalScanners) throws IOException {
public RegionScanner getScanner(Scan scan, List<KeyValueScanner> additionalScanners)
throws IOException {
return getScanner(scan, additionalScanners, HConstants.NO_NONCE, HConstants.NO_NONCE);
}
private RegionScanner getScanner(Scan scan, List<KeyValueScanner> additionalScanners,
long nonceGroup, long nonce) throws IOException {
startRegionOperation(Operation.SCAN);
try {
// Verify families are all valid
@ -2639,7 +2644,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
checkFamily(family);
}
}
return instantiateRegionScanner(scan, additionalScanners);
return instantiateRegionScanner(scan, additionalScanners, nonceGroup, nonce);
} finally {
closeRegionOperation(Operation.SCAN);
}
@ -2647,6 +2652,12 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
protected RegionScanner instantiateRegionScanner(Scan scan,
List<KeyValueScanner> additionalScanners) throws IOException {
return instantiateRegionScanner(scan, additionalScanners, HConstants.NO_NONCE,
HConstants.NO_NONCE);
}
protected RegionScanner instantiateRegionScanner(Scan scan,
List<KeyValueScanner> additionalScanners, long nonceGroup, long nonce) throws IOException {
if (scan.isReversed()) {
if (scan.getFilter() != null) {
scan.getFilter().setReversed(true);
@ -5648,8 +5659,8 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
return region.getRegionInfo();
}
RegionScannerImpl(Scan scan, List<KeyValueScanner> additionalScanners, HRegion region)
throws IOException {
RegionScannerImpl(Scan scan, List<KeyValueScanner> additionalScanners, HRegion region,
long nonceGroup, long nonce) throws IOException {
this.region = region;
this.maxResultSize = scan.getMaxResultSize();
if (scan.hasFilter()) {
@ -5678,10 +5689,25 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
// getSmallestReadPoint, before scannerReadPoints is updated.
IsolationLevel isolationLevel = scan.getIsolationLevel();
synchronized(scannerReadPoints) {
this.readPt = getReadpoint(isolationLevel);
if (nonce == HConstants.NO_NONCE || rsServices == null
|| rsServices.getNonceManager() == null) {
this.readPt = getReadpoint(isolationLevel);
} else {
this.readPt = rsServices.getNonceManager().getMvccFromOperationContext(nonceGroup, nonce);
}
scannerReadPoints.put(this, this.readPt);
}
initializeScanners(scan, additionalScanners);
}
RegionScannerImpl(Scan scan, List<KeyValueScanner> additionalScanners, HRegion region)
throws IOException {
this(scan, additionalScanners, region, HConstants.NO_NONCE, HConstants.NO_NONCE);
}
protected void initializeScanners(Scan scan, List<KeyValueScanner> additionalScanners)
throws IOException {
// Here we separate all scanners into two lists - scanner that provide data required
// by the filter to operate (scanners list) and all others (joinedScanners list).
List<KeyValueScanner> scanners = new ArrayList<KeyValueScanner>(scan.getFamilyMap().size());
@ -6964,25 +6990,31 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
@Override
public List<Cell> get(Get get, boolean withCoprocessor) throws IOException {
return get(get, withCoprocessor, HConstants.NO_NONCE, HConstants.NO_NONCE);
}
@Override
public List<Cell> get(Get get, boolean withCoprocessor, long nonceGroup, long nonce)
throws IOException {
List<Cell> results = new ArrayList<Cell>();
// pre-get CP hook
if (withCoprocessor && (coprocessorHost != null)) {
if (coprocessorHost.preGet(get, results)) {
return results;
}
if (coprocessorHost.preGet(get, results)) {
return results;
}
}
long before = EnvironmentEdgeManager.currentTime();
long before = EnvironmentEdgeManager.currentTime();
Scan scan = new Scan(get);
RegionScanner scanner = null;
try {
scanner = getScanner(scan);
scanner = getScanner(scan, null, nonceGroup, nonce);
scanner.next(results);
} finally {
if (scanner != null)
if (scanner != null) {
scanner.close();
}
}
// post-get CP hook
@ -7456,7 +7488,12 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
}
// Do a get on the write entry... this will block until sequenceid is assigned... w/o it,
// TestAtomicOperation fails.
walKey.getWriteEntry();
WriteEntry writeEntry = walKey.getWriteEntry();
// save mvcc to this nonce's OperationContext
if (rsServices != null && rsServices.getNonceManager() != null) {
rsServices.getNonceManager().addMvccToOperationContext(nonceGroup, nonce,
writeEntry.getWriteNumber());
}
// Actually write to Memstore now
if (!tempMemstore.isEmpty()) {
@ -7682,7 +7719,12 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi
walKey = this.appendEmptyEdit(this.wal);
}
// Get WriteEntry. Will wait on assign of the sequence id.
walKey.getWriteEntry();
WriteEntry writeEntry = walKey.getWriteEntry();
// save mvcc to this nonce's OperationContext
if (rsServices != null && rsServices.getNonceManager() != null) {
rsServices.getNonceManager().addMvccToOperationContext(nonceGroup, nonce,
writeEntry.getWriteNumber());
}
// Now write to memstore, a family at a time.
for (Map.Entry<Store, List<Cell>> entry: forMemStore.entrySet()) {

64
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
View File

@ -349,11 +349,11 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
* Starts the nonce operation for a mutation, if needed.
* @param mutation Mutation.
* @param nonceGroup Nonce group from the request.
* @returns Nonce used (can be NO_NONCE).
* @returns whether to proceed this mutation.
*/
private long startNonceOperation(final MutationProto mutation, long nonceGroup)
private boolean startNonceOperation(final MutationProto mutation, long nonceGroup)
throws IOException, OperationConflictException {
if (regionServer.nonceManager == null || !mutation.hasNonce()) return HConstants.NO_NONCE;
if (regionServer.nonceManager == null || !mutation.hasNonce()) return true;
boolean canProceed = false;
try {
canProceed = regionServer.nonceManager.startOperation(
@ -361,14 +361,7 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
} catch (InterruptedException ex) {
throw new InterruptedIOException("Nonce start operation interrupted");
}
if (!canProceed) {
// TODO: instead, we could convert append/increment to get w/mvcc
String message = "The operation with nonce {" + nonceGroup + ", " + mutation.getNonce()
+ "} on row [" + Bytes.toString(mutation.getRow().toByteArray())
+ "] may have already completed";
throw new OperationConflictException(message);
}
return mutation.getNonce();
return canProceed;
}
/**
@ -530,29 +523,41 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
* Execute an append mutation.
*
* @param region
* @param m
* @param mutation
* @param cellScanner
* @param nonce group
* @return result to return to client if default operation should be
* bypassed as indicated by RegionObserver, null otherwise
* @throws IOException
*/
private Result append(final Region region, final OperationQuota quota, final MutationProto m,
final CellScanner cellScanner, long nonceGroup) throws IOException {
private Result append(final Region region, final OperationQuota quota,
final MutationProto mutation, final CellScanner cellScanner, long nonceGroup)
throws IOException {
long before = EnvironmentEdgeManager.currentTime();
Append append = ProtobufUtil.toAppend(m, cellScanner);
Append append = ProtobufUtil.toAppend(mutation, cellScanner);
quota.addMutation(append);
Result r = null;
if (region.getCoprocessorHost() != null) {
r = region.getCoprocessorHost().preAppend(append);
}
if (r == null) {
long nonce = startNonceOperation(m, nonceGroup);
boolean canProceed = startNonceOperation(mutation, nonceGroup);
boolean success = false;
try {
r = region.append(append, nonceGroup, nonce);
long nonce = mutation.hasNonce() ? mutation.getNonce() : HConstants.NO_NONCE;
if (canProceed) {
r = region.append(append, nonceGroup, nonce);
} else {
// convert duplicate append to get
List<Cell> results = region.get(ProtobufUtil.toGet(mutation, cellScanner), false,
nonceGroup, nonce);
r = Result.create(results);
}
success = true;
} finally {
endNonceOperation(m, nonceGroup, success);
if (canProceed) {
endNonceOperation(mutation, nonceGroup, success);
}
}
if (region.getCoprocessorHost() != null) {
region.getCoprocessorHost().postAppend(append, r);
@ -570,26 +575,39 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
*
* @param region
* @param mutation
* @param cellScanner
* @param nonce group
* @return the Result
* @throws IOException
*/
private Result increment(final Region region, final OperationQuota quota,
final MutationProto mutation, final CellScanner cells, long nonceGroup) throws IOException {
final MutationProto mutation, final CellScanner cellScanner, long nonceGroup)
throws IOException {
long before = EnvironmentEdgeManager.currentTime();
Increment increment = ProtobufUtil.toIncrement(mutation, cells);
Increment increment = ProtobufUtil.toIncrement(mutation, cellScanner);
quota.addMutation(increment);
Result r = null;
if (region.getCoprocessorHost() != null) {
r = region.getCoprocessorHost().preIncrement(increment);
}
if (r == null) {
long nonce = startNonceOperation(mutation, nonceGroup);
boolean canProceed = startNonceOperation(mutation, nonceGroup);
boolean success = false;
try {
r = region.increment(increment, nonceGroup, nonce);
long nonce = mutation.hasNonce() ? mutation.getNonce() : HConstants.NO_NONCE;
if (canProceed) {
r = region.increment(increment, nonceGroup, nonce);
} else {
// convert duplicate increment to get
List<Cell> results = region.get(ProtobufUtil.toGet(mutation, cellScanner), false,
nonceGroup, nonce);
r = Result.create(results);
}
success = true;
} finally {
endNonceOperation(mutation, nonceGroup, success);
if (canProceed) {
endNonceOperation(mutation, nonceGroup, success);
}
}
if (region.getCoprocessorHost() != null) {
r = region.getCoprocessorHost().postIncrement(increment, r);

11
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/Region.java
View File

@ -379,6 +379,17 @@ public interface Region extends ConfigurationObserver {
*/
List<Cell> get(Get get, boolean withCoprocessor) throws IOException;
/**
* Do a get for duplicate non-idempotent operation.
* @param get query parameters.
* @param withCoprocessor
* @param nonceGroup Nonce group.
* @param nonce Nonce.
* @return list of cells resulting from the operation
* @throws IOException
*/
List<Cell> get(Get get, boolean withCoprocessor, long nonceGroup, long nonce) throws IOException;
/**
* Return all the data for the row that matches <i>row</i> exactly,
* or the one that immediately preceeds it, at or immediately before

43
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ServerNonceManager.java
View File

@ -62,6 +62,8 @@ public class ServerNonceManager {
private static final long WAITING_BIT = 4;
private static final long ALL_FLAG_BITS = WAITING_BIT | STATE_BITS;
private long mvcc;
@Override
public String toString() {
return "[state " + getState() + ", hasWait " + hasWait() + ", activity "
@ -101,6 +103,14 @@ public class ServerNonceManager {
private long getActivityTime() {
return this.data >>> 3;
}
public void setMvcc(long mvcc) {
this.mvcc = mvcc;
}
public long getMvcc() {
return this.mvcc;
}
}
/**
@ -191,6 +201,39 @@ public class ServerNonceManager {
}
}
/**
* Store the write point in OperationContext when the operation succeed.
* @param group Nonce group.
* @param nonce Nonce.
* @param mvcc Write point of the succeed operation.
*/
public void addMvccToOperationContext(long group, long nonce, long mvcc) {
if (nonce == HConstants.NO_NONCE) {
return;
}
NonceKey nk = new NonceKey(group, nonce);
OperationContext result = nonces.get(nk);
assert result != null;
synchronized (result) {
result.setMvcc(mvcc);
}
}
/**
* Return the write point of the previous succeed operation.
* @param group Nonce group.
* @param nonce Nonce.
* @return write point of the previous succeed operation.
*/
public long getMvccFromOperationContext(long group, long nonce) {
if (nonce == HConstants.NO_NONCE) {
return Long.MAX_VALUE;
}
NonceKey nk = new NonceKey(group, nonce);
OperationContext result = nonces.get(nk);
return result == null ? Long.MAX_VALUE : result.getMvcc();
}
/**
* Reports the operation from WAL during replay.
* @param group Nonce group.

45
hbase-server/src/test/java/org/apache/hadoop/hbase/client/HConnectionTestingUtility.java
View File

@ -18,6 +18,7 @@
package org.apache.hadoop.hbase.client;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.RegionLocations;
@ -29,7 +30,11 @@ import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.protobuf.generated.AdminProtos;
import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hbase.client.ConnectionManager.HConnectionImplementation;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
@ -218,4 +223,44 @@ public class HConnectionTestingUtility {
return ConnectionManager.CONNECTION_INSTANCES.size();
}
}
/**
* This coproceesor sleep 2s at first increment/append rpc call.
*/
public static class SleepAtFirstRpcCall extends BaseRegionObserver {
static final AtomicLong ct = new AtomicLong(0);
static final String SLEEP_TIME_CONF_KEY =
"hbase.coprocessor.SleepAtFirstRpcCall.sleepTime";
static final long DEFAULT_SLEEP_TIME = 2000;
static final AtomicLong sleepTime = new AtomicLong(DEFAULT_SLEEP_TIME);
public SleepAtFirstRpcCall() {
}
@Override
public void postOpen(ObserverContext<RegionCoprocessorEnvironment> e) {
RegionCoprocessorEnvironment env = e.getEnvironment();
Configuration conf = env.getConfiguration();
sleepTime.set(conf.getLong(SLEEP_TIME_CONF_KEY, DEFAULT_SLEEP_TIME));
}
@Override
public Result postIncrement(final ObserverContext<RegionCoprocessorEnvironment> e,
final Increment increment, final Result result) throws IOException {
if (ct.incrementAndGet() == 1) {
Threads.sleep(sleepTime.get());
}
return result;
}
@Override
public Result postAppend(final ObserverContext<RegionCoprocessorEnvironment> e,
final Append append, final Result result) throws IOException {
if (ct.incrementAndGet() == 1) {
Threads.sleep(sleepTime.get());
}
return result;
}
}
}

45
hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestFromClientSide.java
View File

@ -32,6 +32,7 @@ import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@ -170,6 +171,50 @@ public class TestFromClientSide {
// Nothing to do.
}
/**
* Test append result when there are duplicate rpc request.
*/
@Test
public void testDuplicateAppend() throws Exception {
HTableDescriptor hdt = TEST_UTIL.createTableDescriptor("HCM-testDuplicateAppend");
Map<String, String> kvs = new HashMap<String, String>();
kvs.put(HConnectionTestingUtility.SleepAtFirstRpcCall.SLEEP_TIME_CONF_KEY, "2000");
hdt.addCoprocessor(HConnectionTestingUtility.SleepAtFirstRpcCall.class.getName(), null, 1, kvs);
TEST_UTIL.createTable(hdt, new byte[][] { ROW }).close();
Configuration c = new Configuration(TEST_UTIL.getConfiguration());
c.setInt(HConstants.HBASE_CLIENT_PAUSE, 50);
// Client will retry beacuse rpc timeout is small than the sleep time of first rpc call
c.setInt(HConstants.HBASE_RPC_TIMEOUT_KEY, 1500);
Connection connection = ConnectionFactory.createConnection(c);
Table t = connection.getTable(TableName.valueOf("HCM-testDuplicateAppend"));
if (t instanceof HTable) {
HTable table = (HTable) t;
table.setOperationTimeout(3 * 1000);
try {
Append append = new Append(ROW);
append.add(TEST_UTIL.fam1, QUALIFIER, VALUE);
Result result = table.append(append);
// Verify expected result
Cell[] cells = result.rawCells();
assertEquals(1, cells.length);
assertKey(cells[0], ROW, TEST_UTIL.fam1, QUALIFIER, VALUE);
// Verify expected result again
Result readResult = table.get(new Get(ROW));
cells = readResult.rawCells();
assertEquals(1, cells.length);
assertKey(cells[0], ROW, TEST_UTIL.fam1, QUALIFIER, VALUE);
} finally {
table.close();
connection.close();
}
}
}
/**
* Basic client side validation of HBASE-4536
*/

48
hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestIncrementsFromClientSide.java
View File

@ -24,6 +24,8 @@ import static org.junit.Assert.fail;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@ -32,6 +34,8 @@ import org.apache.hadoop.hbase.CategoryBasedTimeout;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.HBaseTestingUtility;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
import org.apache.hadoop.hbase.coprocessor.MultiRowMutationEndpoint;
@ -61,6 +65,7 @@ public class TestIncrementsFromClientSide {
protected final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
private static byte [] ROW = Bytes.toBytes("testRow");
private static byte [] FAMILY = Bytes.toBytes("testFamily");
private static byte [] QUALIFIER = Bytes.toBytes("testQualifier");
// This test depends on there being only one slave running at at a time. See the @Before
// method where we do rolling restart.
protected static int SLAVES = 1;
@ -89,6 +94,49 @@ public class TestIncrementsFromClientSide {
TEST_UTIL.shutdownMiniCluster();
}
/**
* Test increment result when there are duplicate rpc request.
*/
@Test
public void testDuplicateIncrement() throws Exception {
HTableDescriptor hdt = TEST_UTIL.createTableDescriptor("HCM-testDuplicateIncrement");
Map<String, String> kvs = new HashMap<String, String>();
kvs.put(HConnectionTestingUtility.SleepAtFirstRpcCall.SLEEP_TIME_CONF_KEY, "2000");
hdt.addCoprocessor(HConnectionTestingUtility.SleepAtFirstRpcCall.class.getName(), null, 1, kvs);
TEST_UTIL.createTable(hdt, new byte[][] { ROW }).close();
Configuration c = new Configuration(TEST_UTIL.getConfiguration());
c.setInt(HConstants.HBASE_CLIENT_PAUSE, 50);
// Client will retry beacuse rpc timeout is small than the sleep time of first rpc call
c.setInt(HConstants.HBASE_RPC_TIMEOUT_KEY, 1500);
Connection connection = ConnectionFactory.createConnection(c);
Table t = connection.getTable(TableName.valueOf("HCM-testDuplicateIncrement"));
if (t instanceof HTable) {
HTable table = (HTable) t;
table.setOperationTimeout(3 * 1000);
try {
Increment inc = new Increment(ROW);
inc.addColumn(TEST_UTIL.fam1, QUALIFIER, 1);
Result result = table.increment(inc);
Cell [] cells = result.rawCells();
assertEquals(1, cells.length);
assertIncrementKey(cells[0], ROW, TEST_UTIL.fam1, QUALIFIER, 1);
// Verify expected result
Result readResult = table.get(new Get(ROW));
cells = readResult.rawCells();
assertEquals(1, cells.length);
assertIncrementKey(cells[0], ROW, TEST_UTIL.fam1, QUALIFIER, 1);
} finally {
table.close();
connection.close();
}
}
}
@Test
public void testIncrementWithDeletes() throws Exception {
LOG.info("Starting " + this.name.getMethodName());

13
hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestMultiParallel.java
View File

@ -516,14 +516,14 @@ public class TestMultiParallel {
table.increment(inc);
inc = new Increment(ONE_ROW);
inc.addColumn(BYTES_FAMILY, QUALIFIER, 1L);
try {
table.increment(inc);
fail("Should have thrown an exception");
} catch (OperationConflictException ex) {
}
// duplicate increment
Result result = table.increment(inc);
validateResult(result, QUALIFIER, Bytes.toBytes(1L));
Get get = new Get(ONE_ROW);
get.addColumn(BYTES_FAMILY, QUALIFIER);
Result result = table.get(get);
result = table.get(get);
validateResult(result, QUALIFIER, Bytes.toBytes(1L));
// Now run a bunch of requests in parallel, exactly half should succeed.
@ -551,7 +551,6 @@ public class TestMultiParallel {
}
try {
table.increment(inc);
} catch (OperationConflictException ex) { // Some threads are expected to fail.
} catch (IOException ioEx) {
fail("Not expected");
}

2
hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestScannerHeartbeatMessages.java
View File

@ -492,7 +492,7 @@ public class TestScannerHeartbeatMessages {
// Instantiate the custom heartbeat region scanners
@Override
protected RegionScanner instantiateRegionScanner(Scan scan,
List<KeyValueScanner> additionalScanners) throws IOException {
List<KeyValueScanner> additionalScanners, long nonceGroup, long nonce) throws IOException {
if (scan.isReversed()) {
if (scan.getFilter() != null) {
scan.getFilter().setReversed(true);