You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluo.apache.org by kt...@apache.org on 2016/11/15 16:53:40 UTC
[1/2] incubator-fluo git commit: fixes #785 Added meaningful message
to commit exception
Repository: incubator-fluo
Updated Branches:
refs/heads/master efee307c5 -> 25839df16
fixes #785 Added meaningful message to commit exception
Project: http://git-wip-us.apache.org/repos/asf/incubator-fluo/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-fluo/commit/2e446e6c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-fluo/tree/2e446e6c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-fluo/diff/2e446e6c
Branch: refs/heads/master
Commit: 2e446e6c10731e3446bb8c4e84d157ea265a2730
Parents: efee307
Author: Keith Turner <ke...@deenlo.com>
Authored: Mon Nov 14 21:15:32 2016 -0500
Committer: Keith Turner <ke...@deenlo.com>
Committed: Mon Nov 14 22:10:24 2016 -0500
----------------------------------------------------------------------
.../fluo/core/async/AsyncCommitObserver.java | 2 +-
.../apache/fluo/core/async/CommitManager.java | 4 +-
.../fluo/core/async/SyncCommitObserver.java | 6 ++-
.../core/client/LoaderExecutorAsyncImpl.java | 2 +-
.../apache/fluo/core/impl/TransactionImpl.java | 39 ++++++++++++++++++--
.../fluo/core/log/TracingTransaction.java | 4 +-
.../apache/fluo/core/worker/WorkTaskAsync.java | 2 +-
7 files changed, 47 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/2e446e6c/modules/core/src/main/java/org/apache/fluo/core/async/AsyncCommitObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/async/AsyncCommitObserver.java b/modules/core/src/main/java/org/apache/fluo/core/async/AsyncCommitObserver.java
index 6a1e2ed..4e78f60 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/async/AsyncCommitObserver.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/async/AsyncCommitObserver.java
@@ -41,6 +41,6 @@ public interface AsyncCommitObserver {
/**
* Called when async commit of a transaction fails because it overlapped with another transaction
*/
- void commitFailed();
+ void commitFailed(String msg);
}
http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/2e446e6c/modules/core/src/main/java/org/apache/fluo/core/async/CommitManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/async/CommitManager.java b/modules/core/src/main/java/org/apache/fluo/core/async/CommitManager.java
index 67f465c..403fb66 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/async/CommitManager.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/async/CommitManager.java
@@ -112,9 +112,9 @@ public class CommitManager {
}
@Override
- public void commitFailed() {
+ public void commitFailed(String msg) {
try {
- aco.commitFailed();
+ aco.commitFailed(msg);
} finally {
finish(TxResult.COMMIT_EXCEPTION);
}
http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/2e446e6c/modules/core/src/main/java/org/apache/fluo/core/async/SyncCommitObserver.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/async/SyncCommitObserver.java b/modules/core/src/main/java/org/apache/fluo/core/async/SyncCommitObserver.java
index 2053c99..314859e 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/async/SyncCommitObserver.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/async/SyncCommitObserver.java
@@ -27,6 +27,7 @@ public class SyncCommitObserver implements AsyncCommitObserver {
private volatile boolean committed = false;
private volatile boolean aacked = false;
private volatile Exception error = null;
+ private volatile String commitFailMsg = "";
@Override
public void committed() {
@@ -47,8 +48,9 @@ public class SyncCommitObserver implements AsyncCommitObserver {
}
@Override
- public void commitFailed() {
+ public void commitFailed(String msg) {
committed = false;
+ commitFailMsg = msg;
cdl.countDown();
}
@@ -65,7 +67,7 @@ public class SyncCommitObserver implements AsyncCommitObserver {
} else if (aacked) {
throw new AlreadyAcknowledgedException();
} else if (!committed) {
- throw new CommitException();
+ throw new CommitException(commitFailMsg);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/2e446e6c/modules/core/src/main/java/org/apache/fluo/core/client/LoaderExecutorAsyncImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/client/LoaderExecutorAsyncImpl.java b/modules/core/src/main/java/org/apache/fluo/core/client/LoaderExecutorAsyncImpl.java
index be375c6..ad36fd2 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/client/LoaderExecutorAsyncImpl.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/client/LoaderExecutorAsyncImpl.java
@@ -100,7 +100,7 @@ public class LoaderExecutorAsyncImpl implements LoaderExecutor {
}
@Override
- public void commitFailed() {
+ public void commitFailed(String msg) {
txi = null;
// retry transaction
executor.submit(this);
http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/2e446e6c/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java b/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java
index 441517a..e2e6d32 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java
@@ -77,6 +77,7 @@ import org.apache.fluo.core.util.ColumnUtil;
import org.apache.fluo.core.util.ConditionalFlutation;
import org.apache.fluo.core.util.FluoCondition;
import org.apache.fluo.core.util.Flutation;
+import org.apache.fluo.core.util.Hex;
import org.apache.fluo.core.util.SpanUtil;
/**
@@ -408,6 +409,38 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
return prow + " " + pcol + " " + pval + " " + rejected.size();
}
+ public String getShortCollisionMessage() {
+ StringBuilder sb = new StringBuilder();
+ if (rejected.size() > 0) {
+ int numCollisions = 0;
+ for (Set<Column> cols : rejected.values()) {
+ numCollisions += cols.size();
+ }
+
+ sb.append("Collsions(");
+ sb.append(numCollisions);
+ sb.append("):");
+
+ String sep = "";
+ outer: for (Entry<Bytes, Set<Column>> entry : rejected.entrySet()) {
+ Bytes row = entry.getKey();
+ for (Column col : entry.getValue()) {
+ sb.append(sep);
+ sep = ", ";
+ Hex.encNonAscii(sb, row);
+ sb.append(" ");
+ Hex.encNonAscii(sb, col, " ");
+ if (sb.length() > 100) {
+ sb.append(" ...");
+ break outer;
+ }
+ }
+ }
+ }
+
+ return sb.toString();
+ }
+
// async stuff
private AsyncConditionalWriter acw;
private AsyncConditionalWriter bacw;
@@ -843,7 +876,7 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
if (checkForAckCollision(pcm)) {
cd.commitObserver.alreadyAcknowledged();
} else {
- cd.commitObserver.commitFailed();
+ cd.commitObserver.commitFailed(cd.getShortCollisionMessage());
}
return;
}
@@ -959,7 +992,7 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
Futures.addCallback(future, new CommitCallback<Void>(cd) {
@Override
protected void onSuccess(CommitData cd, Void v) throws Exception {
- cd.commitObserver.commitFailed();
+ cd.commitObserver.commitFailed(cd.getShortCollisionMessage());
}
}, env.getSharedResources().getAsyncCommitExecutor());
}
@@ -1111,7 +1144,7 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
private void postCommitPrimary(CommitData cd, long commitTs, Status mutationStatus)
throws Exception {
if (mutationStatus != Status.ACCEPTED) {
- cd.commitObserver.commitFailed();
+ cd.commitObserver.commitFailed(cd.getShortCollisionMessage());
} else {
if (stopAfterPrimaryCommit) {
cd.commitObserver.committed();
http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/2e446e6c/modules/core/src/main/java/org/apache/fluo/core/log/TracingTransaction.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/log/TracingTransaction.java b/modules/core/src/main/java/org/apache/fluo/core/log/TracingTransaction.java
index db2fefb..646113f 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/log/TracingTransaction.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/log/TracingTransaction.java
@@ -266,8 +266,8 @@ public class TracingTransaction extends AbstractTransactionBase implements Async
}
@Override
- public void commitFailed() {
- aco.commitFailed();
+ public void commitFailed(String msg) {
+ aco.commitFailed(msg);
logUnsuccessfulCommit();
}
http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/2e446e6c/modules/core/src/main/java/org/apache/fluo/core/worker/WorkTaskAsync.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/worker/WorkTaskAsync.java b/modules/core/src/main/java/org/apache/fluo/core/worker/WorkTaskAsync.java
index 05b67e2..895fc21 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/worker/WorkTaskAsync.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/worker/WorkTaskAsync.java
@@ -59,7 +59,7 @@ public class WorkTaskAsync implements Runnable {
}
@Override
- public void commitFailed() {
+ public void commitFailed(String msg) {
notificationProcessor.requeueNotification(notificationFinder, notification);
}
}
[2/2] incubator-fluo git commit: fixes #801 corrected collision
tracking
Posted by kt...@apache.org.
fixes #801 corrected collision tracking
Project: http://git-wip-us.apache.org/repos/asf/incubator-fluo/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-fluo/commit/25839df1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-fluo/tree/25839df1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-fluo/diff/25839df1
Branch: refs/heads/master
Commit: 25839df1671608517b9a74b24aef575ec02567d7
Parents: 2e446e6
Author: Keith Turner <ke...@deenlo.com>
Authored: Mon Nov 14 22:10:02 2016 -0500
Committer: Keith Turner <ke...@deenlo.com>
Committed: Mon Nov 14 22:11:00 2016 -0500
----------------------------------------------------------------------
.../org/apache/fluo/core/impl/TransactionImpl.java | 14 ++++++++------
1 file changed, 8 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-fluo/blob/25839df1/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java b/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java
index e2e6d32..07a14b4 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java
@@ -381,14 +381,16 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
private Bytes pval;
private HashSet<Bytes> acceptedRows;
- private Map<Bytes, Set<Column>> rejected = new HashMap<>();
+ private Map<Bytes, Set<Column>> rejected = null;
private void addPrimaryToRejected() {
rejected = Collections.singletonMap(prow, Collections.singleton(pcol));
}
private void addToRejected(Bytes row, Set<Column> columns) {
- rejected = new HashMap<>();
+ if (rejected == null) {
+ rejected = new HashMap<>();
+ }
Set<Column> ret = rejected.put(row, columns);
if (ret != null) {
@@ -406,14 +408,14 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
@Override
public String toString() {
- return prow + " " + pcol + " " + pval + " " + rejected.size();
+ return prow + " " + pcol + " " + pval + " " + getRejected().size();
}
public String getShortCollisionMessage() {
StringBuilder sb = new StringBuilder();
- if (rejected.size() > 0) {
+ if (getRejected().size() > 0) {
int numCollisions = 0;
- for (Set<Column> cols : rejected.values()) {
+ for (Set<Column> cols : getRejected().values()) {
numCollisions += cols.size();
}
@@ -422,7 +424,7 @@ public class TransactionImpl extends AbstractTransactionBase implements AsyncTra
sb.append("):");
String sep = "";
- outer: for (Entry<Bytes, Set<Column>> entry : rejected.entrySet()) {
+ outer: for (Entry<Bytes, Set<Column>> entry : getRejected().entrySet()) {
Bytes row = entry.getKey();
for (Column col : entry.getValue()) {
sb.append(sep);