PostgresStorage implements IStorage, IDistributedStorage { @Override public boolean renewLead(UUID leaderId) { if (null != jdbi) { try (Handle h = jdbi.open()) { int rowsUpdated = getPostgresStorage(h).renewLead( leaderId, reaperInstanceId, AppContext.REAPER_INSTANCE_ADDRESS ); if (rowsUpdated == 1) { LOG.debug("Renewed lead on segment {}", leaderId); return true; } LOG.error("Failed to renew lead on segment {}", leaderId); } } return false; } PostgresStorage(UUID reaperInstanceId, DBI jdbi); @VisibleForTesting  PostgresStorage(UUID reaperInstanceId,
                         DBI jdbi,
                         int leaderTimeoutInMinutes,
                         int reaperTimeoutInMinutes,
                         int metricsTimeoutInMinutes,
                         int nodeOperationsTimeoutInMinutes); @Override Cluster getCluster(String clusterName); @Override Cluster deleteCluster(String clusterName); @Override boolean isStorageConnected(); @Override Collection<Cluster> getClusters(); @Override boolean addCluster(Cluster cluster); @Override boolean updateCluster(Cluster cluster); @Override Optional<RepairRun> getRepairRun(UUID id); @Override Collection<RepairRun> getRepairRunsForCluster(String clusterName, Optional<Integer> limit); @Override Collection<RepairRun> getRepairRunsForUnit(UUID repairUnitId); @Override Collection<RepairRun> getRepairRunsWithState(RepairRun.RunState runState); @Override Optional<RepairRun> deleteRepairRun(UUID id); @Override RepairRun addRepairRun(RepairRun.Builder newRepairRun, Collection<RepairSegment.Builder> newSegments); @Override boolean updateRepairRun(RepairRun repairRun); @Override boolean updateRepairRun(RepairRun repairRun, Optional<Boolean> updateRepairState); @Override RepairUnit addRepairUnit(RepairUnit.Builder newRepairUnit); @Override RepairUnit getRepairUnit(UUID id); @Override Optional<RepairUnit> getRepairUnit(RepairUnit.Builder params); @Override boolean updateRepairSegment(RepairSegment repairSegment); @Override Optional<RepairSegment> getRepairSegment(UUID runId, UUID segmentId); @Override Collection<RepairSegment> getRepairSegmentsForRun(UUID runId); @Override Optional<RepairSegment> getNextFreeSegmentInRange(UUID runId, Optional<RingRange> range); @Override Collection<RepairSegment> getSegmentsWithState(UUID runId, RepairSegment.State segmentState); @Override Collection<RepairParameters> getOngoingRepairsInCluster(String clusterName); @Override SortedSet<UUID> getRepairRunIdsForCluster(String clusterName, Optional<Integer> limit); @Override int getSegmentAmountForRepairRun(UUID runId); @Override int getSegmentAmountForRepairRunWithState(UUID runId, RepairSegment.State state); @Override RepairSchedule addRepairSchedule(RepairSchedule.Builder repairSchedule); @Override Optional<RepairSchedule> getRepairSchedule(UUID repairScheduleId); @Override Collection<RepairSchedule> getRepairSchedulesForCluster(String clusterName); @Override Collection<RepairSchedule> getRepairSchedulesForKeyspace(String keyspaceName); @Override Collection<RepairSchedule> getRepairSchedulesForClusterAndKeyspace(String clusterName, String keyspaceName); @Override Collection<RepairSchedule> getAllRepairSchedules(); @Override boolean updateRepairSchedule(RepairSchedule newRepairSchedule); @Override Optional<RepairSchedule> deleteRepairSchedule(UUID id); @Override Collection<RepairRunStatus> getClusterRunStatuses(String clusterName, int limit); @Override Collection<RepairScheduleStatus> getClusterScheduleStatuses(String clusterName); @Override boolean saveSnapshot(Snapshot snapshot); @Override boolean deleteSnapshot(Snapshot snapshot); @Override Snapshot getSnapshot(String clusterName, String snapshotName); @Override Collection<DiagEventSubscription> getEventSubscriptions(); @Override Collection<DiagEventSubscription> getEventSubscriptions(String clusterName); @Override DiagEventSubscription getEventSubscription(UUID id); @Override DiagEventSubscription addEventSubscription(DiagEventSubscription subscription); @Override boolean deleteEventSubscription(UUID id); @Override boolean takeLead(UUID leaderId); @Override boolean takeLead(UUID leaderId, int ttl); @Override boolean renewLead(UUID leaderId); @Override boolean renewLead(UUID leaderId, int ttl); @Override List<UUID> getLeaders(); @Override void releaseLead(UUID leaderId); @Override void saveHeartbeat(); @Override int countRunningReapers(); @Override void storeNodeMetrics(UUID runId, NodeMetrics nodeMetrics); @Override Collection<NodeMetrics> getNodeMetrics(UUID runId); @Override Optional<NodeMetrics> getNodeMetrics(UUID runId, String node); @Override void deleteNodeMetrics(UUID runId, String node); @Override void purgeNodeMetrics(); @Override Optional<RepairSegment> getNextFreeSegmentForRanges(
      UUID runId,
      Optional<RingRange> parallelRange,
      List<RingRange> ranges); @Override void storeMetric(GenericMetric metric); @Override List<GenericMetric> getMetrics(
      String clusterName,
      Optional<String> host,
      String metricDomain,
      String metricType,
      long since); @Override void purgeMetrics(); @Override void storeOperations(String clusterName, OpType operationType, String host, String operationsJson); @Override String listOperations(String clusterName, OpType operationType, String host); @Override void purgeNodeOperations();  }
@Test public void testRenewLead() throws InterruptedException { DBI dbi = new DBI(DB_URL); UUID reaperInstanceId = UUID.randomUUID(); PostgresStorage storage = new PostgresStorage(reaperInstanceId, dbi); Assertions.assertThat(storage.isStorageConnected()).isTrue(); Handle handle = dbi.open(); handle.execute("DELETE from leader"); UUID leaderId = UUID.randomUUID(); int sleepTime = 3; final Instant initialTime = Instant.now(); storage.takeLead(leaderId); TimeUnit.SECONDS.sleep(sleepTime); Assertions.assertThat(storage.renewLead(leaderId)).isTrue(); Instant hbTime = handle.createQuery("SELECT last_heartbeat FROM leader") .mapTo(Timestamp.class) .first() .toInstant(); Duration between = Duration.between(initialTime, hbTime); Assertions.assertThat(between.getSeconds()).isGreaterThanOrEqualTo(sleepTime); }