ClusterNodeInfo { public boolean renewLease() throws DocumentStoreException { long now = getCurrentTime(); if (LOG.isTraceEnabled()) { LOG.trace("renewLease - leaseEndTime: " + leaseEndTime + ", leaseTime: " + leaseTime + ", leaseUpdateInterval: " + leaseUpdateInterval); } if (now < leaseEndTime - leaseTime + leaseUpdateInterval) { return false; } long updatedLeaseEndTime; synchronized(this) { if (leaseCheckFailed) { throw leaseExpired(LEASE_CHECK_FAILED_MSG + " (since " + asISO8601(leaseEndTime) + ")", true); } now = getCurrentTime(); updatedLeaseEndTime = now + leaseTime; } if (leaseCheckMode == LeaseCheckMode.STRICT) { if (isLeaseExpired(now)) { synchronized (this) { if (leaseCheckFailed) { throw leaseExpired(LEASE_CHECK_FAILED_MSG + " (since " + asISO8601(leaseEndTime) + ")", true); } leaseCheckFailed = true; } String template = "%s (mode: %s, leaseEndTime: %d (%s), leaseTime: %d, leaseFailureMargin: %d, " + "lease check end time (leaseEndTime - leaseFailureMargin): %d (%s), now: %d (%s), remaining: %d)" + " Need to stop oak-store-document/DocumentNodeStoreService."; String errorMsg = String.format(template, LEASE_CHECK_FAILED_MSG, leaseCheckMode.name(), leaseEndTime, asISO8601(leaseEndTime), leaseTime, leaseFailureMargin, leaseEndTime - leaseFailureMargin, asISO8601(leaseEndTime - leaseFailureMargin), now, asISO8601(now), (leaseEndTime - leaseFailureMargin) - now); LOG.error(errorMsg); handleLeaseFailure(errorMsg); return false; } } UpdateOp update = new UpdateOp("" + id, false); update.set(LEASE_END_KEY, updatedLeaseEndTime); if (leaseCheckMode != LeaseCheckMode.DISABLED) { update.equals(LEASE_END_KEY, null, previousLeaseEndTime); update.equals(STATE, null, ACTIVE.name()); update.notEquals(REV_RECOVERY_LOCK, ACQUIRED.name()); } if (LOG.isDebugEnabled()) { LOG.debug("Renewing lease for cluster id " + id + " with UpdateOp " + update); } Stopwatch sw = Stopwatch.createStarted(); DocumentStoreException dse; Object result = null; try { ClusterNodeInfoDocument doc = store.findAndUpdate(Collection.CLUSTER_NODES, update); result = doc; if (doc == null) { synchronized(this) { if (leaseCheckFailed) { throw leaseExpired(LEASE_CHECK_FAILED_MSG, true); } leaseCheckFailed = true; } String errorMsg = LEASE_CHECK_FAILED_MSG + " (Could not update lease anymore, someone else in the cluster " + "must have noticed this instance' slowness already. " + "Going to invoke leaseFailureHandler!)"; try { ClusterNodeInfoDocument current = store.find(Collection.CLUSTER_NODES, "" + id); if (current != null) { Object leaseEnd = current.get(LEASE_END_KEY); Object recoveryLock = current.get(REV_RECOVERY_LOCK); Object recoveryBy = current.get(REV_RECOVERY_BY); Object cnState = current.get(STATE); errorMsg += " (leaseEnd: " + leaseEnd + " (expected: " + leaseEndTime + ")" + ", (state: " + cnState + " (expected: " + ACTIVE.name() + ")" + ", recoveryLock: " + recoveryLock + ", recoveryBy: " + recoveryBy + ")"; } } catch (DocumentStoreException ex) { LOG.error("trying to read ClusterNodeInfo for cluster id " + id, ex); } LOG.error(errorMsg); handleLeaseFailure(errorMsg); return false; } leaseEndTime = updatedLeaseEndTime; previousLeaseEndTime = leaseEndTime; String mode = (String) doc.get(READ_WRITE_MODE_KEY); if (mode != null && !mode.equals(readWriteMode)) { readWriteMode = mode; store.setReadWriteMode(mode); } return true; } catch (DocumentStoreException e) { dse = e; result = e.toString(); } finally { sw.stop(); String msg = "Lease renewal for cluster id {} took {}, resulted in: {}"; if (sw.elapsed(TimeUnit.SECONDS) > 10) { LOG.warn(msg, id, sw, result); } else if (sw.elapsed(TimeUnit.SECONDS) > 1) { LOG.info(msg, id, sw, result); } else if (LOG.isDebugEnabled()) { LOG.debug(msg, id, sw, result); } } while (getCurrentTime() < updatedLeaseEndTime) { synchronized (this) { if (leaseCheckFailed) { break; } } long t1 = clock.getTime(); ClusterNodeInfoDocument doc; try { doc = store.find(Collection.CLUSTER_NODES, String.valueOf(id)); } catch (DocumentStoreException e) { LOG.warn("Reading ClusterNodeInfoDocument for id " + id + " failed", e); try { clock.waitUntil(t1 + 1000); } catch (InterruptedException iex) { } continue; } if (doc != null) { if (!doc.isActive()) { LOG.warn("ClusterNodeInfoDocument for id {} is not active " + "anymore. {}", id, doc); break; } else if (doc.getLeaseEndTime() == previousLeaseEndTime || doc.getLeaseEndTime() == updatedLeaseEndTime) { previousLeaseEndTime = doc.getLeaseEndTime(); leaseEndTime = doc.getLeaseEndTime(); break; } else { break; } } else { LOG.warn("ClusterNodeInfoDocument for id {} does not exist anymore", id); break; } } throw dse; } private  ClusterNodeInfo(int id, DocumentStore store, String machineId,
                            String instanceId, boolean newEntry, boolean invisible); int getId(); static ClusterNodeInfo getReadOnlyInstance(DocumentStore store); static ClusterNodeInfo getInstance(DocumentStore store,
        RecoveryHandler recoveryHandler,
        String machineId,
        String instanceId,
        int configuredClusterId); static ClusterNodeInfo getInstance(DocumentStore store,
                                              RecoveryHandler recoveryHandler,
                                              String machineId,
                                              String instanceId,
                                              int configuredClusterId,
                                              boolean invisible); void performLeaseCheck(); boolean renewLease(); void setInfo(Map<String, String> info); long getLeaseTime(); long getLeaseEndTime(); void setLeaseFailureHandler(LeaseFailureHandler leaseFailureHandler); void dispose(); @Override String toString(); static final String LEASE_END_KEY; static final String START_TIME_KEY; static final String LAST_WRITTEN_ROOT_REV_KEY; static final String STATE; static final String BROADCAST_ID; static final String BROADCAST_LISTENER; static final String REV_RECOVERY_LOCK; static final String REV_RECOVERY_BY; static final String INVISIBLE; static final int DEFAULT_LEASE_DURATION_MILLIS; static final int DEFAULT_LEASE_UPDATE_INTERVAL_MILLIS; static final int DEFAULT_LEASE_FAILURE_MARGIN_MILLIS; static final boolean DEFAULT_LEASE_CHECK_DISABLED; }
@Test public void renewLease() throws Exception { ClusterNodeInfo info = newClusterNodeInfo(1); long leaseEnd = info.getLeaseEndTime(); waitLeaseUpdateInterval(); assertTrue(info.renewLease()); assertTrue(info.getLeaseEndTime() > leaseEnd); assertFalse(handler.isLeaseFailure()); }