You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@trafodion.apache.org by su...@apache.org on 2016/09/26 23:31:44 UTC
[1/2] incubator-trafodion git commit: Recovery changes for region
transaction changes Fixes are part of JIRA [TRAFODION-1435]
Repository: incubator-trafodion
Updated Branches:
refs/heads/master 3756e10f1 -> 34ddd4cc7
Recovery changes for region transaction changes
Fixes are part of JIRA [TRAFODION-1435]
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/commit/e5f52d85
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/tree/e5f52d85
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/diff/e5f52d85
Branch: refs/heads/master
Commit: e5f52d855b64c87806e828a82f6eff1d8bed7aa5
Parents: 3756e10
Author: Sean Broeder <sb...@edev06.esgyn.local>
Authored: Mon Sep 26 18:27:34 2016 +0000
Committer: Sean Broeder <sb...@edev06.esgyn.local>
Committed: Mon Sep 26 18:27:34 2016 +0000
----------------------------------------------------------------------
.../transactional/TrxRegionEndpoint.java.tmpl | 112 +++++++++++--------
.../transactional/TrxRegionObserver.java.tmpl | 54 ++++++---
.../transactional/TransactionState.java | 14 ++-
.../transactional/TrxTransactionState.java.tmpl | 4 +-
4 files changed, 119 insertions(+), 65 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/e5f52d85/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/coprocessor/transactional/TrxRegionEndpoint.java.tmpl
----------------------------------------------------------------------
diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/coprocessor/transactional/TrxRegionEndpoint.java.tmpl b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/coprocessor/transactional/TrxRegionEndpoint.java.tmpl
index 75ef76c..ed74336 100644
--- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/coprocessor/transactional/TrxRegionEndpoint.java.tmpl
+++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/coprocessor/transactional/TrxRegionEndpoint.java.tmpl
@@ -427,6 +427,9 @@ CoprocessorService, Coprocessor {
public static final int TS_COMMIT = 2;
public static final int TS_ABORT = 3;
public static final int TS_CONTROL_POINT_COMMIT = 4;
+ public static final int TS_REGION_TX_ACTIVE = 5;
+ public static final int TS_REGION_TX_COMMIT_REQUEST = 6;
+ public static final int TS_REGION_TX_COMMIT = 7;
public static final int REGION_STATE_RECOVERING = 0;
public static final int REGION_STATE_START = 2;
@@ -909,8 +912,8 @@ CoprocessorService, Coprocessor {
int numOfRegion = request.getRegionNameCount();
String requestRegionName;
- if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor: commitRequestMultiple - txId " + transactionId + ", master regionName " + regionInfo.getRegionNameAsString());
- if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor: commitRequestMultiple - txId " + transactionId + " number of region is commitMultiple " + numOfRegion);
+ if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor: commitRequestMultiple - txId " + transactionId
+ + " number of region is commitMultiple " + numOfRegion + ", master regionName " + regionInfo.getRegionNameAsString());
org.apache.hadoop.hbase.coprocessor.transactional.generated.TrxRegionProtos.CommitRequestMultipleResponse.Builder commitRequestMultipleResponseBuilder = CommitRequestMultipleResponse.newBuilder();
commitRequestMultipleResponseBuilder.setHasException(false);
@@ -4515,13 +4518,18 @@ CoprocessorService, Coprocessor {
List<Tag> tagList = new ArrayList<Tag>();
if (state.hasWrite() || state.isReinstated()) {
if (!state.getFullEditInCommit()) {
- commitTag = state.formTransactionalContextTag(TS_COMMIT);
+ if (state.getIsRegionTx()){
+ commitTag = state.formTransactionalContextTag(TS_REGION_TX_COMMIT);
+ }
+ else {
+ commitTag = state.formTransactionalContextTag(TS_COMMIT);
+ }
tagList.add(commitTag);
WALEdit e1 = state.getEdit();
WALEdit e = new WALEdit();
if (e1.isEmpty() || e1.getCells().size() <= 0) {
if (LOG.isInfoEnabled()) LOG.info("TRAF RCOV endpoint CP: commit - txId "
- + transactionId
+ + transactionId + " regionTx: " + state.getIsRegionTx()
+ ", Encountered empty TS WAL Edit list during commit, HLog txid "
+ txid);
}
@@ -4542,11 +4550,12 @@ CoprocessorService, Coprocessor {
AtomicLong lv_seqid = this.m_Region.getSequenceId();
txid = this.tHLog.append(this.m_Region.getTableDesc(), this.regionInfo, wk, e, lv_seqid, false, null);
#endif
- if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor: commit - txId " + transactionId + ", Write commit HLOG seq " + txid);
+ if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor: commit - txId " + transactionId
+ + " regionTx: " + state.getIsRegionTx() + ", Write commit HLOG seq " + txid);
}
catch (IOException exp1) {
if (LOG.isTraceEnabled()) LOG.trace("TRAF RCOV endpoint CP: commit -"
- + " txId " + transactionId
+ + " txId " + transactionId + " regionTx: " + state.getIsRegionTx()
+ ", Writing to HLOG : Threw an exception " + exp1.toString()
);
throw exp1;
@@ -4554,7 +4563,8 @@ CoprocessorService, Coprocessor {
} // e1 is not empty
} // not full edit write in commit record during phase 2
else { //do this for rollover case
- if (LOG.isTraceEnabled()) LOG.trace("TRAF RCOV endpoint CP:commit -- HLOG rollover txId: " + transactionId);
+ if (LOG.isTraceEnabled()) LOG.trace("TRAF RCOV endpoint CP:commit -- HLOG rollover txId: "
+ + transactionId + " regionTx: " + state.getIsRegionTx());
commitTag = state.formTransactionalContextTag(TS_CONTROL_POINT_COMMIT);
tagList.add(commitTag);
WALEdit e1 = state.getEdit();
@@ -4577,17 +4587,18 @@ CoprocessorService, Coprocessor {
txid = this.tHLog.append(this.m_Region.getTableDesc(), this.regionInfo, wk, e, lv_seqid, false, null);
#endif
this.tHLog.sync(txid);
- if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor: commit - txId " + transactionId + ", Y11 write commit HLOG seq " + txid);
+ if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor: commit - txId " + transactionId
+ + " regionTx: " + state.getIsRegionTx() + ", Y11 write commit HLOG seq " + txid);
}
catch (IOException exp1) {
if (LOG.isTraceEnabled()) LOG.trace("TRAF RCOV endpoint CP: commit -"
- + " txId " + transactionId
+ + " txId " + transactionId + " regionTx: " + state.getIsRegionTx()
+ ", Writing to HLOG : Threw an exception " + exp1.toString()
);
throw exp1;
}
- if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor:commit -- EXIT"
- + " txId: " + transactionId
+ if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor:commit -- EXIT"
+ + " txId " + transactionId + " regionTx: " + state.getIsRegionTx()
+ " HLog seq " + txid
);
// if (this.fullEditInCommit) this.fullEditInCommit = false;
@@ -4599,7 +4610,7 @@ CoprocessorService, Coprocessor {
synchronized (commitPendingTransactions) {
if (!commitPendingTransactions.remove(state)) {
LOG.fatal("TrxRegionEndpoint coprocessor: commit -"
- + " txid: " + transactionId
+ + " txId " + transactionId + " regionTx: " + state.getIsRegionTx()
+ ", Commiting a non-query transaction that is not in commitPendingTransactions"
);
// synchronized statements are cleared for a throw
@@ -4631,23 +4642,23 @@ CoprocessorService, Coprocessor {
);
deleteRecoveryzNode(tmid, lv_encoded);
} catch (IOException e) {
- LOG.error("TrxRegionEndpoint coprocessor: commit -"
- + " txId " + transactionId
+ LOG.error("TrxRegionEndpoint coprocessor: commit -"
+ + " txId " + transactionId
+ ", Trafodion Recovery: delete recovery zNode failed. Caught exception " + e.toString()
);
}
}
if ((indoubtTransactionsById == null) || (indoubtTransactionsById.size() == 0)) {
- if (indoubtTransactionsById == null)
- if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor: commit -"
- + " txId " + transactionId
+ if (indoubtTransactionsById == null)
+ if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor: commit -"
+ + " txId " + transactionId
+ ", Trafodion Recovery: start region in commit with indoubtTransactionsById null"
);
else
- if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor: commit -"
- + " txId " + transactionId
- + ", Trafodion Recovery: start region in commit with indoubtTransactionsById size "
+ if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor: commit -"
+ + " txId " + transactionId
+ + ", Trafodion Recovery: start region in commit with indoubtTransactionsById size "
+ indoubtTransactionsById.size()
);
startRegionAfterRecovery();
@@ -5740,7 +5751,7 @@ CoprocessorService, Coprocessor {
TrxTransactionState state;
if (startEpoch < onlineEpoch) {
- LOG.info("commitRequest txId: "
+ LOG.error("commitRequest txId: "
+ transactionId + " startEpoch " + startEpoch + " is less than region's onlineEpoch " + onlineEpoch
+ " for regionName " + lv_regionName
+ " must return COMMIT_UNSUCCESSFUL_FROM_COPROCESSOR ");
@@ -5767,20 +5778,20 @@ CoprocessorService, Coprocessor {
try {
state = getTransactionState(transactionId);
} catch (UnknownTransactionException e) {
- if (LOG.isDebugEnabled()) LOG.debug("TrxRegionEndpoint coprocessor: commitRequest Unknown transaction ["
+ if (LOG.isWarnEnabled()) LOG.warn("TrxRegionEndpoint coprocessor: commitRequest Unknown transaction ["
+ transactionId + "] in region ["
+ lv_regionName
+ "], participantNum " + participantNum + " ignoring");
- state = null;
+ state = null;
}
- // may change to indicate a NOTFOUND case then depends on the TM ts state, if reinstated tx, ignore the exception
- if (state == null) {
+ // may change to indicate a NOTFOUND case then depends on the TM ts state, if reinstated tx, ignore the exception
+ if (state == null) {
String errMsg = "TrxRegionEndpoint coprocessor: commitRequest for participant " + participantNum
+ " encountered unknown transactionID txId: " + transactionId + " in region " + lv_regionName;
- if (LOG.isTraceEnabled()) LOG.trace(errMsg);
+ if (LOG.isWarnEnabled()) LOG.warn(errMsg);
//return COMMIT_UNSUCCESSFUL_FROM_COPROCESSOR;
throw new UnknownTransactionException(errMsg);
- }
+ }
if (LOG.isInfoEnabled())
hasConflictStartTime = System.nanoTime();
@@ -5804,9 +5815,8 @@ CoprocessorService, Coprocessor {
// No conflicts, we can commit.
if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor: No conflicts for transaction " + transactionId
- + " found in region "
- + lv_regionName
- + ". Participant " + participantNum + " votes to commit");
+ + " regionTx: " + state.getIsRegionTx() + " found in region " + lv_regionName
+ + ". Participant " + participantNum + " votes to commit");
// If there are writes we must keep record of the transaction
putBySequenceStartTime = System.nanoTime();
@@ -5834,29 +5844,35 @@ CoprocessorService, Coprocessor {
//call HLog by passing tagged WALEdits and associated fields
try {
- // Once we append edit into HLOG during DML operation, there is no need to do any HLOG write in phase 1.
- // Likely all the edits have been synced, so just do a sync(state.getLargestFlushTxId()) --> most likely, it is an no-op
- // This is to leverage the time between last DML and phase 1, so there is no need to do any logging (and waited) in phase 1.
- // And there is no need to write "prepared" record since we try to optimize normal running mode (99.99% commit)
- // We can append "commit" or "abort" into HLOG in phase 2 in a no-sync mode, but it can also be eliminated if necessary.
- // All the transaction reinstated during recovery will be treated as "in-doubt" and need to use TLOG to resolve.
- // So TLOG must be kept for a while (likely a few CP), besides, HRegion chore thread can perform ~ CP action by writing
- // the smallest sequenceId (not logSeqid), so we don't need to deal with commite case (but we did write abort edit since
- // the number of abort is < 0.1% -- we can set this as a configurable property).
+ // Once we append edit into HLOG during DML operation, there is no need to do any HLOG write in phase 1.
+ // Likely all the edits have been synced, so just do a sync(state.getLargestFlushTxId()) --> most likely,
+ // it is a no-op. This is to leverage the time between last DML and phase 1, so there is no need to do any
+ // logging (and waited) in phase 1. And there is no need to write "prepared" record since we try to optimize
+ // normal running mode (99.99% commit). We can append "commit" or "abort" into HLOG in phase 2 in a no-sync
+ // mode, but it can also be eliminated if necessary. All the transaction reinstated during recovery will be
+ // treated as "in-doubt" and need to use TLOG to resolve. So TLOG must be kept for a while (likely a few CP),
+ // besides, HRegion chore thread can perform ~ CP action by writing the smallest sequenceId (not logSeqid),
+ // so we don't need to deal with commit case (but we did write abort edit since the number of abort is
+ // < 0.1% -- we can set this as a configurable property).
if (!state.getEarlyLogging()) {
- final WALKey wk = new WALKey(this.regionInfo.getEncodedNameAsBytes(), this.regionInfo.getTable(), EnvironmentEdgeManager.currentTime());
+ final WALKey wk = new WALKey(this.regionInfo.getEncodedNameAsBytes(), this.regionInfo.getTable(),
+ EnvironmentEdgeManager.currentTime());
#ifdef CDH5.7 APACHE1.2
txid = this.tHLog.append(this.m_Region.getTableDesc(), this.regionInfo, wk, state.getEdit(), false);
if ((m_isTrafodionMetadata) || (LOG.isTraceEnabled()))
- LOG.info("TrxRegionEndpoint coprocessor: PH1 commitRequest COMMIT_OK -- EXIT " + m_regionName + " isTrafodionMD: " + m_isTrafodionMetadata
- + " txId: " + transactionId + " HLog seq: " + txid + " flushHLOG: " + flushHLOG);
+ LOG.info("TrxRegionEndpoint coprocessor: PH1 commitRequest COMMIT_OK -- EXIT " + m_regionName
+ + " isTrafodionMD: " + m_isTrafodionMetadata + " txId: " + transactionId + " HLog seq: " + txid
+ + " regionTx: " + state.getIsRegionTx() + " flushHLOG: " + flushHLOG);
#else
AtomicLong lv_seqid = this.m_Region.getSequenceId();
- txid = this.tHLog.append(this.m_Region.getTableDesc(), this.regionInfo, wk, state.getEdit(), lv_seqid, false, null);
+ txid = this.tHLog.append(this.m_Region.getTableDesc(), this.regionInfo, wk, state.getEdit(),
+ lv_seqid, false, null);
if ((m_isTrafodionMetadata) || (LOG.isTraceEnabled()))
- LOG.info("TrxRegionEndpoint coprocessor: PH1 commitRequest COMMIT_OK -- EXIT " + m_regionName + " isTrafodionMD: " + m_isTrafodionMetadata
- + " txId: " + transactionId + " seqId: " + lv_seqid + " HLog seq: " + txid + " flushHLOG: " + flushHLOG);
+ LOG.info("TrxRegionEndpoint coprocessor: PH1 commitRequest COMMIT_OK -- EXIT " + m_regionName
+ + " isTrafodionMD: " + m_isTrafodionMetadata + " txId: " + transactionId + " seqId: "
+ + lv_seqid + " HLog seq: " + txid + " regionTx: " + state.getIsRegionTx()
+ + " flushHLOG: " + flushHLOG);
#endif
if (flushHLOG) this.tHLog.sync(txid);
}
@@ -5869,12 +5885,14 @@ CoprocessorService, Coprocessor {
writeToLogTimes[lv_timeIndex] = writeToLogEndTime - commitCheckEndTime;
writeToLogOperations.getAndIncrement();
}
- //if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor:commitRequest COMMIT_OK -- EXIT txId: " + transactionId + " HLog seq " + txid);
+ //if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor:commitRequest COMMIT_OK -- EXIT txId: "
+ // + transactionId + " regionTx: " + state.getIsRegionTx() + " HLog seq " + txid);
} catch (IOException exp) {
if (LOG.isTraceEnabled()) LOG.trace("TrxRegionEndpoint coprocessor: commitRequest - Caught IOException in HLOG appendNoSync -- EXIT txId: " + transactionId + " HLog seq " + txid);
throw exp;
}
- if (LOG.isDebugEnabled()) LOG.debug("TrxRegionEndpoint coprocessor: commitRequest COMMIT_OK -- EXIT txId: " + transactionId);
+ if (LOG.isDebugEnabled()) LOG.debug("TrxRegionEndpoint coprocessor: commitRequest COMMIT_OK -- EXIT txId: "
+ + transactionId + " regionTx " + state.getIsRegionTx());
returnPending = true;
}
// No write pending
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/e5f52d85/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/coprocessor/transactional/TrxRegionObserver.java.tmpl
----------------------------------------------------------------------
diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/coprocessor/transactional/TrxRegionObserver.java.tmpl b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/coprocessor/transactional/TrxRegionObserver.java.tmpl
index a63df71..a29b9d9 100644
--- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/coprocessor/transactional/TrxRegionObserver.java.tmpl
+++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/coprocessor/transactional/TrxRegionObserver.java.tmpl
@@ -119,6 +119,9 @@ public static final int TS_COMMIT_REQUEST = 1;
public static final int TS_COMMIT = 2;
public static final int TS_ABORT = 3;
public static final int TS_CONTROL_POINT_COMMIT = 4;
+public static final int TS_REGION_TX_ACTIVE = 5;
+public static final int TS_REGION_TX_COMMIT_REQUEST = 6;
+public static final int TS_REGION_TX_COMMIT = 7;
public static final byte TS_TRAFODION_TXN_TAG_TYPE = 41;
public static final int ACTIVETXN_DELAY_DEFAULT = 15 * 1000;
public static final int PENDINGTXN_DELAY_DEFAULT = 500;
@@ -165,6 +168,8 @@ long activeCount = 0;
long abortCount = 0;
long commitCount = 0;
long commitRequestCount = 0;
+long regionTxCommitCount = 0;
+long regionTxCommitRequestCount = 0;
long totalEdits = 0;
long skippedEdits = 0;
int cleanAT = 0;
@@ -375,6 +380,7 @@ static ConcurrentHashMap<String, Object> getRefMap() {
int op = Bytes.toInt(b,Bytes.SIZEOF_INT+offset);
long tid = Bytes.toLong(b,Bytes.SIZEOF_INT+Bytes.SIZEOF_INT+offset);
long logSeqId = Bytes.toLong(b,Bytes.SIZEOF_INT+Bytes.SIZEOF_INT+Bytes.SIZEOF_LONG+offset);
+ ArrayList <WALEdit> editList;
if ((m_isTrafodionMetadata) || LOG.isTraceEnabled()) LOG.info("Trafodion Recovery Region Observer CP:PWR01 Find transactional tag within Edits for " + lv_regionName + " tid " + tid + " op " + op + " log seq " + logSeqId + " version " + version);
//Trafodion Recovery : process each edit according to its nature (prepare, commit, abort)
@@ -386,14 +392,14 @@ static ConcurrentHashMap<String, Object> getRefMap() {
if ((m_isTrafodionMetadata) || LOG.isTraceEnabled()) LOG.info("Trafodion Recovery Region Observer CP: " + regionInfo.getRegionNameAsString() + " processing active edit for transaction: " + tid
+ ", already find a previous edit with that id");
//get the editList from pendingTransactionsById, and add the logEdit into the editList
- ArrayList <WALEdit> editList = (ArrayList<WALEdit>) (pendingTransactionsById.get(tid));
+ editList = (ArrayList<WALEdit>) (pendingTransactionsById.get(tid));
editList.add(logEdit);
activeCount++;
// no need to do the "put" back to the map since it gets the reference already
}
else {
// create the editList
- List<WALEdit> editList = new ArrayList<WALEdit>();
+ editList = new ArrayList<WALEdit>();
editList.add(logEdit);
pendingTransactionsById.put(tid, editList);
activeCount++;
@@ -407,13 +413,13 @@ static ConcurrentHashMap<String, Object> getRefMap() {
+ ", already find a previous edit with that id");
//throw new IOException("Corrupted transaction log");
// get the editList from pendingTransactionsById, and add the logEdit into the editList
- ArrayList<WALEdit> editList = (ArrayList<WALEdit>) (pendingTransactionsById.get(tid));
+ editList = (ArrayList<WALEdit>) (pendingTransactionsById.get(tid));
editList.add(logEdit);
// no need to do the "put" back to the map since it gets the reference already
}
else {
// create the editList
- List<WALEdit> editList = new ArrayList<WALEdit>();
+ editList = new ArrayList<WALEdit>();
editList.add(logEdit);
pendingTransactionsById.put(tid, editList);
//pendingTransactionsById.put(tid, logEdit);
@@ -424,7 +430,8 @@ static ConcurrentHashMap<String, Object> getRefMap() {
case TS_ABORT:
if (!pendingTransactionsById.containsKey(tid)) {
- LOG.info("TRAF RCOV Region Observer CP: " + regionInfo.getRegionNameAsString() + " processing abort for transaction: " + tid
+ LOG.info("TRAF RCOV Region Observer CP: " + regionInfo.getRegionNameAsString()
+ + " processing abort for transaction: " + tid
+ ", but don't find a previous edit with that id");
abortCount++;
break;
@@ -436,21 +443,24 @@ static ConcurrentHashMap<String, Object> getRefMap() {
case TS_COMMIT:
if (!pendingTransactionsById.containsKey(tid)) {
- LOG.info("TRAF RCOV Region Observer CP: " + regionInfo.getRegionNameAsString() + " processing commit for transaction: " + tid
+ LOG.warn("TRAF RCOV Region Observer CP: " + regionInfo.getRegionNameAsString()
+ + " processing commit for transaction: " + tid
+ ", but don't find a previous edit with that id");
commitCount++;
break;
//throw new IOException("Corrupted transaction log");
}
- ArrayList<WALEdit> editList = (ArrayList<WALEdit>) (pendingTransactionsById.get(tid));
- if ((m_isTrafodionMetadata) || (LOG.isTraceEnabled())) LOG.info("TRAF RCOV Region Observer CP: " + regionInfo.getRegionNameAsString() + " find in-doubt transaction " + tid + " to commit");
+ editList = (ArrayList<WALEdit>) (pendingTransactionsById.get(tid));
+ if ((m_isTrafodionMetadata) || (LOG.isTraceEnabled())) LOG.info("TRAF RCOV Region Observer CP: "
+ + regionInfo.getRegionNameAsString() + " find in-doubt transaction " + tid + " to commit");
replayCommittedTransaction(tid, editList);
pendingTransactionsById.remove(tid);
commitCount++;
break;
case TS_CONTROL_POINT_COMMIT:
- if ((m_isTrafodionMetadata) || LOG.isTraceEnabled()) LOG.info("AAA3 get CP edit : " + tid + " region " + regionInfo.getRegionNameAsString());
+ if ((m_isTrafodionMetadata) || LOG.isTraceEnabled()) LOG.info("AAA3 get CP edit : " + tid
+ + " region " + regionInfo.getRegionNameAsString());
ArrayList<WALEdit> tempEditList = new ArrayList<WALEdit>();
tempEditList.add(logEdit);
replayCommittedTransaction(tid, tempEditList);
@@ -458,16 +468,34 @@ static ConcurrentHashMap<String, Object> getRefMap() {
commitCount++;
break;
+ case TS_REGION_TX_COMMIT:
+ if ((m_isTrafodionMetadata) || LOG.isTraceEnabled()) LOG.info("TRAF RCOV Region Observer CP: Region Transaction commit " + tid
+ + " region " + regionInfo.getRegionNameAsString());
+ regionTxCommitCount++;
+ break;
+
+ case TS_REGION_TX_COMMIT_REQUEST: // Treat as commit
+ if ((m_isTrafodionMetadata) || LOG.isTraceEnabled()) LOG.info("TRAF RCOV Region Observer CP: Region Transaction commitRequest " + tid
+ + " region " + regionInfo.getRegionNameAsString());
+ regionTxCommitRequestCount++;
+ // create the editList
+ editList = new ArrayList<WALEdit>();
+ editList.add(logEdit);
+ replayCommittedTransaction(tid, editList);
+ if ((m_isTrafodionMetadata) || LOG.isTraceEnabled()) LOG.info("Trafodion Recovery Region Observer CP: " + regionInfo.getRegionNameAsString() + " find in-doubt REGION transaction " + tid + " in prepared state");
+ break;
+
default:
- throw new IllegalStateException("Trafodion Recovery Region Observer CP: Unexpected log entry type detected in audit replay");
+ throw new IllegalStateException("Trafodion Recovery Region Observer CP: Unexpected log entry type detected in audit replay");
} // switch op
totalEdits++;
- if (LOG.isDebugEnabled()) LOG.debug("TRAF RCOV PREWAL Region Observer CP: " + regionInfo.getRegionNameAsString() + " Edits replay read " + totalEdits + " transactional operations (skipped " + skippedEdits
+ if (LOG.isDebugEnabled()) LOG.debug("TRAF RCOV PREWAL Region Observer CP: " + regionInfo.getRegionNameAsString()
+ + " Edits replay read " + totalEdits + " transactional operations (skipped " + skippedEdits
+ " because sequence id <= " + minSeqID + "): " + commitRequestCount + " commitRequests, "
- + abortCount + " aborts, " + commitCount + " commits, and " + flushCount + " flushes.");
-
+ + regionTxCommitRequestCount + " regionTxCommitRequests, " + abortCount + " aborts, " + commitCount
+ + " commits, " + regionTxCommitCount + " regionTxCommits, and " + flushCount + " flushes.");
//recovery patch
env.complete(); // do not need to invoke further down coprocessor
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/e5f52d85/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/regionserver/transactional/TransactionState.java
----------------------------------------------------------------------
diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/regionserver/transactional/TransactionState.java b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/regionserver/transactional/TransactionState.java
index dc076a1..a7eae1f 100644
--- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/regionserver/transactional/TransactionState.java
+++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/regionserver/transactional/TransactionState.java
@@ -118,7 +118,10 @@ public class TransactionState {
public static final int TS_ACTIVE = 0;
public static final int TS_COMMIT_REQUEST = 1;
- public static byte TS_TRAFODION_TXN_TAG_TYPE = 41;
+ public static final int TS_REGION_TX_ACTIVE = 5;
+ public static final int TS_REGION_TX_COMMIT_REQUEST = 6;
+ public static final int TS_REGION_TX_COMMIT = 7;
+ public static final byte TS_TRAFODION_TXN_TAG_TYPE = 41;
public TransactionState(final long transactionId, final long rLogStartSequenceId, AtomicLong hlogSeqId, final HRegionInfo regionInfo,
HTableDescriptor htd, WAL hLog, boolean logging, boolean isRegionTx) {
@@ -133,7 +136,10 @@ public class TransactionState {
this.tabledescriptor = htd;
this.earlyLogging = logging;
this.tHLog = hLog;
- if (earlyLogging) {
+ if(isRegionTx){ // RegionTX takes precedence
+ transactionalTag = this.formTransactionalContextTag(TS_REGION_TX_COMMIT_REQUEST);
+ }
+ else if (earlyLogging) {
transactionalTag = this.formTransactionalContextTag(TS_ACTIVE);
}
else {
@@ -374,8 +380,8 @@ public class TransactionState {
@Override
public String toString() {
- return "transactionId: " + transactionId + ", status: " + status +
- ", regionInfo: " + regionInfo;
+ return "transactionId: " + transactionId + ", regionTX: " + getIsRegionTx()
+ + ", status: " + status + ", regionInfo: " + regionInfo;
}
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/e5f52d85/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/regionserver/transactional/TrxTransactionState.java.tmpl
----------------------------------------------------------------------
diff --git a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/regionserver/transactional/TrxTransactionState.java.tmpl b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/regionserver/transactional/TrxTransactionState.java.tmpl
index 6353250..42d5a2d 100644
--- a/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/regionserver/transactional/TrxTransactionState.java.tmpl
+++ b/core/sqf/src/seatrans/hbase-trx/src/main/java/org/apache/hadoop/hbase/regionserver/transactional/TrxTransactionState.java.tmpl
@@ -208,7 +208,7 @@ public class TrxTransactionState extends TransactionState {
ListIterator<WriteAction> writeOrderIter = writeOrdering.listIterator(writeOrdering.size());
writeOrderIter.add(waction = new WriteAction(write));
- if (this.earlyLogging) { // immeditaely write edit out to HLOG during DML (actve transaction state)
+ if (this.earlyLogging) { // immediately write edit out to HLOG during DML (active transaction state)
for (Cell value : waction.getCells()) {
// KeyValue kv = KeyValueUtil.ensureKeyValue(value);
kv = KeyValue.cloneAndAddTags(value, tagList);
@@ -517,6 +517,8 @@ public class TrxTransactionState extends TransactionState {
StringBuilder result = new StringBuilder();
result.append("[transactionId: ");
result.append(transactionId);
+ result.append(" regionTX: ");
+ result.append(getIsRegionTx());
result.append(" status: ");
result.append(status.name());
result.append(" scan Size: ");
[2/2] incubator-trafodion git commit: Merge [TRAFODION-1435] PR-728
Recovery changes for region transaction changes
Posted by su...@apache.org.
Merge [TRAFODION-1435] PR-728 Recovery changes for region transaction changes
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/commit/34ddd4cc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/tree/34ddd4cc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/diff/34ddd4cc
Branch: refs/heads/master
Commit: 34ddd4cc7a83b001043f7a4310f92e272a38deab
Parents: 3756e10 e5f52d8
Author: Suresh Subbiah <su...@apache.org>
Authored: Mon Sep 26 23:31:09 2016 +0000
Committer: Suresh Subbiah <su...@apache.org>
Committed: Mon Sep 26 23:31:09 2016 +0000
----------------------------------------------------------------------
.../transactional/TrxRegionEndpoint.java.tmpl | 112 +++++++++++--------
.../transactional/TrxRegionObserver.java.tmpl | 54 ++++++---
.../transactional/TransactionState.java | 14 ++-
.../transactional/TrxTransactionState.java.tmpl | 4 +-
4 files changed, 119 insertions(+), 65 deletions(-)
----------------------------------------------------------------------