You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@trafodion.apache.org by sa...@apache.org on 2017/05/12 20:25:43 UTC
[08/22] incubator-trafodion git commit: This is a large contribution
of changes from Esgyn TransactionManager and libraries that are collectively
much better tested and hardened than Trafodion,
but are too numerous and complex to cherry pick individually
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/91794b53/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionRegionLocation.java
----------------------------------------------------------------------
diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionRegionLocation.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionRegionLocation.java
index d3118db..38f598b 100644
--- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionRegionLocation.java
+++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionRegionLocation.java
@@ -28,7 +28,7 @@ import org.apache.commons.logging.LogFactory;
import org.apache.commons.codec.binary.Hex;
import org.apache.hadoop.hbase.util.Bytes;
-
+import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.ServerName;
@@ -63,62 +63,79 @@ public class TransactionRegionLocation extends HRegionLocation {
@Override
public int compareTo(HRegionLocation o) {
- if (LOG.isTraceEnabled()) LOG.trace("compareTo ENTRY: " + o);
- int result = super.getHostname().compareTo(o.getHostname());
+ if (o == null) {
+ if (LOG.isDebugEnabled()) LOG.debug("CompareTo TransactionRegionLocation object is null");
+ return 1;
+ }
+
+ if (LOG.isDebugEnabled()) LOG.debug("CompareTo TransactionRegionLocation Entry: TableNames :\n mine: "
+ + this.getRegionInfo().getTable().getNameAsString() + "\n object's : " + o.getRegionInfo().getTable().getNameAsString());
+
+ // Make sure this is the same table
+ int result = this.getRegionInfo().getTable().compareTo(o.getRegionInfo().getTable());
if (result != 0){
- if (LOG.isTraceEnabled()) LOG.trace("compareTo hostnames differ: mine: " + super.getHostname() + " hex: " +
- Hex.encodeHexString(Bytes.toBytes(super.getHostname())) + " object's: " + o.getHostname() + " hex: " + Hex.encodeHexString(Bytes.toBytes(o.getHostname())));
+ if (LOG.isDebugEnabled()) LOG.debug("compareTo TransactionRegionLocation TableNames are different: result is " + result);
return result;
}
- result = super.getPort() - o.getPort();
- if (result != 0) {
- if (LOG.isTraceEnabled()) LOG.trace("compareTo ports differ: mine: " + super.getPort() + " object's: " + o.getPort());
- return result;
+
+ if (LOG.isDebugEnabled()) LOG.debug("Tables match - comparing keys for "
+ + this.getRegionInfo().getTable().getNameAsString()
+ + "\n This start key : " + Hex.encodeHexString(this.getRegionInfo().getStartKey())
+ + "\n Object's start key: " + Hex.encodeHexString(o.getRegionInfo().getStartKey())
+ + "\n This end key : " + Hex.encodeHexString(this.getRegionInfo().getEndKey())
+ + "\n Object's end key: " + Hex.encodeHexString(o.getRegionInfo().getEndKey()));
+
+ // Here we are going to compare the keys as a range we can return 0
+ // For these comparisons it's important to remember that 'this' is the object that is being added
+ // and that 'object' is an object already added in the participationRegions set.
+ //
+ // We are trying to limit the registration of daughter regions after a region split.
+ // So if a location is already added whose startKey is less than ours and whose end
+ // key is greater than ours we will return 0 so that 'this' does not get added into
+ // the participatingRegions list.
+
+ // firstKeyInRange will be true if object's startKey is less than ours.
+ int startKeyResult = Bytes.compareTo(this.getRegionInfo().getStartKey(), o.getRegionInfo().getStartKey());
+ boolean firstKeyInRange = startKeyResult >= 0;
+ boolean objLastKeyInfinite = Bytes.equals(o.getRegionInfo().getEndKey(), HConstants.EMPTY_END_ROW);
+ boolean thisLastKeyInfinite = Bytes.equals(this.getRegionInfo().getEndKey(), HConstants.EMPTY_END_ROW);
+ int endKeyResult = Bytes.compareTo(this.getRegionInfo().getEndKey(), o.getRegionInfo().getEndKey());
+
+ // lastKey is in range if the existing object has an infinite end key, no matter what this end key is.
+ boolean lastKeyInRange = objLastKeyInfinite || ( ! thisLastKeyInfinite && endKeyResult <= 0);
+ if (LOG.isDebugEnabled()) LOG.debug("firstKeyInRange " + firstKeyInRange + " lastKeyInRange " + lastKeyInRange);
+
+ if (firstKeyInRange && lastKeyInRange) {
+ if (LOG.isDebugEnabled()) LOG.debug("Object's region contains this region's start and end keys. Regions match for "
+ + o.getRegionInfo().getTable().getNameAsString());
+ return 0;
}
- result = super.getRegionInfo().compareTo(o.getRegionInfo());
- if (result != 0) {
- if (LOG.isTraceEnabled()) LOG.trace("compareTo regionInfo differs: mine: " + super.getRegionInfo() + " hex: " +
- Hex.encodeHexString(super.getRegionInfo().getRegionName()) + " object's: " + o.getRegionInfo() + "hex: " +
- Hex.encodeHexString(o.getRegionInfo().getRegionName()));
+
+ if (startKeyResult != 0){
+ if (LOG.isDebugEnabled()) LOG.debug("compareTo TransactionRegionLocation startKeys don't match: result is " + startKeyResult);
+ return startKeyResult;
}
- if (LOG.isTraceEnabled()) LOG.trace("compareTo HRegionLocation result is: " + result );
- return result;
- }
- public int compareTo(TransactionRegionLocation o) {
- if (LOG.isTraceEnabled()) LOG.trace("compareTo TransactionRegionLocation ENTRY: " + o.getRegionInfo().getRegionNameAsString());
- int result = super.getHostname().compareTo(o.getHostname());
- if (result != 0){
- if (LOG.isTraceEnabled()) LOG.trace("compareTo hostnames differ: mine: " + super.getHostname() + " hex: " +
- Hex.encodeHexString(Bytes.toBytes(super.getHostname())) + " object's: " + o.getHostname() + " hex: " + Hex.encodeHexString(Bytes.toBytes(o.getHostname())));
- return result;
+ if (objLastKeyInfinite) {
+ if (LOG.isInfoEnabled()) LOG.info("Object's region contains this region's end keys for "
+ + o.getRegionInfo().getTable().getNameAsString()
+ + "\n This start key : " + Hex.encodeHexString(this.getRegionInfo().getStartKey())
+ + "\n Object's start key: " + Hex.encodeHexString(o.getRegionInfo().getStartKey())
+ + "\n This end key : " + Hex.encodeHexString(this.getRegionInfo().getEndKey())
+ + "\n Object's end key: " + Hex.encodeHexString(o.getRegionInfo().getEndKey()));
}
- result = super.getPort() - o.getPort();
- if (result != 0) {
- if (LOG.isTraceEnabled()) LOG.trace("compareTo ports differ: mine: " + super.getPort() + " object's: " + o.getPort());
- return result;
+
+ if (this.getRegionInfo().getStartKey().length != 0 && this.getRegionInfo().getEndKey().length == 0) {
+ if (LOG.isDebugEnabled()) LOG.debug("compareTo TransactionRegionLocation \"this\" is the last region: result is 1");
+ return 1; // this is last region
}
- result = super.getRegionInfo().compareTo(o.getRegionInfo());
- if (result != 0) {
- if (super.getRegionInfo().getEncodedName().compareTo(o.getRegionInfo().getEncodedName()) == 0){
- if (LOG.isTraceEnabled()) LOG.trace("compareTo regionInfo RegionNames match " + o.getRegionInfo().getEncodedName());
- if (super.getRegionInfo().containsRange(o.getRegionInfo().getStartKey(), o.getRegionInfo().getEndKey())) {
- if (LOG.isTraceEnabled()) LOG.trace("This region contains object's start and end keys. Regions match " + o.getRegionInfo().getEncodedName());
- result = 0;
- return result;
- }
- if (LOG.isTraceEnabled()) LOG.trace("compareTo TransactionRegionLocation regionInfo RegionNames match, but object end keys differ:\n this hex name: "
- + Hex.encodeHexString(super.getRegionInfo().getRegionName()) + "\n obj hex name: "
- + Hex.encodeHexString(o.getRegionInfo().getRegionName())
- + "\n This end key : " + Hex.encodeHexString(super.getRegionInfo().getEndKey())
- + "\n Object's end key: " + Hex.encodeHexString(o.getRegionInfo().getEndKey()) + "\n result " + result);
- }
- else {
- if (LOG.isTraceEnabled()) LOG.trace("compareTo TransactionRegionLocation regionInfo.getEncodedName differs:\n mine: "
- + super.getRegionInfo().getEncodedName() + "\n object's : " + o.getRegionInfo().getEncodedName() + " result is " + result);
- }
+ if (o.getRegionInfo().getStartKey().length != 0 && o.getRegionInfo().getEndKey().length == 0) {
+ if (LOG.isDebugEnabled()) LOG.debug("compareTo TransactionRegionLocation \"object\" is the last region: result is -1");
+ return -1; // o is the last region
}
- return result;
+ if (LOG.isDebugEnabled()) LOG.debug("compareTo TransactionRegionLocation endKeys comparison: result is " + endKeyResult);
+ return endKeyResult;
+
}
/**
@@ -126,20 +143,32 @@ public class TransactionRegionLocation extends HRegionLocation {
*/
@Override
public boolean equals(Object o) {
- if (LOG.isTraceEnabled()) LOG.trace("equals ENTRY: this: " + this + " o: " + o);
+ if (LOG.isDebugEnabled()) LOG.debug("equals ENTRY: this: " + this + " o: " + o);
if (this == o) {
- if (LOG.isTraceEnabled()) LOG.trace("equals same object: " + o);
+ if (LOG.isDebugEnabled()) LOG.debug("equals same object: " + o);
return true;
}
if (o == null) {
- if (LOG.isTraceEnabled()) LOG.trace("equals o is null");
+ if (LOG.isDebugEnabled()) LOG.debug("equals o is null");
return false;
}
- if (!(o instanceof TransactionRegionLocation)) {
- if (LOG.isTraceEnabled()) LOG.trace("equals o is not an instance of: " + o);
+ if (!(o instanceof HRegionLocation)) {
+ if (LOG.isDebugEnabled()) LOG.debug("equals o is not an instance of: " + o);
return false;
}
- return this.compareTo((TransactionRegionLocation)o) == 0;
+ return this.compareTo((HRegionLocation)o) == 0;
}
+ /**
+ * toString
+ * @return String this
+ *
+ */
+ @Override
+ public String toString() {
+ return super.toString() + "encodedName " + super.getRegionInfo().getEncodedName()
+ + " start key: " + Hex.encodeHexString(super.getRegionInfo().getStartKey())
+ + " end key: " + Hex.encodeHexString(super.getRegionInfo().getEndKey())
+ + " tableRecodedDropped " + isTableRecodedDropped();
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/91794b53/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionState.java
----------------------------------------------------------------------
diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionState.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionState.java
index fee5d54..4b5fdf9 100644
--- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionState.java
+++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionState.java
@@ -33,6 +33,7 @@ import org.apache.commons.codec.binary.Hex;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.HConstants;
@@ -45,11 +46,16 @@ public class TransactionState {
static final Log LOG = LogFactory.getLog(TransactionState.class);
+ // Current transactionId has the following composition:
+ // int sequenceId
+ // short nodeId
+ // short clusterId
private final long transactionId;
private TransState status;
private long startEpoch;
private long startId;
private long commitId;
+ private long recoveryASN;
/**
*
@@ -71,9 +77,11 @@ public class TransactionState {
private boolean ddlTrans;
private static boolean useConcurrentHM = false;
private boolean hasRetried = false;
- private boolean uteLogged = false;
+ private boolean exceptionLogged = false;
+ private long nodeId;
+ private long clusterId;
private String recordException;
-
+ private static long TM_MAX_REGIONSERVER_STRING = 3072;
public Set<String> tableNames = Collections.synchronizedSet(new HashSet<String>());
public Set<TransactionRegionLocation> participatingRegions;
/**
@@ -82,7 +90,7 @@ public class TransactionState {
public Set<TransactionRegionLocation> regionsToIgnore = Collections.synchronizedSet(new HashSet<TransactionRegionLocation>());
private Set<TransactionRegionLocation> retryRegions = Collections.synchronizedSet(new HashSet<TransactionRegionLocation>());
- private native void registerRegion(long transid, long startId, int port, byte[] hostname, long startcode, byte[] regionInfo);
+ private native void registerRegion(long transid, long startid, int port, byte[] hostname, long startcode, byte[] regionInfo);
public boolean islocalTransaction() {
return localTransaction;
@@ -115,6 +123,9 @@ public class TransactionState {
commitSendDone = false;
hasError = null;
ddlTrans = false;
+ recoveryASN = -1;
+ setNodeId();
+ setClusterId();
recordException = null;
if(useConcurrentHM) {
@@ -279,15 +290,13 @@ public class TransactionState {
recordException = null;
}
- public void registerLocation(final TransactionRegionLocation location) throws IOException {
+ // Used at the client end - the one performing the mutation - e.g. the SQL process
+ public void registerLocation(final TransactionRegionLocation location) throws IOException {
byte [] lv_hostname = location.getHostname().getBytes();
int lv_port = location.getPort();
long lv_startcode = location.getServerName().getStartcode();
+ HRegionInfo regionInfo = location.getRegionInfo();
- /* ByteArrayOutputStream lv_bos = new ByteArrayOutputStream();
- DataOutputStream lv_dos = new DataOutputStream(lv_bos);
- location.getRegionInfo().write(lv_dos);
- lv_dos.flush(); */
byte [] lv_byte_region_info = location.getRegionInfo().toByteArray();
if (LOG.isTraceEnabled()) LOG.trace("TransactionState.registerLocation: [" + location.getRegionInfo().getEncodedName() +
"], endKey: " + Hex.encodeHexString(location.getRegionInfo().getEndKey()) + " transaction [" + transactionId + "]");
@@ -296,27 +305,50 @@ public class TransactionState {
if(LOG.isTraceEnabled()) LOG.trace("TransactionState.registerLocation local transaction not sending registerRegion.");
}
else {
- if (LOG.isTraceEnabled()) LOG.trace("TransactionState.registerLocation global transaction registering region for ts: " + transactionId + " and startId: " + startId);
+ if (lv_byte_region_info.length > TM_MAX_REGIONSERVER_STRING){
+ String skey = (Bytes.equals(location.getRegionInfo().getStartKey(), HConstants.EMPTY_START_ROW)) ? "skey=null" : ("skey=" + Hex.encodeHexString(location.getRegionInfo().getStartKey()));
+ String ekey = (Bytes.equals(location.getRegionInfo().getEndKey(), HConstants.EMPTY_END_ROW)) ? "ekey=null" : ("ekey=" + Hex.encodeHexString(location.getRegionInfo().getEndKey()));
+ IOException ioe = new IOException("RegionInfo is too large (" + lv_byte_region_info.length
+ + "), try reducing table keys for "
+ + location.getRegionInfo().getTable().getNameAsString()
+ + " skey: " + skey + " ekey: " + ekey);
+ LOG.error("RegionInfo is too large (" + lv_byte_region_info.length
+ + "), try reducing table keys for "
+ + location.getRegionInfo().getTable().getNameAsString()
+ + " skey: " + skey + " ekey: " + ekey, ioe);
+ throw ioe;
+ }
+ if (LOG.isDebugEnabled()){
+ String skey = (Bytes.equals(location.getRegionInfo().getStartKey(), HConstants.EMPTY_START_ROW)) ? "skey=null" : ("skey=" + Hex.encodeHexString(location.getRegionInfo().getStartKey()));
+ String ekey = (Bytes.equals(location.getRegionInfo().getEndKey(), HConstants.EMPTY_END_ROW)) ? "ekey=null" : ("ekey=" + Hex.encodeHexString(location.getRegionInfo().getEndKey()));
+ LOG.debug("TransactionState.registerLocation global transaction registering region "
+ + location.getRegionInfo().getTable().getNameAsString()
+ + " using byte array size: " + lv_byte_region_info.length + " for ts: "
+ + transactionId + " and startId: " + startId + " skey: " + skey + " ekey: " + ekey + " ");
+ }
registerRegion(transactionId, startId, lv_port, lv_hostname, lv_startcode, lv_byte_region_info);
}
}
public boolean addRegion(final TransactionRegionLocation trRegion) {
- if (LOG.isTraceEnabled()) LOG.trace("addRegion ENTRY with trRegion [" + trRegion.getRegionInfo().getEncodedName() + "], endKey: "
+ if (LOG.isDebugEnabled()) LOG.debug("addRegion ENTRY with trRegion [" + trRegion.getRegionInfo().getRegionNameAsString()
+ + "], startKey: " + Hex.encodeHexString(trRegion.getRegionInfo().getStartKey()) + "], endKey: "
+ Hex.encodeHexString(trRegion.getRegionInfo().getEndKey()) + " and transaction [" + transactionId + "]");
boolean added = participatingRegions.add(trRegion);
if (added) {
- if (LOG.isTraceEnabled()) LOG.trace("Added new trRegion (#" + participatingRegions.size()
+ if (LOG.isDebugEnabled()) LOG.debug("Added new trRegion (#" + participatingRegions.size()
+ ") to participatingRegions [" + trRegion.getRegionInfo().getRegionNameAsString()
- + "], endKey: " + Hex.encodeHexString(trRegion.getRegionInfo().getEndKey())
+ + "], startKey: " + Hex.encodeHexString(trRegion.getRegionInfo().getStartKey())
+ + "], endKey: " + Hex.encodeHexString(trRegion.getRegionInfo().getEndKey())
+ " and transaction [" + transactionId + "]");
}
else {
- if (LOG.isTraceEnabled()) LOG.trace("trRegion already present in (" + participatingRegions.size()
- + ") participatinRegions [" + trRegion.getRegionInfo().getEncodedName()
- + "], endKey: " + Hex.encodeHexString(trRegion.getRegionInfo().getEndKey())
+ if (LOG.isDebugEnabled()) LOG.debug("trRegion already present in (" + participatingRegions.size()
+ + ") participatinRegions [" + trRegion.getRegionInfo().getRegionNameAsString()
+ + "], startKey: " + Hex.encodeHexString(trRegion.getRegionInfo().getStartKey())
+ + "], endKey: " + Hex.encodeHexString(trRegion.getRegionInfo().getEndKey())
+ " and transaction [" + transactionId + "]");
}
@@ -325,10 +357,10 @@ public class TransactionState {
public boolean addRegion(final HRegionLocation hregion) {
- if (LOG.isTraceEnabled()) LOG.trace("Creating new TransactionRegionLocation from HRegionLocation [" + hregion.getRegionInfo().getRegionNameAsString() +
+ if (LOG.isDebugEnabled()) LOG.debug("Creating new TransactionRegionLocation from HRegionLocation [" + hregion.getRegionInfo().getRegionNameAsString() +
" endKey: " + Hex.encodeHexString(hregion.getRegionInfo().getEndKey()) + " for transaction [" + transactionId + "]");
TransactionRegionLocation trRegion = new TransactionRegionLocation(hregion.getRegionInfo(),
- hregion.getServerName());
+ hregion.getServerName());
// Save hregion for now
boolean added = participatingRegions.add(trRegion);
@@ -409,8 +441,16 @@ public class TransactionState {
* @return Return the nodeId.
*/
public long getNodeId() {
+ return nodeId;
+ }
+
+ /**
+ * Set the originating node of the transaction.
+ *
+ */
+ private void setNodeId() {
- return ((transactionId >> 32) & 0xFFL);
+ nodeId = ((transactionId >> 32) & 0xFFL);
}
/**
@@ -424,6 +464,37 @@ public class TransactionState {
}
/**
+ * Get the originating clusterId of the transaction.
+ *
+ * @return Return the clusterId.
+ */
+ public long getClusterId() {
+
+ return clusterId;
+
+ }
+
+
+ /**
+ * Get the originating clusterId of the passed in transaction.
+ *
+ * @return Return the clusterId.
+ */
+ public static long getClusterId(long transId) {
+
+ return (transId >> 48);
+ }
+
+ /**
+ * Set the originating clusterId of the passed in transaction.
+ *
+ */
+ private void setClusterId() {
+
+ clusterId = (transactionId >> 48);
+ }
+
+ /**
* Set the startEpoc.
*
*/
@@ -475,6 +546,23 @@ public class TransactionState {
}
/**
+ * Set the recoveryASN.
+ *
+ */
+ public void setRecoveryASN(final long value) {
+ this.recoveryASN = value;
+ }
+
+ /**
+ * Get the recoveryASN.
+ *
+ * @return Return the recoveryASN.
+ */
+ public long getRecoveryASN() {
+ return recoveryASN;
+ }
+
+ /**
* @see java.lang.Object#toString()
*/
@Override
@@ -482,7 +570,7 @@ public class TransactionState {
return "transactionId: " + transactionId + ", startId: " + startId + ", commitId: " + commitId +
", startEpoch: " + startEpoch + ", participants: " + participatingRegions.size()
+ ", ignoring: " + regionsToIgnore.size() + ", hasDDL: " + hasDDLTx()
- + ", state: " + status.toString();
+ + ", recoveryASN: " + getRecoveryASN() + ", state: " + status.toString();
}
public int getParticipantCount() {
@@ -521,18 +609,26 @@ public class TransactionState {
return this.hasRetried;
}
- public void logUteDetails()
+ public boolean hasPlaceHolder() {
+ return false;
+ }
+
+ public synchronized void logExceptionDetails(final boolean ute)
{
- if (uteLogged)
+ if (exceptionLogged)
return;
int participantNum = 0;
byte[] startKey;
byte[] endKey_orig;
byte[] endKey;
+ boolean isIgnoredRegion = false;
+ LOG.error("Starting " + (ute == true ? "UTE " : "NPTE ") + "for trans: " + this.toString());
for (TransactionRegionLocation location : getParticipatingRegions()) {
- participantNum++;
- final byte[] regionName = location.getRegionInfo().getRegionName();
+ isIgnoredRegion = getRegionsToIgnore().contains(location);
+ if (! isIgnoredRegion) {
+ participantNum++;
+ }
startKey = location.getRegionInfo().getStartKey();
endKey_orig = location.getRegionInfo().getEndKey();
@@ -541,13 +637,16 @@ public class TransactionState {
else
endKey = TransactionManager.binaryIncrementPos(endKey_orig, -1);
- LOG.warn("UTE for transId: " + getTransactionId()
- + " participantNum " + participantNum
+ LOG.error((ute == true ? "UTE " : "NPTE ") + "for transId: " + getTransactionId()
+ + " participantNum " + (isIgnoredRegion ? " Ignored region " : participantNum)
+ " location " + location.getRegionInfo().getRegionNameAsString()
- + " startKey " + ((startKey != null)? Hex.encodeHexString(startKey) : "NULL")
- + " endKey " + ((endKey != null) ? Hex.encodeHexString(endKey) : "NULL")
- + " RegionEndKey " + ((endKey_orig != null) ? Hex.encodeHexString(endKey_orig) : "NULL"));
+ + " startKey " + ((startKey != null)?
+ (Bytes.equals(startKey, HConstants.EMPTY_START_ROW) ? "INFINITE" : Hex.encodeHexString(startKey)) : "NULL")
+ + " endKey " + ((endKey != null) ?
+ (Bytes.equals(endKey, HConstants.EMPTY_END_ROW) ? "INFINITE" : Hex.encodeHexString(endKey)) : "NULL")
+ + " RegionEndKey " + ((endKey_orig != null) ?
+ (Bytes.equals(endKey_orig, HConstants.EMPTY_END_ROW) ? "INFINITE" : Hex.encodeHexString(endKey_orig)) : "NULL"));
}
- uteLogged = true;
+ exceptionLogged = true;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/91794b53/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalAggregationClient.java
----------------------------------------------------------------------
diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalAggregationClient.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalAggregationClient.java
index 8d54246..52edcd9 100644
--- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalAggregationClient.java
+++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalAggregationClient.java
@@ -98,6 +98,7 @@ public class TransactionalAggregationClient {
* given range. In case qualifier is null, a max of all values for the given
* family is returned.
* @param transactionId
+ * @param startId
* @param tableName
* @param ci
* @param scan
@@ -107,12 +108,12 @@ public class TransactionalAggregationClient {
* & propagated to it.
*/
public <R, S, P extends Message, Q extends Message, T extends Message> R max(
- final long transactionId, final TableName tableName, final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan)
+ final long transactionId, final long startId, final TableName tableName, final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan)
throws Throwable {
TransactionalTable table = null;
try {
table = new TransactionalTable(tableName.getName(), connection);
- return max(transactionId, table, ci, scan);
+ return max(transactionId, startId, table, ci, scan);
} finally {
if (table != null) {
table.close();
@@ -133,12 +134,12 @@ public class TransactionalAggregationClient {
* & propagated to it.
*/
public <R, S, P extends Message, Q extends Message, T extends Message>
- R max(final long transactionId, final TransactionalTable table, final ColumnInterpreter<R, S, P, Q, T> ci,
+ R max(final long transactionId, final long startId, final TransactionalTable table, final ColumnInterpreter<R, S, P, Q, T> ci,
final Scan scan) throws Throwable {
byte [] currentBeginKey = scan.getStartRow();
HRegionInfo currentRegion = table.getRegionLocation(currentBeginKey).getRegionInfo();
com.google.protobuf.ByteString regionName = ByteString.copyFromUtf8(currentRegion.getRegionNameAsString());
- final TransactionalAggregateRequest requestArg = validateArgAndGetPB(regionName, transactionId, scan, ci, false);
+ final TransactionalAggregateRequest requestArg = validateArgAndGetPB(regionName, transactionId, startId, scan, ci, false);
class MaxCallBack implements Batch.Callback<R> {
R max = null;
@@ -199,6 +200,7 @@ public class TransactionalAggregationClient {
* given range. In case qualifier is null, a min of all values for the given
* family is returned.
* @param transactionId
+ * @param startId
* @param tableName
* @param ci
* @param scan
@@ -206,12 +208,12 @@ public class TransactionalAggregationClient {
* @throws Throwable
*/
public <R, S, P extends Message, Q extends Message, T extends Message> R min(
- final long transactionId, final TableName tableName, final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan)
+ final long transactionId, final long startId, final TableName tableName, final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan)
throws Throwable {
TransactionalTable table = null;
try {
table = new TransactionalTable(tableName.getName(), connection);
- return min(transactionId, table, ci, scan);
+ return min(transactionId, startId, table, ci, scan);
} finally {
if (table != null) {
table.close();
@@ -230,12 +232,12 @@ public class TransactionalAggregationClient {
* @throws Throwable
*/
public <R, S, P extends Message, Q extends Message, T extends Message>
- R min(final long transactionId, final TransactionalTable table, final ColumnInterpreter<R, S, P, Q, T> ci,
+ R min(final long transactionId, final long startId, final TransactionalTable table, final ColumnInterpreter<R, S, P, Q, T> ci,
final Scan scan) throws Throwable {
byte [] currentBeginKey = scan.getStartRow();
HRegionInfo currentRegion = table.getRegionLocation(currentBeginKey).getRegionInfo();
com.google.protobuf.ByteString regionName = ByteString.copyFromUtf8(currentRegion.getRegionNameAsString());
- final TransactionalAggregateRequest requestArg = validateArgAndGetPB(regionName, transactionId, scan, ci, false);
+ final TransactionalAggregateRequest requestArg = validateArgAndGetPB(regionName, transactionId, startId, scan, ci, false);
class MinCallBack implements Batch.Callback<R> {
private R min = null;
@@ -283,6 +285,7 @@ public class TransactionalAggregationClient {
* not of the given filter: in this case, this particular row will not be
* counted ==> an error.
* @param transactionId
+ * @param startId
* @param tableName
* @param ci
* @param scan
@@ -291,6 +294,7 @@ public class TransactionalAggregationClient {
*/
public <R, S, P extends Message, Q extends Message, T extends Message> long rowCount(
final long transactionId,
+ final long startId,
final TableName tableName,
final ColumnInterpreter<R, S, P, Q, T> ci,
final Scan scan)
@@ -298,7 +302,7 @@ public class TransactionalAggregationClient {
TransactionalTable table = null;
try {
table = new TransactionalTable(tableName.getName(), connection);
- return rowCount(transactionId, table, ci, scan);
+ return rowCount(transactionId, startId, table, ci, scan);
} finally {
if (table != null) {
table.close();
@@ -320,12 +324,12 @@ public class TransactionalAggregationClient {
* @throws Throwable
*/
public <R, S, P extends Message, Q extends Message, T extends Message>
- long rowCount(final long transactionId, final TransactionalTable table,
+ long rowCount(final long transactionId, final long startId, final TransactionalTable table,
final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan) throws Throwable {
byte [] currentBeginKey = scan.getStartRow();
HRegionInfo currentRegion = table.getRegionLocation(currentBeginKey).getRegionInfo();
com.google.protobuf.ByteString regionName = ByteString.copyFromUtf8(currentRegion.getRegionNameAsString());
- final TransactionalAggregateRequest requestArg = validateArgAndGetPB(regionName, transactionId, scan, ci, true);
+ final TransactionalAggregateRequest requestArg = validateArgAndGetPB(regionName, transactionId, startId, scan, ci, true);
class RowNumCallback implements Batch.Callback<Long> {
private final AtomicLong rowCountL = new AtomicLong(0);
@@ -364,6 +368,7 @@ public class TransactionalAggregationClient {
* It sums up the value returned from various regions. In case qualifier is
* null, summation of all the column qualifiers in the given family is done.
* @param transactionId
+ * @param startId
* @param tableName
* @param ci
* @param scan
@@ -371,12 +376,12 @@ public class TransactionalAggregationClient {
* @throws Throwable
*/
public <R, S, P extends Message, Q extends Message, T extends Message> S sum(
- final long transactionId, final TableName tableName, final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan)
+ final long transactionId, final long startId, final TableName tableName, final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan)
throws Throwable {
TransactionalTable table = null;
try {
table = new TransactionalTable(tableName.getName(), connection);
- return sum(transactionId, table, ci, scan);
+ return sum(transactionId, startId, table, ci, scan);
} finally {
if (table != null) {
table.close();
@@ -394,12 +399,12 @@ public class TransactionalAggregationClient {
* @throws Throwable
*/
public <R, S, P extends Message, Q extends Message, T extends Message>
- S sum(final long transactionId, final TransactionalTable table, final ColumnInterpreter<R, S, P, Q, T> ci,
+ S sum(final long transactionId, final long startId, final TransactionalTable table, final ColumnInterpreter<R, S, P, Q, T> ci,
final Scan scan) throws Throwable {
byte [] currentBeginKey = scan.getStartRow();
HRegionInfo currentRegion = table.getRegionLocation(currentBeginKey).getRegionInfo();
com.google.protobuf.ByteString regionName = ByteString.copyFromUtf8(currentRegion.getRegionNameAsString());
- final TransactionalAggregateRequest requestArg = validateArgAndGetPB(regionName, transactionId, scan, ci, false);
+ final TransactionalAggregateRequest requestArg = validateArgAndGetPB(regionName, transactionId, startId, scan, ci, false);
class SumCallBack implements Batch.Callback<S> {
S sumVal = null;
@@ -443,17 +448,18 @@ public class TransactionalAggregationClient {
* corresponding regions. Approach is to compute a global sum of region level
* sum and rowcount and then compute the average.
* @param transactionId
+ * @param startId
* @param tableName
* @param scan
* @throws Throwable
*/
private <R, S, P extends Message, Q extends Message, T extends Message> Pair<S, Long> getAvgArgs(
- final long transactionId, final TableName tableName, final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan)
+ final long transactionId, final long startId, final TableName tableName, final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan)
throws Throwable {
TransactionalTable table = null;
try {
table = new TransactionalTable(tableName.getName(), connection);
- return getAvgArgs(transactionId, table, ci, scan);
+ return getAvgArgs(transactionId, startId, table, ci, scan);
} finally {
if (table != null) {
table.close();
@@ -470,12 +476,12 @@ public class TransactionalAggregationClient {
* @throws Throwable
*/
private <R, S, P extends Message, Q extends Message, T extends Message>
- Pair<S, Long> getAvgArgs(final long transactionId, final TransactionalTable table,
+ Pair<S, Long> getAvgArgs(final long transactionId, final long startId, final TransactionalTable table,
final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan) throws Throwable {
byte [] currentBeginKey = scan.getStartRow();
HRegionInfo currentRegion = table.getRegionLocation(currentBeginKey).getRegionInfo();
com.google.protobuf.ByteString regionName = ByteString.copyFromUtf8(currentRegion.getRegionNameAsString());
- final TransactionalAggregateRequest requestArg = validateArgAndGetPB(regionName, transactionId, scan, ci, false);
+ final TransactionalAggregateRequest requestArg = validateArgAndGetPB(regionName, transactionId, startId, scan, ci, false);
class AvgCallBack implements Batch.Callback<Pair<S, Long>> {
S sum = null;
Long rowCount = 0l;
@@ -528,6 +534,7 @@ public class TransactionalAggregationClient {
* columninterpreter says. So, this methods collects the necessary parameters
* to compute the average and returs the double value.
* @param transactionId
+ * @param startId
* @param tableName
* @param ci
* @param scan
@@ -535,9 +542,9 @@ public class TransactionalAggregationClient {
* @throws Throwable
*/
public <R, S, P extends Message, Q extends Message, T extends Message>
- double avg( final long transactionId, final TableName tableName,
+ double avg( final long transactionId, final long startId, final TableName tableName,
final ColumnInterpreter<R, S, P, Q, T> ci, Scan scan) throws Throwable {
- Pair<S, Long> p = getAvgArgs(transactionId, tableName, ci, scan);
+ Pair<S, Long> p = getAvgArgs(transactionId, startId, tableName, ci, scan);
return ci.divideForAvg(p.getFirst(), p.getSecond());
}
@@ -554,8 +561,8 @@ public class TransactionalAggregationClient {
* @throws Throwable
*/
public <R, S, P extends Message, Q extends Message, T extends Message> double avg(
- final long transactionId, final TransactionalTable table, final ColumnInterpreter<R, S, P, Q, T> ci, Scan scan) throws Throwable {
- Pair<S, Long> p = getAvgArgs(transactionId, table, ci, scan);
+ final long transactionId, final long startId, final TransactionalTable table, final ColumnInterpreter<R, S, P, Q, T> ci, Scan scan) throws Throwable {
+ Pair<S, Long> p = getAvgArgs(transactionId, startId, table, ci, scan);
return ci.divideForAvg(p.getFirst(), p.getSecond());
}
@@ -571,12 +578,12 @@ public class TransactionalAggregationClient {
* @throws Throwable
*/
private <R, S, P extends Message, Q extends Message, T extends Message>
- Pair<List<S>, Long> getStdArgs(final long transactionId, final TransactionalTable table,
+ Pair<List<S>, Long> getStdArgs(final long transactionId, final long startId, final TransactionalTable table,
final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan) throws Throwable {
byte [] currentBeginKey = scan.getStartRow();
HRegionInfo currentRegion = table.getRegionLocation(currentBeginKey).getRegionInfo();
com.google.protobuf.ByteString regionName = ByteString.copyFromUtf8(currentRegion.getRegionNameAsString());
- final TransactionalAggregateRequest requestArg = validateArgAndGetPB(regionName, transactionId, scan, ci, false);
+ final TransactionalAggregateRequest requestArg = validateArgAndGetPB(regionName, transactionId, startId, scan, ci, false);
class StdCallback implements Batch.Callback<Pair<List<S>, Long>> {
long rowCountVal = 0l;
S sumVal = null, sumSqVal = null;
@@ -640,6 +647,7 @@ public class TransactionalAggregationClient {
* columninterpreter says. So, this methods collects the necessary parameters
* to compute the std and returns the double value.
* @param transactionId
+ * @param startId
* @param tableName
* @param ci
* @param scan
@@ -647,12 +655,12 @@ public class TransactionalAggregationClient {
* @throws Throwable
*/
public <R, S, P extends Message, Q extends Message, T extends Message>
- double std( final long transactionId, final TableName tableName, ColumnInterpreter<R, S, P, Q, T> ci,
+ double std( final long transactionId, final long startId, final TableName tableName, ColumnInterpreter<R, S, P, Q, T> ci,
Scan scan) throws Throwable {
TransactionalTable table = null;
try {
table = new TransactionalTable(tableName.getName(), connection);
- return std(transactionId, table, ci, scan);
+ return std(transactionId, startId, table, ci, scan);
} finally {
if (table != null) {
table.close();
@@ -673,8 +681,8 @@ public class TransactionalAggregationClient {
* @throws Throwable
*/
public <R, S, P extends Message, Q extends Message, T extends Message> double std(
- final long transactionId, final TransactionalTable table, ColumnInterpreter<R, S, P, Q, T> ci, Scan scan) throws Throwable {
- Pair<List<S>, Long> p = getStdArgs(transactionId, table, ci, scan);
+ final long transactionId, final long startId, final TransactionalTable table, ColumnInterpreter<R, S, P, Q, T> ci, Scan scan) throws Throwable {
+ Pair<List<S>, Long> p = getStdArgs(transactionId, startId, table, ci, scan);
double res = 0d;
double avg = ci.divideForAvg(p.getFirst().get(0), p.getSecond());
double avgOfSumSq = ci.divideForAvg(p.getFirst().get(1), p.getSecond());
@@ -697,12 +705,12 @@ public class TransactionalAggregationClient {
*/
private <R, S, P extends Message, Q extends Message, T extends Message>
Pair<NavigableMap<byte[], List<S>>, List<S>>
- getMedianArgs(final long transactionId, final TransactionalTable table,
+ getMedianArgs(final long transactionId, final long startId, final TransactionalTable table,
final ColumnInterpreter<R, S, P, Q, T> ci, final Scan scan) throws Throwable {
byte [] currentBeginKey = scan.getStartRow();
HRegionInfo currentRegion = table.getRegionLocation(currentBeginKey).getRegionInfo();
com.google.protobuf.ByteString regionName = ByteString.copyFromUtf8(currentRegion.getRegionNameAsString());
- final TransactionalAggregateRequest requestArg = validateArgAndGetPB(regionName, transactionId, scan, ci, false);
+ final TransactionalAggregateRequest requestArg = validateArgAndGetPB(regionName, transactionId, startId, scan, ci, false);
final NavigableMap<byte[], List<S>> map =
new TreeMap<byte[], List<S>>(Bytes.BYTES_COMPARATOR);
class StdCallback implements Batch.Callback<List<S>> {
@@ -757,6 +765,7 @@ public class TransactionalAggregationClient {
* given cf-cq combination. This method collects the necessary parameters
* to compute the median and returns the median.
* @param transactionId
+ * @param startId
* @param tableName
* @param ci
* @param scan
@@ -764,12 +773,12 @@ public class TransactionalAggregationClient {
* @throws Throwable
*/
public <R, S, P extends Message, Q extends Message, T extends Message>
- R median( final long transactionId, final TableName tableName, ColumnInterpreter<R, S, P, Q, T> ci,
+ R median( final long transactionId, final long startId, final TableName tableName, ColumnInterpreter<R, S, P, Q, T> ci,
Scan scan) throws Throwable {
TransactionalTable table = null;
try {
table = new TransactionalTable(tableName.getName(), connection);
- return median(transactionId, table, ci, scan);
+ return median(transactionId, startId, table, ci, scan);
} finally {
if (table != null) {
table.close();
@@ -788,9 +797,9 @@ public class TransactionalAggregationClient {
* @throws Throwable
*/
public <R, S, P extends Message, Q extends Message, T extends Message>
- R median(final long transactionId, final TransactionalTable table, ColumnInterpreter<R, S, P, Q, T> ci,
+ R median(final long transactionId, final long startId, final TransactionalTable table, ColumnInterpreter<R, S, P, Q, T> ci,
Scan scan) throws Throwable {
- Pair<NavigableMap<byte[], List<S>>, List<S>> p = getMedianArgs(transactionId, table, ci, scan);
+ Pair<NavigableMap<byte[], List<S>>, List<S>> p = getMedianArgs(transactionId, startId, table, ci, scan);
byte[] startRow = null;
byte[] colFamily = scan.getFamilies()[0];
NavigableSet<byte[]> quals = scan.getFamilyMap().get(colFamily);
@@ -859,19 +868,21 @@ public class TransactionalAggregationClient {
}
<R, S, P extends Message, Q extends Message, T extends Message> TransactionalAggregateRequest
- validateArgAndGetPB(com.google.protobuf.ByteString regionName, long transactionId, Scan scan, ColumnInterpreter<R,S,P,Q,T> ci, boolean canFamilyBeAbsent)
+ validateArgAndGetPB(com.google.protobuf.ByteString regionName, long transactionId, final long startId, Scan scan, ColumnInterpreter<R,S,P,Q,T> ci, boolean canFamilyBeAbsent)
throws IOException {
validateParameters(scan, canFamilyBeAbsent);
final TransactionalAggregateRequest.Builder requestBuilder =
TransactionalAggregateRequest.newBuilder();
requestBuilder.setInterpreterClassName(ci.getClass().getCanonicalName());
P columnInterpreterSpecificData = null;
+ TransactionState ts = null;
if ((columnInterpreterSpecificData = ci.getRequestData())
!= null) {
requestBuilder.setInterpreterSpecificBytes(columnInterpreterSpecificData.toByteString());
}
requestBuilder.setScan(ProtobufUtil.toScan(scan));
requestBuilder.setTransactionId(transactionId);
+ requestBuilder.setStartId(startId);
requestBuilder.setRegionName(regionName);
return requestBuilder.build();
}
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/91794b53/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalScanner.java.tmpl
----------------------------------------------------------------------
diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalScanner.java.tmpl b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalScanner.java.tmpl
index af7ed70..35dc84a 100755
--- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalScanner.java.tmpl
+++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalScanner.java.tmpl
@@ -64,7 +64,7 @@ public class TransactionalScanner extends AbstractClientScanner {
public TransactionalTable ttable;
protected boolean closed = false;
// Experiment with this parameter, may be faster without having to send the final close()
- protected boolean doNotCloseOnLast = true;
+ protected boolean regionShouldNotCloseOnLast = true;
protected int nbRows = 100;
protected long nextCallSeq = 0;
private boolean hasMore = true;
@@ -141,21 +141,26 @@ public class TransactionalScanner extends AbstractClientScanner {
protected boolean nextScanner(final boolean done) throws IOException{
if(LOG.isTraceEnabled()) LOG.trace("nextScanner() -- ENTRY txID: " + ts.getTransactionId());
if(this.currentBeginKey != null) {
- if(LOG.isTraceEnabled()) LOG.trace("nextScanner() currentBeginKey != null txID: " + ts.getTransactionId());
- if (doNotCloseOnLast)
+ if(LOG.isTraceEnabled()) LOG.trace("nextScanner() currentBeginKey != null txID: " + ts.getTransactionId()
+ + " currentBeginKey=" + (Bytes.equals(this.currentBeginKey, HConstants.EMPTY_START_ROW) ?
+ "INFINITE" : Hex.encodeHexString(this.currentBeginKey)));
+ if (regionShouldNotCloseOnLast){ // if the region won't automatically we need to tell it to close
+ if(LOG.isTraceEnabled()) LOG.trace("nextScanner() regionShouldNotCloseOnLast is true; closing scanner");
close();
+ }
if((this.currentEndKey == HConstants.EMPTY_END_ROW) ||
Bytes.equals(this.currentEndKey, HConstants.EMPTY_BYTE_ARRAY) ||
checkScanStopRow(this.currentEndKey) ||
done) {
- if(LOG.isTraceEnabled()) LOG.trace("endKey: " + Bytes.toString(this.currentEndKey));
+ if(LOG.isTraceEnabled()) LOG.trace("endKey: " + (Bytes.equals(this.currentEndKey, HConstants.EMPTY_END_ROW) ?
+ "INFINITE" : Hex.encodeHexString(this.currentEndKey)));
if(LOG.isTraceEnabled()) LOG.trace("nextScanner() -- EXIT -- returning false txID: " + ts.getTransactionId());
this.moreScanners = false;
return false;
}
else
- //this.currentBeginKey = TransactionManager.binaryIncrementPos(this.currentEndKey,1);
- this.currentBeginKey = this.currentEndKey;
+ this.currentBeginKey = TransactionManager.binaryIncrementPos(this.currentEndKey,1);
+ //this.currentBeginKey = this.currentEndKey;
}
else {
// First call to nextScanner
@@ -165,15 +170,25 @@ public class TransactionalScanner extends AbstractClientScanner {
this.currentRegion = ttable.getRegionLocation(this.currentBeginKey, false).getRegionInfo();
this.currentEndKey = this.currentRegion.getEndKey();
- if(LOG.isTraceEnabled()) LOG.trace("nextScanner() txID: " + ts.getTransactionId() + " Region Info: " + currentRegion.getRegionNameAsString()+
- "currentBeginKey: " + Hex.encodeHexString(this.currentBeginKey) + "currentEndKey: " + Hex.encodeHexString(this.currentEndKey));
- //if(this.currentEndKey != HConstants.EMPTY_END_ROW)
- // this.currentEndKey = TransactionManager.binaryIncrementPos(currentRegion.getEndKey(), -1);
+ if(LOG.isTraceEnabled()) LOG.trace("nextScanner() txID: " + ts.getTransactionId() + " Region Info: " + currentRegion.getRegionNameAsString()
+ + "currentBeginKey: " + (Bytes.equals(this.currentBeginKey, HConstants.EMPTY_START_ROW) ?
+ "INFINITE" : Hex.encodeHexString(this.currentBeginKey))
+ + " currentEndKey: " + (Bytes.equals(this.currentEndKey, HConstants.EMPTY_END_ROW) ?
+ "INFINITE" : Hex.encodeHexString(this.currentEndKey)));
+ if ( ! (Bytes.equals(this.currentEndKey, HConstants.EMPTY_END_ROW)))
+ this.currentEndKey = TransactionManager.binaryIncrementPos(currentRegion.getEndKey(), -1);
this.closed = false;
+ // Need to add the next region as a participant to the transaction so it gets phase 1 notification
+ // and does not orphan the TransactionState object in the region that gets created during the scan
+ if (LOG.isTraceEnabled()) LOG.trace("nextScanner() calling RMInterface.registerTransaction for startKey "
+ + Hex.encodeHexString(this.currentBeginKey));
+ RMInterface.registerTransaction(ttable, ts, this.currentBeginKey);
+
TrxRegionProtos.OpenScannerRequest.Builder requestBuilder = OpenScannerRequest.newBuilder();
requestBuilder.setTransactionId(ts.getTransactionId());
+ requestBuilder.setStartId(ts.getStartId());
requestBuilder.setRegionName(ByteString.copyFromUtf8(currentRegion.getRegionNameAsString()));
requestBuilder.setScan(ProtobufUtil.toScan(scan));
TrxRegionProtos.OpenScannerRequest openRequest = requestBuilder.build();
@@ -212,10 +227,11 @@ public class TransactionalScanner extends AbstractClientScanner {
final long nextCallSeqInput = this.nextCallSeq;
TrxRegionProtos.PerformScanRequest.Builder requestBuilder = PerformScanRequest.newBuilder();
requestBuilder.setTransactionId(ts.getTransactionId());
+ requestBuilder.setStartId(ts.getStartId());
requestBuilder.setRegionName(ByteString.copyFromUtf8(currentRegion.getRegionNameAsString()));
requestBuilder.setScannerId(scannerID);
requestBuilder.setNumberOfRows(nbRows);
- if (doNotCloseOnLast)
+ if (regionShouldNotCloseOnLast)
requestBuilder.setCloseScanner(false);
else
requestBuilder.setCloseScanner(true);
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/91794b53/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalTable.java
----------------------------------------------------------------------
diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalTable.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalTable.java
index a7d99c0..322af24 100644
--- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalTable.java
+++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/client/transactional/TransactionalTable.java
@@ -38,16 +38,15 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
import org.apache.commons.codec.binary.Hex;
-
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValueUtil;
@@ -58,6 +57,7 @@ import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.TrafParallelClientScanner;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
@@ -107,12 +107,15 @@ import java.util.concurrent.ThreadPoolExecutor;
* Table with transactional support.
*/
public class TransactionalTable extends HTable implements TransactionalTableClient {
- static final Log LOG = LogFactory.getLog(RMInterface.class);
+ static final Log LOG = LogFactory.getLog(TransactionalTable.class);
+ static Configuration config;
static private Connection connection = null;
static ThreadPoolExecutor threadPool = null;
static int retries = 15;
static int delay = 1000;
- private String retryErrMsg = "Coprocessor result is null, retries exhausted";
+ static int regionNotReadyDelay = 30000;
+
+ private String retryErrMsg = "Coprocessor result is null, retries exhausted for table " + this.getName().getNameAsString();
static {
Configuration config = HBaseConfiguration.create();
@@ -176,6 +179,7 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
org.apache.hadoop.hbase.coprocessor.transactional.generated.TrxRegionProtos.GetTransactionalRequest.Builder builder = GetTransactionalRequest.newBuilder();
builder.setGet(ProtobufUtil.toGet(get));
builder.setTransactionId(transactionState.getTransactionId());
+ builder.setStartId(transactionState.getStartId());
builder.setRegionName(ByteString.copyFromUtf8(regionName));
instance.get(controller, builder.build(), rpcCallback);
@@ -198,8 +202,17 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
retry = false;
}
- if(result == null || result.getException().contains("closing region")) {
- Thread.sleep(TransactionalTable.delay);
+ if(result == null || result.getException().contains("closing region")
+ || result.getException().contains("NewTransactionStartedBefore")) {
+ if (result != null && result.getException().contains("NewTransactionStartedBefore")) {
+ if (LOG.isTraceEnabled()) LOG.trace("Get retrying because region is recovering,"
+ + " transaction [" + transactionState.getTransactionId() + "]");
+
+ Thread.sleep(TransactionalTable.regionNotReadyDelay);
+ }
+ else{
+ Thread.sleep(TransactionalTable.delay);
+ }
retry = true;
transactionState.setRetried(true);
retryCount++;
@@ -242,6 +255,7 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
public DeleteTransactionalResponse call(TrxRegionService instance) throws IOException {
org.apache.hadoop.hbase.coprocessor.transactional.generated.TrxRegionProtos.DeleteTransactionalRequest.Builder builder = DeleteTransactionalRequest.newBuilder();
builder.setTransactionId(transactionState.getTransactionId());
+ builder.setStartId(transactionState.getStartId());
builder.setRegionName(ByteString.copyFromUtf8(regionName));
MutationProto m1 = ProtobufUtil.toMutation(MutationType.DELETE, delete);
@@ -268,8 +282,17 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
retry = false;
}
- if(result == null || result.getException().contains("closing region")) {
- Thread.sleep(TransactionalTable.delay);
+ if(result == null || result.getException().contains("closing region")
+ || result.getException().contains("NewTransactionStartedBefore")) {
+ if (result != null && result.getException().contains("NewTransactionStartedBefore")) {
+ if (LOG.isTraceEnabled()) LOG.trace("Delete retrying because region is recovering,"
+ + " transaction [" + transactionState.getTransactionId() + "]");
+
+ Thread.sleep(TransactionalTable.regionNotReadyDelay);
+ }
+ else{
+ Thread.sleep(TransactionalTable.delay);
+ }
retry = true;
transactionState.setRetried(true);
retryCount++;
@@ -286,7 +309,8 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
}
public void deleteRegionTx(final long tid, final Delete delete, final boolean autoCommit) throws IOException {
- if (LOG.isTraceEnabled()) LOG.trace("TransactionalTable.deleteRegionTx ENTRY, autoCommit: " + autoCommit);
+ if (LOG.isTraceEnabled()) LOG.trace("TransactionalTable.deleteRegionTx ENTRY, autoCommit: "
+ + autoCommit);
SingleVersionDeleteNotSupported.validateDelete(delete);
final String regionName = super.getRegionLocation(delete.getRow()).getRegionInfo().getRegionNameAsString();
@@ -300,6 +324,7 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
public DeleteRegionTxResponse call(TrxRegionService instance) throws IOException {
org.apache.hadoop.hbase.coprocessor.transactional.generated.TrxRegionProtos.DeleteRegionTxRequest.Builder builder = DeleteRegionTxRequest.newBuilder();
builder.setTid(tid);
+ builder.setCommitId(-1);
builder.setAutoCommit(autoCommit);
builder.setRegionName(ByteString.copyFromUtf8(regionName));
@@ -358,7 +383,6 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
}
public synchronized void put(final TransactionState transactionState, final Put put, final boolean bool_addLocation) throws IOException {
- validatePut(put);
if (LOG.isTraceEnabled()) LOG.trace("TransactionalTable.put ENTRY");
if (bool_addLocation) addLocation(transactionState, super.getRegionLocation(put.getRow()));
@@ -373,6 +397,7 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
public PutTransactionalResponse call(TrxRegionService instance) throws IOException {
org.apache.hadoop.hbase.coprocessor.transactional.generated.TrxRegionProtos.PutTransactionalRequest.Builder builder = PutTransactionalRequest.newBuilder();
builder.setTransactionId(transactionState.getTransactionId());
+ builder.setStartId(transactionState.getStartId());
builder.setRegionName(ByteString.copyFromUtf8(regionName));
@@ -398,8 +423,17 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
retry = false;
}
- if(result == null || result.getException().contains("closing region")) {
- Thread.sleep(TransactionalTable.delay);
+ if(result == null || result.getException().contains("closing region")
+ || result.getException().contains("NewTransactionStartedBefore")) {
+ if (result != null && result.getException().contains("NewTransactionStartedBefore")) {
+ if (LOG.isTraceEnabled()) LOG.trace("Put retrying because region is recovering,"
+ + " transaction [" + transactionState.getTransactionId() + "]");
+
+ Thread.sleep(TransactionalTable.regionNotReadyDelay);
+ }
+ else{
+ Thread.sleep(TransactionalTable.delay);
+ }
retry = true;
transactionState.setRetried(true);
retryCount++;
@@ -420,9 +454,9 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
}
public synchronized void putRegionTx(final long tid, final Put put, final boolean autoCommit) throws IOException{
- if (LOG.isTraceEnabled()) LOG.trace("TransactionalTable.putRegionTx ENTRY, autoCommit: " + autoCommit);
+ if (LOG.isTraceEnabled()) LOG.trace("TransactionalTable.putRegionTx ENTRY, autoCommit: "
+ + autoCommit);
- validatePut(put);
final String regionName = super.getRegionLocation(put.getRow()).getRegionInfo().getRegionNameAsString();
Batch.Call<TrxRegionService, PutRegionTxResponse> callable =
@@ -434,6 +468,7 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
public PutRegionTxResponse call(TrxRegionService instance) throws IOException {
org.apache.hadoop.hbase.coprocessor.transactional.generated.TrxRegionProtos.PutRegionTxRequest.Builder builder = PutRegionTxRequest.newBuilder();
builder.setTid(tid);
+ builder.setCommitId(-1);
builder.setAutoCommit(autoCommit);
builder.setRegionName(ByteString.copyFromUtf8(regionName));
MutationProto m1 = ProtobufUtil.toMutation(MutationType.PUT, put);
@@ -480,13 +515,21 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
}
public synchronized ResultScanner getScanner(final TransactionState transactionState, final Scan scan) throws IOException {
- if (LOG.isTraceEnabled()) LOG.trace("Enter TransactionalTable.getScanner");
+ if (LOG.isTraceEnabled()) LOG.trace("Enter TransactionalTable.getScanner for transaction " + transactionState.getTransactionId() + " scan ");
if (scan.getCaching() <= 0) {
scan.setCaching(getScannerCaching());
}
Long value = (long) 0;
TransactionalScanner scanner = new TransactionalScanner(this, transactionState, scan, value);
+ if (LOG.isTraceEnabled()){
+ String startRow = (Bytes.equals(scan.getStartRow(), HConstants.EMPTY_START_ROW) ?
+ "INFINITE" : Hex.encodeHexString(scan.getStartRow()));
+ String stopRow = (Bytes.equals(scan.getStopRow(), HConstants.EMPTY_END_ROW) ?
+ "INFINITE" : Hex.encodeHexString(scan.getStopRow()));
+ LOG.trace("Exit TransactionalTable.getScanner for transaction "
+ + transactionState.getTransactionId() + " scan startRow=" + startRow + ", stopRow=" + stopRow);
+ }
return scanner;
}
@@ -495,7 +538,8 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
final byte[] family, final byte[] qualifier, final byte[] value,
final Delete delete, final boolean autoCommit) throws IOException {
if (LOG.isTraceEnabled()) LOG.trace("Enter TransactionalTable.checkAndDeleteRegionTx row: " + row
- + " family: " + family + " qualifier: " + qualifier + " value: " + value);
+ + " family: " + family + " qualifier: " + qualifier + " value: " + value
+ + " autoCommit: " + autoCommit);
if (!Bytes.equals(row, delete.getRow())) {
throw new IOException("checkAndDeleteRegionTx action's getRow must match the passed row");
}
@@ -512,6 +556,7 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
public CheckAndDeleteRegionTxResponse call(TrxRegionService instance) throws IOException {
org.apache.hadoop.hbase.coprocessor.transactional.generated.TrxRegionProtos.CheckAndDeleteRegionTxRequest.Builder builder = CheckAndDeleteRegionTxRequest.newBuilder();
builder.setTid(tid);
+ builder.setCommitId(-1);
builder.setRegionName(ByteString.copyFromUtf8(regionName));
builder.setRow(HBaseZeroCopyByteString.wrap(row));
builder.setAutoCommit(autoCommit);
@@ -589,6 +634,7 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
public CheckAndDeleteResponse call(TrxRegionService instance) throws IOException {
org.apache.hadoop.hbase.coprocessor.transactional.generated.TrxRegionProtos.CheckAndDeleteRequest.Builder builder = CheckAndDeleteRequest.newBuilder();
builder.setTransactionId(transactionState.getTransactionId());
+ builder.setStartId(transactionState.getStartId());
builder.setRegionName(ByteString.copyFromUtf8(regionName));
builder.setRow(HBaseZeroCopyByteString.wrap(row));
if(family != null)
@@ -629,8 +675,17 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
retry = false;
}
- if(result == null || result.getException().contains("closing region")) {
- Thread.sleep(TransactionalTable.delay);
+ if(result == null || result.getException().contains("closing region")
+ || result.getException().contains("NewTransactionStartedBefore")) {
+ if (result != null && result.getException().contains("NewTransactionStartedBefore")) {
+ if (LOG.isTraceEnabled()) LOG.trace("CheckAndDelete retrying because region is recovering, "
+ + " transaction [" + transactionState.getTransactionId() + "]");
+
+ Thread.sleep(TransactionalTable.regionNotReadyDelay);
+ }
+ else{
+ Thread.sleep(TransactionalTable.delay);
+ }
retry = true;
transactionState.setRetried(true);
retryCount++;
@@ -651,8 +706,8 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
final byte[] row, final byte[] family, final byte[] qualifier,
final byte[] value, final Put put) throws IOException {
- if (LOG.isTraceEnabled()) LOG.trace("Enter TransactionalTable.checkAndPut row: " + row
- + " family: " + family + " qualifier: " + qualifier
+ if (LOG.isTraceEnabled()) LOG.trace("Enter TransactionalTable.checkAndPut row: " + Hex.encodeHexString(row)
+ + " put.row " + Hex.encodeHexString(put.getRow()) + " family: " + family + " qualifier: " + qualifier
+ " value: " + value);
if (!Bytes.equals(row, put.getRow())) {
throw new IOException("Action's getRow must match the passed row");
@@ -672,6 +727,8 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
public CheckAndPutResponse call(TrxRegionService instance) throws IOException {
org.apache.hadoop.hbase.coprocessor.transactional.generated.TrxRegionProtos.CheckAndPutRequest.Builder builder = CheckAndPutRequest.newBuilder();
builder.setTransactionId(transactionState.getTransactionId());
+ if (LOG.isTraceEnabled()) LOG.trace("checkAndPut, seting request startid: " + transactionState.getStartId());
+ builder.setStartId(transactionState.getStartId());
builder.setRegionName(ByteString.copyFromUtf8(regionName));
builder.setRow(HBaseZeroCopyByteString.wrap(row));
if (family != null)
@@ -711,8 +768,17 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
retry = false;
}
- if(result == null || result.getException().contains("closing region")) {
- Thread.sleep(TransactionalTable.delay);
+ if(result == null || result.getException().contains("closing region")
+ || result.getException().contains("NewTransactionStartedBefore")) {
+ if (result != null && result.getException().contains("NewTransactionStartedBefore")) {
+ if (LOG.isTraceEnabled()) LOG.trace("CheckAndPut retrying because region is recovering ,"
+ + " transaction [" + transactionState.getTransactionId() + "]");
+
+ Thread.sleep(TransactionalTable.regionNotReadyDelay);
+ }
+ else{
+ Thread.sleep(TransactionalTable.delay);
+ }
retry = true;
transactionState.setRetried(true);
retryCount++;
@@ -775,6 +841,7 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
public CheckAndPutRegionTxResponse call(TrxRegionService instance) throws IOException {
org.apache.hadoop.hbase.coprocessor.transactional.generated.TrxRegionProtos.CheckAndPutRegionTxRequest.Builder builder = CheckAndPutRegionTxRequest.newBuilder();
builder.setTid(tid);
+ builder.setCommitId(-1);
builder.setRegionName(ByteString.copyFromUtf8(regionName));
builder.setRow(HBaseZeroCopyByteString.wrap(row));
if (family != null)
@@ -881,6 +948,7 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
public DeleteMultipleTransactionalResponse call(TrxRegionService instance) throws IOException {
org.apache.hadoop.hbase.coprocessor.transactional.generated.TrxRegionProtos.DeleteMultipleTransactionalRequest.Builder builder = DeleteMultipleTransactionalRequest.newBuilder();
builder.setTransactionId(transactionState.getTransactionId());
+ builder.setStartId(transactionState.getStartId());
builder.setRegionName(ByteString.copyFromUtf8(regionName));
for(Delete delete : rowsInSameRegion) {
@@ -908,8 +976,19 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
retry = false;
}
- if(result == null || result.getException().contains("closing region")) {
- Thread.sleep(TransactionalTable.delay);
+ if(result == null || result.getException().contains("closing region")
+ || result.getException().contains("NewTransactionStartedBefore")) {
+ if (result != null && result.getException().contains("NewTransactionStartedBefore")) {
+ if (LOG.isTraceEnabled()) LOG.trace("delete <List> retrying because region is recovering trRegion ["
+ + location.getRegionInfo().getEncodedName() + "], endKey: "
+ + Hex.encodeHexString(location.getRegionInfo().getEndKey())
+ + " and transaction [" + transactionId + "]");
+
+ Thread.sleep(TransactionalTable.regionNotReadyDelay);
+ }
+ else{
+ Thread.sleep(TransactionalTable.delay);
+ }
retry = true;
transactionState.setRetried(true);
retryCount++;
@@ -946,7 +1025,6 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
List<Put> list = null;
int size = 0;
for (Put put : puts) {
- validatePut(put);
hlocation = this.getRegionLocation(put.getRow(), false);
location = new TransactionRegionLocation(hlocation.getRegionInfo(), hlocation.getServerName());
if (LOG.isTraceEnabled()) LOG.trace("put <List> with trRegion [" + location.getRegionInfo().getEncodedName() + "], endKey: "
@@ -976,6 +1054,7 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
public PutMultipleTransactionalResponse call(TrxRegionService instance) throws IOException {
org.apache.hadoop.hbase.coprocessor.transactional.generated.TrxRegionProtos.PutMultipleTransactionalRequest.Builder builder = PutMultipleTransactionalRequest.newBuilder();
builder.setTransactionId(transactionState.getTransactionId());
+ builder.setStartId(transactionState.getStartId());
builder.setRegionName(ByteString.copyFromUtf8(regionName));
for (Put put : rowsInSameRegion){
@@ -1002,8 +1081,19 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
retry = false;
}
- if(result == null || result.getException().contains("closing region")) {
- Thread.sleep(TransactionalTable.delay);
+ if(result == null || result.getException().contains("closing region")
+ || result.getException().contains("NewTransactionStartedBefore")) {
+ if (result != null && result.getException().contains("NewTransactionStartedBefore")) {
+ if (LOG.isTraceEnabled()) LOG.trace("put <List> retrying because region is recovering trRegion ["
+ + location.getRegionInfo().getEncodedName() + "], endKey: "
+ + Hex.encodeHexString(location.getRegionInfo().getEndKey())
+ + " and transaction [" + transactionId + "]");
+
+ Thread.sleep(TransactionalTable.regionNotReadyDelay);
+ }
+ else{
+ Thread.sleep(TransactionalTable.delay);
+ }
retry = true;
transactionState.setRetried(true);
retryCount++;
@@ -1020,24 +1110,6 @@ public class TransactionalTable extends HTable implements TransactionalTableClie
}
}
- // validate for well-formedness
- public void validatePut(final Put put) throws IllegalArgumentException {
- if (put.isEmpty()) {
- throw new IllegalArgumentException("No columns to insert");
- }
- if (maxKeyValueSize > 0) {
- for (List<Cell> list : put.getFamilyCellMap().values()) {
- for (Cell c : list) {
- if (KeyValueUtil.length(c) > maxKeyValueSize) {
- throw new IllegalArgumentException("KeyValue size too large");
- }
- }
- }
- }
- }
-
- private int maxKeyValueSize;
-
public HRegionLocation getRegionLocation(byte[] row, boolean f)
throws IOException {
return super.getRegionLocation(row,f);