@Test public void testReallocateOrphanClustersIsLeader() throws TransactionExecutionException { when(periscopeNodeRepository.countByLeaderIsTrueAndLastUpdatedIsGreaterThan(anyLong())).thenReturn(0L); SpyTimer spyTimer = new SpyTimer(); when(timerFactory.get()).thenReturn(spyTimer); PeriscopeNode leader = new PeriscopeNode("nodeid"); leader.setLeader(true); when(periscopeNodeRepository.findAllByLastUpdatedIsGreaterThan(anyLong())).thenReturn(Collections.singletonList(leader)); when(clusterRepository.findAllByPeriscopeNodeIdNotInOrPeriscopeNodeIdIsNull(any())).thenReturn(Collections.singletonList(new Cluster())); underTest.leaderElection(); spyTimer.lastTask.run(); verify(transactionService, times(2)).required(any(Supplier.class)); verify(clusterRepository, times(1)).findAllByPeriscopeNodeIdNotInOrPeriscopeNodeIdIsNull(any(List.class)); verify(clusterRepository, times(1)).saveAll(any(List.class)); }
@Scheduled(initialDelay = 35000L, fixedDelay = 30000L) public void leaderElection() { if (periscopeNodeConfig.isNodeIdSpecified()) { long leaders = periscopeNodeRepository.countByLeaderIsTrueAndLastUpdatedIsGreaterThan(clock.getCurrentTimeMillis() - heartbeatThresholdRate); if (leaders == 0L) { metricService.submit(MetricType.LEADER, 0); LOGGER.info("There is no active leader available"); resetTimer(); try { transactionService.required(() -> { periscopeNodeRepository.deallocateLeader(); PeriscopeNode me = periscopeNodeRepository.findById(periscopeNodeConfig.getId()) .orElseThrow(notFound("Periscope node", periscopeNodeConfig.getId())); me.setLeader(true); periscopeNodeRepository.save(me); return null; }); } catch (TransactionExecutionException e) { LOGGER.error("Failed to select node as leader, something went wrong. Message: {}", e.getMessage()); return; } metricService.submit(MetricType.LEADER, 1); LOGGER.info("Selected {} as leader", periscopeNodeConfig.getId()); timer.schedule(new TimerTask() { @Override public void run() { try { long limit = clock.getCurrentTimeMillis() - heartbeatThresholdRate; List<PeriscopeNode> activeNodes = periscopeNodeRepository.findAllByLastUpdatedIsGreaterThan(limit); reallocateOrphanClusters(activeNodes); if (!activeNodes.isEmpty()) { cleanupInactiveNodesByActiveNodes(activeNodes); } } catch (RuntimeException e) { LOGGER.error("Error happend during fetching cluster allocating them to nodes", e); } } }, LEADER_TASK_DELAY, STACK_COLLECTOR_PERIOD); } } }
LeaderElectionService { @Scheduled(initialDelay = 35000L, fixedDelay = 30000L) public void leaderElection() { if (periscopeNodeConfig.isNodeIdSpecified()) { long leaders = periscopeNodeRepository.countByLeaderIsTrueAndLastUpdatedIsGreaterThan(clock.getCurrentTimeMillis() - heartbeatThresholdRate); if (leaders == 0L) { metricService.submit(MetricType.LEADER, 0); LOGGER.info("There is no active leader available"); resetTimer(); try { transactionService.required(() -> { periscopeNodeRepository.deallocateLeader(); PeriscopeNode me = periscopeNodeRepository.findById(periscopeNodeConfig.getId()) .orElseThrow(notFound("Periscope node", periscopeNodeConfig.getId())); me.setLeader(true); periscopeNodeRepository.save(me); return null; }); } catch (TransactionExecutionException e) { LOGGER.error("Failed to select node as leader, something went wrong. Message: {}", e.getMessage()); return; } metricService.submit(MetricType.LEADER, 1); LOGGER.info("Selected {} as leader", periscopeNodeConfig.getId()); timer.schedule(new TimerTask() { @Override public void run() { try { long limit = clock.getCurrentTimeMillis() - heartbeatThresholdRate; List<PeriscopeNode> activeNodes = periscopeNodeRepository.findAllByLastUpdatedIsGreaterThan(limit); reallocateOrphanClusters(activeNodes); if (!activeNodes.isEmpty()) { cleanupInactiveNodesByActiveNodes(activeNodes); } } catch (RuntimeException e) { LOGGER.error("Error happend during fetching cluster allocating them to nodes", e); } } }, LEADER_TASK_DELAY, STACK_COLLECTOR_PERIOD); } } } }
LeaderElectionService { @Scheduled(initialDelay = 35000L, fixedDelay = 30000L) public void leaderElection() { if (periscopeNodeConfig.isNodeIdSpecified()) { long leaders = periscopeNodeRepository.countByLeaderIsTrueAndLastUpdatedIsGreaterThan(clock.getCurrentTimeMillis() - heartbeatThresholdRate); if (leaders == 0L) { metricService.submit(MetricType.LEADER, 0); LOGGER.info("There is no active leader available"); resetTimer(); try { transactionService.required(() -> { periscopeNodeRepository.deallocateLeader(); PeriscopeNode me = periscopeNodeRepository.findById(periscopeNodeConfig.getId()) .orElseThrow(notFound("Periscope node", periscopeNodeConfig.getId())); me.setLeader(true); periscopeNodeRepository.save(me); return null; }); } catch (TransactionExecutionException e) { LOGGER.error("Failed to select node as leader, something went wrong. Message: {}", e.getMessage()); return; } metricService.submit(MetricType.LEADER, 1); LOGGER.info("Selected {} as leader", periscopeNodeConfig.getId()); timer.schedule(new TimerTask() { @Override public void run() { try { long limit = clock.getCurrentTimeMillis() - heartbeatThresholdRate; List<PeriscopeNode> activeNodes = periscopeNodeRepository.findAllByLastUpdatedIsGreaterThan(limit); reallocateOrphanClusters(activeNodes); if (!activeNodes.isEmpty()) { cleanupInactiveNodesByActiveNodes(activeNodes); } } catch (RuntimeException e) { LOGGER.error("Error happend during fetching cluster allocating them to nodes", e); } } }, LEADER_TASK_DELAY, STACK_COLLECTOR_PERIOD); } } }  }
LeaderElectionService { @Scheduled(initialDelay = 35000L, fixedDelay = 30000L) public void leaderElection() { if (periscopeNodeConfig.isNodeIdSpecified()) { long leaders = periscopeNodeRepository.countByLeaderIsTrueAndLastUpdatedIsGreaterThan(clock.getCurrentTimeMillis() - heartbeatThresholdRate); if (leaders == 0L) { metricService.submit(MetricType.LEADER, 0); LOGGER.info("There is no active leader available"); resetTimer(); try { transactionService.required(() -> { periscopeNodeRepository.deallocateLeader(); PeriscopeNode me = periscopeNodeRepository.findById(periscopeNodeConfig.getId()) .orElseThrow(notFound("Periscope node", periscopeNodeConfig.getId())); me.setLeader(true); periscopeNodeRepository.save(me); return null; }); } catch (TransactionExecutionException e) { LOGGER.error("Failed to select node as leader, something went wrong. Message: {}", e.getMessage()); return; } metricService.submit(MetricType.LEADER, 1); LOGGER.info("Selected {} as leader", periscopeNodeConfig.getId()); timer.schedule(new TimerTask() { @Override public void run() { try { long limit = clock.getCurrentTimeMillis() - heartbeatThresholdRate; List<PeriscopeNode> activeNodes = periscopeNodeRepository.findAllByLastUpdatedIsGreaterThan(limit); reallocateOrphanClusters(activeNodes); if (!activeNodes.isEmpty()) { cleanupInactiveNodesByActiveNodes(activeNodes); } } catch (RuntimeException e) { LOGGER.error("Error happend during fetching cluster allocating them to nodes", e); } } }, LEADER_TASK_DELAY, STACK_COLLECTOR_PERIOD); } } }  @PostConstruct void init(); @Scheduled(initialDelay = 35000L, fixedDelay = 30000L) void leaderElection(); }
LeaderElectionService { @Scheduled(initialDelay = 35000L, fixedDelay = 30000L) public void leaderElection() { if (periscopeNodeConfig.isNodeIdSpecified()) { long leaders = periscopeNodeRepository.countByLeaderIsTrueAndLastUpdatedIsGreaterThan(clock.getCurrentTimeMillis() - heartbeatThresholdRate); if (leaders == 0L) { metricService.submit(MetricType.LEADER, 0); LOGGER.info("There is no active leader available"); resetTimer(); try { transactionService.required(() -> { periscopeNodeRepository.deallocateLeader(); PeriscopeNode me = periscopeNodeRepository.findById(periscopeNodeConfig.getId()) .orElseThrow(notFound("Periscope node", periscopeNodeConfig.getId())); me.setLeader(true); periscopeNodeRepository.save(me); return null; }); } catch (TransactionExecutionException e) { LOGGER.error("Failed to select node as leader, something went wrong. Message: {}", e.getMessage()); return; } metricService.submit(MetricType.LEADER, 1); LOGGER.info("Selected {} as leader", periscopeNodeConfig.getId()); timer.schedule(new TimerTask() { @Override public void run() { try { long limit = clock.getCurrentTimeMillis() - heartbeatThresholdRate; List<PeriscopeNode> activeNodes = periscopeNodeRepository.findAllByLastUpdatedIsGreaterThan(limit); reallocateOrphanClusters(activeNodes); if (!activeNodes.isEmpty()) { cleanupInactiveNodesByActiveNodes(activeNodes); } } catch (RuntimeException e) { LOGGER.error("Error happend during fetching cluster allocating them to nodes", e); } } }, LEADER_TASK_DELAY, STACK_COLLECTOR_PERIOD); } } }  @PostConstruct void init(); @Scheduled(initialDelay = 35000L, fixedDelay = 30000L) void leaderElection();  }