Fixes prevent multiple calls to SyncCommitObserver
package org.apache.fluo.core.async;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import com.google.common.base.Preconditions;
import org.apache.fluo.api.exceptions.CommitException;
import org.apache.fluo.api.exceptions.FluoException;
import org.apache.fluo.core.exceptions.AlreadyAcknowledgedException;
@@ -28,33 +30,44 @@ public class SyncCommitObserver implements AsyncCommitObserver {
private volatile boolean aacked = false;
private volatile Exception error = null;
private volatile String commitFailMsg = "";
+ private AtomicBoolean isMethodCalled = new AtomicBoolean(false);
+ private static final String METHOD_CALLED_VALIDATION_CHECK_MESSAGE =
+ "One of the methods committed,failed,alreadyAcknowledged,commitFailed"
+ + " was already called";
@Override
public void committed() {
+ Preconditions.checkState(isMethodCalled.compareAndSet(false, true),
+ METHOD_CALLED_VALIDATION_CHECK_MESSAGE);
committed = Boolean.TRUE;
cdl.countDown();
}
@Override
public void failed(Throwable t) {
+ Preconditions.checkState(isMethodCalled.compareAndSet(false, true),
+ METHOD_CALLED_VALIDATION_CHECK_MESSAGE);
error = (Exception) t;
cdl.countDown();
}
@Override
public void alreadyAcknowledged() {
+ Preconditions.checkState(isMethodCalled.compareAndSet(false, true),
+ METHOD_CALLED_VALIDATION_CHECK_MESSAGE);
aacked = true;
cdl.countDown();
}
@Override
public void commitFailed(String msg) {
+ Preconditions.checkState(isMethodCalled.compareAndSet(false, true),
+ METHOD_CALLED_VALIDATION_CHECK_MESSAGE);
committed = false;
commitFailMsg = msg;
cdl.countDown();
}
-
public void waitForCommit() {
try {
cdl.await();
