* the next even if errbacks for the current txn get called while
* the next one is being processed.
*
*/
if(!open){
throw new EventDeliveryException("Sink was never opened. " +
"Please fix the configuration.");
}
AtomicBoolean txnFail = new AtomicBoolean(false);
Status status = Status.READY;
Channel channel = getChannel();
txn = channel.getTransaction();
txn.begin();
List<PutRequest> actions = new LinkedList<PutRequest>();
List<AtomicIncrementRequest> increments =
new LinkedList<AtomicIncrementRequest>();
for(int i = 0; i < batchSize; i++){
Event event = channel.take();
if(event == null){
status = Status.BACKOFF;
counterGroup.incrementAndGet("channel.underflow");
break;
} else {
serializer.setEvent(event);
actions.addAll(serializer.getActions());
increments.addAll(serializer.getIncrements());
}
}
CountDownLatch latch =
new CountDownLatch(actions.size() + increments.size());
for(PutRequest action : actions) {
Callback<Object, Object> callback =
new SuccessCallback<Object, Object>(latch);
Callback<Object, Object> errback =
new ErrBack<Object, Object>(latch, txnFail);
client.put(action).addCallbacks(callback, errback);
}
for(AtomicIncrementRequest increment : increments) {
Callback<Long, Long> callback =
new SuccessCallback<Long, Long>(latch);
Callback<Long, Long> errback = new ErrBack<Long, Long>(latch, txnFail);
client.atomicIncrement(increment).addCallbacks(callback, errback);
}
try {
latch.await();
} catch (InterruptedException e1) {
this.handleTransactionFailure(txn);
throw new EventDeliveryException("Sink interrupted while waiting" +
"for Hbase callbacks. Exception follows.", e1);
}
/*
* At this point, either the txn has failed
* or all callbacks received and txn is successful.
*
* This need not be in the monitor, since all callbacks for this txn
* have been received. So txnFail will not be modified any more(even if
* it is, it is set from true to true only - false happens only
* in the next process call).
*
*/
if (txnFail.get()) {
this.handleTransactionFailure(txn);
throw new EventDeliveryException("Could not write events to Hbase. " +
"Transaction failed, and rolled back.");
} else {
try{
txn.commit();
} catch (Throwable e) {
try{
txn.rollback();
} catch (Exception e2) {
logger.error("Exception in rollback. Rollback might not have been" +
"successful." , e2);
}
counterGroup.incrementAndGet("transaction.rollback");
logger.error("Failed to commit transaction." +
"Transaction rolled back.", e);
if(e instanceof Error || e instanceof RuntimeException){
logger.error("Failed to commit transaction." +
"Transaction rolled back.", e);
Throwables.propagate(e);
} else {
logger.error("Failed to commit transaction." +
"Transaction rolled back.", e);
throw new EventDeliveryException("Failed to commit transaction." +
"Transaction rolled back.", e);
}
} finally {
txn.close();
}