public void writeComplete(int rc, long ledgerId, long entryId, Object ctx){
//PendingAddOp pOp;
//synchronized(pendingAdds){
//pOp = pendingAdds.get(entryId);
//}
SubAddOp sAdd = (SubAddOp) ctx;
PendingOp pOp = sAdd.pOp;
Integer sId = sAdd.bIndex;
if(pOp == null){
LOG.error("No such an entry ID: " + entryId + "(" + ledgerId + ")");
return;
}
ArrayList<BookieHandle> list = lh.getBookies();
if(rc == 0){
// Everything went ok with this op
synchronized(pOp){
pOp.bookieIdRecv.add(sId);
lh.setLastRecvCorrectly(sId, entryId);
if(pOp.bookieIdRecv.size() >= lh.getQuorumSize()){
sAdd.op.setReady();
}
}
} else {
//LOG.warn("Error sending write request: " + rc + " : " + ledgerId + ": " + lh.getBookies().size());
/*
* If ledger is closed already, then simply return
*/
if(lh.getId() == -1){
LOG.warn("Ledger identifier is not valid");
return;
}
HashSet<Integer> ids;
synchronized(pOp){
pOp.bookieIdSent.add(sId);
ids = pOp.bookieIdSent;
//Check if we tried all possible bookies already
if(ids.size() == lh.getBookies().size()){
if(pOp.retries++ >= MAXRETRIES){
//Call back with error code
sAdd.op.setErrorCode(BKDefs.ENR);
sAdd.op.setReady();
return;
}
ids.clear();
}
// Select another bookie that we haven't contacted yet
try{
//LOG.info("Selecting another bookie " + entryId);
int bCounter;
if(sId >= (entryId % (lh.getBookies().size() + 1))){
bCounter = sId - (((int) entryId) % (lh.getBookies().size() + 1));
} else {
bCounter = (lh.getBookies().size() + 1) - (((int) entryId) % (lh.getBookies().size() + 1)) - sId;
}
int tmpId = (((int) entryId) + lh.getQuorumSize()) % (lh.getBookies().size() + 1);
int newId = tmpId % lh.getBookies().size();
//LOG.info("Sending a new add operation to bookie: " + newId + ", " + lh.getBookies().get(newId).addr);
BookieHandle bh = lh.getBookies().get(newId);
//LOG.info("Got handle for " + newId);
bh.sendAdd(lh, new SubAddOp(sAdd.op,
pOp,
newId,
this), entryId);
//LOG.info("Ended " + entryId + ", " + newId);