SentinelHelloCheckAction extends AbstractLeaderAwareHealthCheckAction { @Override protected void doTask() { lastStartTime = System.currentTimeMillis(); RedisInstanceInfo info = getActionInstance().getRedisInstanceInfo(); if (instance.getRedisInstanceInfo().isInActiveDc()) { logger.info("[doTask][{}-{}] in active dc, redis {}", info.getClusterId(), info.getShardId(), instance.getEndpoint()); } subError = null; getActionInstance().getRedisSession().subscribeIfAbsent(HELLO_CHANNEL, new RedisSession.SubscribeCallback() { @Override public void message(String channel, String message) { executors.execute(new Runnable() { @Override public void run() { SentinelHello hello = SentinelHello.fromString(message); hellos.add(hello); } }); } @Override public void fail(Throwable e) { if (ExceptionUtils.isStackTraceUnnecessary(e)) { logger.error("[sub-failed][{}] {}", getActionInstance().getRedisInstanceInfo().getHostPort(), e.getMessage()); } else { logger.error("[sub-failed][{}]", getActionInstance().getRedisInstanceInfo().getHostPort(), e); } subError = e; } }); scheduled.schedule(new AbstractExceptionLogTask() { @Override protected void doRun() throws Exception { processSentinelHellos(); } }, SENTINEL_COLLECT_INFO_INTERVAL, TimeUnit.MILLISECONDS); } SentinelHelloCheckAction(ScheduledExecutorService scheduled, RedisHealthCheckInstance instance,
                                    ExecutorService executors, ConsoleDbConfig consoleDbConfig, ClusterService clusterService);  static final String HELLO_CHANNEL; }
@Test public void testDoScheduleTaskWithSentinelHello() throws Exception { RedisInstanceInfo info = instance.getRedisInstanceInfo(); logger.info("{}: {}", info.getActiveDc(), info.isInActiveDc()); when(metaCache.getSentinelMonitorName(anyString(), anyString())).thenReturn(info.getShardId()); when(metaCache.findMasterInSameShard(any(HostPort.class))).thenReturn(info.getHostPort()); StringBuilder sb = new StringBuilder(SUBSCRIBE_HEADER); for(int i = 0; i < 5; i++) { sb.append(String.format(SENTINEL_HELLO_TEMPLATE, 5000 + i)); } result = new Supplier<String>() { @Override public String get() { return sb.toString(); } }; AtomicInteger counter = new AtomicInteger(0); SentinelHelloCheckAction.SENTINEL_COLLECT_INFO_INTERVAL = 100; action.addListener(new SentinelHelloCollector() { @Override public void onAction(SentinelActionContext context) { Assert.assertEquals(5, context.getResult().size()); counter.incrementAndGet(); } @Override public void stopWatch(HealthCheckAction action) { } }); Assert.assertEquals(1, action.getListeners().size()); action.lastStartTime = System.currentTimeMillis() - 3600000; action.doTask(); waitConditionUntilTimeOut(()->server.getConnected() > 0, 500); sleep(SentinelHelloCheckAction.SENTINEL_COLLECT_INFO_INTERVAL + 100); Assert.assertEquals(1, counter.get()); }