28425041198

Committed 30 Jun 2026 06:30AM UTC coverage: 90.88% (-0.3%) from 91.186%

Build # 28425041198

Build Type

push

github

Committed by

Hyshmily

Commit Message

refactor: simplify Worker cluster management and failure detection

Signed-off-by: Hyshmily <cxm8607@outlook.com>

Coverage Stats

1250 of 1437 branches covered (86.99%)

Branch coverage included in aggregate %.

10 of 21 new or added lines in 4 files covered. (47.62%)

1 existing line in 1 file now uncovered.

3543 of 3837 relevant lines covered (92.34%)

4.22 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.11

/common/src/main/java/io/github/hyshmily/hotkey/sync/worker/WorkerHeartbeatVerifier.java

/*
 * Copyright 2026 Hyshmily. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.github.hyshmily.hotkey.sync.worker;

import static io.github.hyshmily.hotkey.constants.HotKeyConstants.*;
import static io.github.hyshmily.hotkey.util.TimeSource.currentTimeMillis;

import io.github.hyshmily.hotkey.sharding.ClusterHealthView;
import io.github.hyshmily.hotkey.util.HotKeyThreadFactory;
import java.util.Set;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import lombok.AccessLevel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpTimeoutException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

/**
 * On-demand active heartbeat verifier that probes suspected-dead Workers when
 * passive heartbeat monitoring indicates a potential failure.
 *
 * <p>Under normal operation, the {@link ClusterHealthView} is updated passively
 * by incoming heartbeat messages. This verifier sends a point-to-point PING
 * message to each suspected Worker's dedicated verification queue
 * ({@code hotkey.verify.ping.{workerId}}) via Direct reply-to and awaits
 * a PONG response.
 *
 * <p>Workers that respond are restored to the alive set via
 * {@link ClusterHealthView#recordPong}; Workers that fail repeatedly are
 * marked as stale via {@link ClusterHealthView#markVerificationFailed} and
 * excluded from health-majority calculations.
 *
 * <p><b>Thread safety:</b> The verification loop runs on a single daemon thread
 * ({@code hb-verifier}). The {@link ClusterHealthView} uses {@code ConcurrentHashMap}
 * internally, so concurrent heartbeat reception and verification are safe.
 *
 * @see ClusterHealthView
 * @see WorkerHeartbeatMessage
 */
@RequiredArgsConstructor(access = AccessLevel.PRIVATE)
@Slf4j
public class WorkerHeartbeatVerifier {

  private final RabbitTemplate rabbitTemplate;
  private final ClusterHealthView healthView;
  private final String appInstanceId;
  private final long verifyIntervalMs;
  private final long pingTimeoutMs;
  private final long verifyMaxBackoffMs;
  private final ScheduledExecutorService scheduler;
  private final ConcurrentHashMap<String, Long> nextVerifyTime = new ConcurrentHashMap<>();
  private final boolean ownsScheduler;
  private ScheduledFuture<?> verifyTask;

  private static final String QUEUE_VERIFY_PING_PREFIX = "hotkey.verify.ping.";
  private static final int MAX_BACKOFF_SHIFT = 30;
  private static final int MAX_RETRY = 5;

  /**
   * Parameter object for {@link WorkerHeartbeatVerifier} construction.
   */
  public record VerifierConfig(long verifyIntervalMs, long pingTimeoutMs, long verifyMaxBackoffMs) {}

  /**
   * Creates a verifier with an internally owned single-thread scheduler.
   * The internal daemon thread is named {@code hb-verifier}.
   */
  public WorkerHeartbeatVerifier(
    RabbitTemplate rabbitTemplate,
    ClusterHealthView healthView,
    String appInstanceId,
    VerifierConfig config
  ) {
    this(
      rabbitTemplate,
      healthView,
      appInstanceId,
      config.verifyIntervalMs,
      config.pingTimeoutMs,
      config.verifyMaxBackoffMs,
      Executors.newSingleThreadScheduledExecutor(new HotKeyThreadFactory("hotkey-hb-verifier")),
      true
    );
  }

  /**
   * Creates a verifier with a caller-supplied external scheduler.
   * The caller manages the scheduler lifecycle — {@link #stop()} cancels
   * the task but does not shut down the scheduler.
   */
  public WorkerHeartbeatVerifier(
    RabbitTemplate rabbitTemplate,
    ClusterHealthView healthView,
    String appInstanceId,
    VerifierConfig config,
    ScheduledExecutorService scheduler
  ) {
    this(
      rabbitTemplate,
      healthView,
      appInstanceId,
      config.verifyIntervalMs,
      config.pingTimeoutMs,
      config.verifyMaxBackoffMs,
      scheduler,
      false
    );
  }

  /**
   * Starts the periodic heartbeat verification at a fixed rate.
   *
   * <p>Every {@code verifyIntervalMs} milliseconds, iterates over all Workers in
   * {@link ClusterHealthView} that have exceeded {@code heartbeatTimeoutMs} without
   * a heartbeat and sends each a PING via Direct reply-to. On PONG, the Worker's
   * health record is restored; on timeout, the failure counter is incremented.
   * <p>
   * The verification runs on a daemon background thread. The task is idempotent:
   * subsequent calls to this method after the verifier is already running are
   * silently ignored.
   *
   * @see #verifySuspectedWorkers
   * @see #stop()
   */
  public void start() {
    if (verifyTask != null) {
      return;
    }
    try {
      verifyTask = scheduler.scheduleWithFixedDelay(
        this::verifySuspectedWorkers,
        verifyIntervalMs,
        verifyIntervalMs,
        TimeUnit.MILLISECONDS
      );
    } catch (Exception e) {
      log.error(
        "Failed to start heartbeat verifier scheduler; Worker liveness verification " +
          "will not run. Application continues but stale Workers may not be detected.",
        e
      );
    }
  }

  /**
   * Gracefully stops the periodic heartbeat verification.
   *
   * <p>Cancels the scheduled verification task and, if this verifier owns its
   * scheduler (created via the 6-argument constructor), shuts down the scheduler.
   * When an external scheduler is used (7-argument constructor), the scheduler is
   * left running for the caller to manage.
   *
   * <p>After {@code stop()}, the verifier can be restarted by calling {@link #start()}
   * again.
   */
  public void stop() {
    if (verifyTask != null) {
      verifyTask.cancel(false);
    }
    if (ownsScheduler) {
      scheduler.shutdown();
    }
  }

  /**
   * Iterates over all Workers registered in {@link ClusterHealthView} that are
   * <em>not</em> currently marked alive, sends each a PING via Direct reply-to,
   * and updates the health view based on the response.
   *
   * <p>When cumulative failures across all suspected Workers reach
   * {@code degradeAfterFailures} and the cluster remains unhealthy, the cluster
   * is marked as degraded.
   *
   * <p>This method is called periodically by the scheduled task and is also
   * safe to invoke manually for testing.
   */
  public void verifySuspectedWorkers() {
    try {
      Set<String> suspected = healthView
        .getAllWorkerIds()
        .stream()
        .filter(id -> !healthView.getAliveWorkerIds().contains(id))
        .filter(id -> healthView.getVerifyFailures(id) < MAX_RETRY)
        .collect(Collectors.toSet());

      if (suspected.isEmpty()) {
        return;
      }

      for (String workerId : suspected) {
        Long skipUntil = nextVerifyTime.get(workerId);
        if (skipUntil != null && currentTimeMillis() < skipUntil) {
          log.trace("Worker {} in backoff, skip (remaining={}ms)", workerId, skipUntil - currentTimeMillis());
          continue;
        }

        boolean alive = sendPingAndWaitPong(workerId);
        if (!alive) {
          healthView.markVerificationFailed(workerId);

          int attempt = healthView.getVerifyFailures(workerId);
          if (attempt >= MAX_RETRY) {
            log.warn("Worker {} confirmed dead ({} failures), removing record", workerId, MAX_RETRY);
            healthView.removeRecord(workerId);
            nextVerifyTime.remove(workerId);
            continue;
          }

          long backoffMs = computeBackoffMs(attempt);

          nextVerifyTime.put(workerId, currentTimeMillis() + backoffMs);
          log.warn("Worker {} verification failed (attempt={}, backoff={}ms)", workerId, attempt, backoffMs);
        } else {
          nextVerifyTime.remove(workerId);
          healthView.recordPong(workerId);
        }

        if (!healthView.isClusterHealthy()) {
          log.warn("Cluster is unhealthy after verifying worker {}", workerId);
        }
      }
    } catch (Exception e) {
      log.error("Scheduled verifySuspectedWorkers failed", e);
    }
  }

  /**
   * <p>Uses a two-phase exponential strategy: <em>dense start, steep extension</em>.
   * First 3 attempts grow slowly (half the naive power) so early retries cluster
   * closely; attempts 4-5 grow aggressively (double the naive power) to spread
   * late retries further apart. Capped at {@code verifyMaxBackoffMs}.
   *
   * <table>
   *   <caption>Multiplier over {@code verifyIntervalMs}</caption>
   *   <tr><th>Attempt</th><th>Shift</th><th>Multiplier</th></tr>
   *   <tr><td>1</td><td>0</td><td>1×</td></tr>
   *   <tr><td>2</td><td>1</td><td>2×</td></tr>
   *   <tr><td>3</td><td>2</td><td>4×</td></tr>
   *   <tr><td>4</td><td>5</td><td>32×</td></tr>
   *   <tr><td>5</td><td>6</td><td>64×</td></tr>
   * </table>
   *
   * @param attempt the number of consecutive failures (1-based)
   * @return the backoff duration in milliseconds
   */
  protected long computeBackoffMs(int attempt) {
    int shift = attempt <= 3 ? (attempt - 1) : (attempt + 1);
    return Math.min(verifyMaxBackoffMs, verifyIntervalMs * (1L << Math.min(shift, MAX_BACKOFF_SHIFT)));
  }

  /**
   * Sends a point-to-point PING message to the given Worker's dedicated verification
   * queue ({@code hotkey.verify.ping.{workerId}}) and waits for a PONG response
   * via AMQP Direct reply-to ({@code amq.rabbitmq.reply-to}).
   *
   * <p>The PING carries the application instance ID in the
   * {@link io.github.hyshmily.hotkey.constants.HotKeyConstants#AMQP_HEADER_VERIFY_APP_INSTANCE} header so the Worker
   * can identify the requester. The reply timeout is set in the message properties.
   *
   * @param workerId the Worker to ping; must not be null
   * @return {@code true} if a non-null PONG response was received within
   *         {@code pingTimeoutMs}; {@code false} if the request timed out
   *         ({@link AmqpTimeoutException}) or the queue is unreachable
   */
  public boolean sendPingAndWaitPong(String workerId) {
    MessageProperties props = new MessageProperties();
    props.setHeader(AMQP_HEADER_VERIFY_TYPE, AMQP_HEADER_VERIFY_PING);
    props.setHeader(AMQP_HEADER_VERIFY_APP_INSTANCE, appInstanceId);
    props.setReplyTo("amq.rabbitmq.reply-to");

    Message ping = new Message(new byte[0], props);

    try {
      rabbitTemplate.setReplyTimeout((int) pingTimeoutMs);
      Message pong = rabbitTemplate.sendAndReceive("", QUEUE_VERIFY_PING_PREFIX + workerId, ping);
      return pong != null;
    } catch (AmqpTimeoutException e) {
      return false;
    }
  }
}

1	/*
2	* Copyright 2026 Hyshmily. All Rights Reserved.
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16	package io.github.hyshmily.hotkey.sync.worker;
17
18	import static io.github.hyshmily.hotkey.constants.HotKeyConstants.*;
19	import static io.github.hyshmily.hotkey.util.TimeSource.currentTimeMillis;
20
21	import io.github.hyshmily.hotkey.sharding.ClusterHealthView;
22	import io.github.hyshmily.hotkey.util.HotKeyThreadFactory;
23	import java.util.Set;
24	import java.util.concurrent.*;
25	import java.util.stream.Collectors;
26	import lombok.AccessLevel;
27	import lombok.RequiredArgsConstructor;
28	import lombok.extern.slf4j.Slf4j;
29	import org.springframework.amqp.AmqpTimeoutException;
30	import org.springframework.amqp.core.Message;
31	import org.springframework.amqp.core.MessageProperties;
32	import org.springframework.amqp.rabbit.core.RabbitTemplate;
33
34	/**
35	* On-demand active heartbeat verifier that probes suspected-dead Workers when
36	* passive heartbeat monitoring indicates a potential failure.
37	*
38	* <p>Under normal operation, the {@link ClusterHealthView} is updated passively
39	* by incoming heartbeat messages. This verifier sends a point-to-point PING
40	* message to each suspected Worker's dedicated verification queue
41	* ({@code hotkey.verify.ping.{workerId}}) via Direct reply-to and awaits
42	* a PONG response.
43	*
44	* <p>Workers that respond are restored to the alive set via
45	* {@link ClusterHealthView#recordPong}; Workers that fail repeatedly are
46	* marked as stale via {@link ClusterHealthView#markVerificationFailed} and
47	* excluded from health-majority calculations.
48	*
49	* <p><b>Thread safety:</b> The verification loop runs on a single daemon thread
50	* ({@code hb-verifier}). The {@link ClusterHealthView} uses {@code ConcurrentHashMap}
51	* internally, so concurrent heartbeat reception and verification are safe.
52	*
53	* @see ClusterHealthView
54	* @see WorkerHeartbeatMessage
55	*/
56	@RequiredArgsConstructor(access = AccessLevel.PRIVATE)
57	@Slf4j	4✔
58	public class WorkerHeartbeatVerifier {
59
60	private final RabbitTemplate rabbitTemplate;
61	private final ClusterHealthView healthView;
62	private final String appInstanceId;
63	private final long verifyIntervalMs;
64	private final long pingTimeoutMs;
65	private final long verifyMaxBackoffMs;
66	private final ScheduledExecutorService scheduler;
67	private final ConcurrentHashMap<String, Long> nextVerifyTime = new ConcurrentHashMap<>();
68	private final boolean ownsScheduler;
69	private ScheduledFuture<?> verifyTask;
70
71	private static final String QUEUE_VERIFY_PING_PREFIX = "hotkey.verify.ping.";
72	private static final int MAX_BACKOFF_SHIFT = 30;
73	private static final int MAX_RETRY = 5;
74
75	/**
76	* Parameter object for {@link WorkerHeartbeatVerifier} construction.
77	*/
78	public record VerifierConfig(long verifyIntervalMs, long pingTimeoutMs, long verifyMaxBackoffMs) {}	12✔
79
80	/**
81	* Creates a verifier with an internally owned single-thread scheduler.
82	* The internal daemon thread is named {@code hb-verifier}.
83	*/
84	public WorkerHeartbeatVerifier(
85	RabbitTemplate rabbitTemplate,
86	ClusterHealthView healthView,
87	String appInstanceId,
88	VerifierConfig config
89	) {
90	this(	15✔
91	rabbitTemplate,
92	healthView,
93	appInstanceId,
94	config.verifyIntervalMs,
95	config.pingTimeoutMs,
96	config.verifyMaxBackoffMs,
97	Executors.newSingleThreadScheduledExecutor(new HotKeyThreadFactory("hotkey-hb-verifier")),	2✔
98	true
99	);
100	}	1✔
101
102	/**
103	* Creates a verifier with a caller-supplied external scheduler.
104	* The caller manages the scheduler lifecycle — {@link #stop()} cancels
105	* the task but does not shut down the scheduler.
106	*/
107	public WorkerHeartbeatVerifier(
108	RabbitTemplate rabbitTemplate,
109	ClusterHealthView healthView,
110	String appInstanceId,
111	VerifierConfig config,
112	ScheduledExecutorService scheduler
113	) {
114	this(	13✔
115	rabbitTemplate,
116	healthView,
117	appInstanceId,
118	config.verifyIntervalMs,
119	config.pingTimeoutMs,
120	config.verifyMaxBackoffMs,
121	scheduler,
122	false
123	);
124	}	1✔
125
126	/**
127	* Starts the periodic heartbeat verification at a fixed rate.
128	*
129	* <p>Every {@code verifyIntervalMs} milliseconds, iterates over all Workers in
130	* {@link ClusterHealthView} that have exceeded {@code heartbeatTimeoutMs} without
131	* a heartbeat and sends each a PING via Direct reply-to. On PONG, the Worker's
132	* health record is restored; on timeout, the failure counter is incremented.
133	* <p>
134	* The verification runs on a daemon background thread. The task is idempotent:
135	* subsequent calls to this method after the verifier is already running are
136	* silently ignored.
137	*
138	* @see #verifySuspectedWorkers
139	* @see #stop()
140	*/
141	public void start() {
142	if (verifyTask != null) {	3✔
143	return;	1✔
144	}
145	try {
146	verifyTask = scheduler.scheduleWithFixedDelay(	12✔
147	this::verifySuspectedWorkers,
148	verifyIntervalMs,
149	verifyIntervalMs,
150	TimeUnit.MILLISECONDS
151	);
152	} catch (Exception e) {	1✔
153	log.error(	4✔
154	"Failed to start heartbeat verifier scheduler; Worker liveness verification " +
155	"will not run. Application continues but stale Workers may not be detected.",
156	e
157	);
158	}	1✔
159	}	1✔
160
161	/**
162	* Gracefully stops the periodic heartbeat verification.
163	*
164	* <p>Cancels the scheduled verification task and, if this verifier owns its
165	* scheduler (created via the 6-argument constructor), shuts down the scheduler.
166	* When an external scheduler is used (7-argument constructor), the scheduler is
167	* left running for the caller to manage.
168	*
169	* <p>After {@code stop()}, the verifier can be restarted by calling {@link #start()}
170	* again.
171	*/
172	public void stop() {
173	if (verifyTask != null) {	3✔
174	verifyTask.cancel(false);	5✔
175	}
176	if (ownsScheduler) {	3!
177	scheduler.shutdown();	3✔
178	}
179	}	1✔
180
181	/**
182	* Iterates over all Workers registered in {@link ClusterHealthView} that are
183	* <em>not</em> currently marked alive, sends each a PING via Direct reply-to,
184	* and updates the health view based on the response.
185	*
186	* <p>When cumulative failures across all suspected Workers reach
187	* {@code degradeAfterFailures} and the cluster remains unhealthy, the cluster
188	* is marked as degraded.
189	*
190	* <p>This method is called periodically by the scheduled task and is also
191	* safe to invoke manually for testing.
192	*/
193	public void verifySuspectedWorkers() {
194	try {
195	Set<String> suspected = healthView	2✔
196	.getAllWorkerIds()	1✔
197	.stream()	3✔
198	.filter(id -> !healthView.getAliveWorkerIds().contains(id))	13✔
199	.filter(id -> healthView.getVerifyFailures(id) < MAX_RETRY)	10!
200	.collect(Collectors.toSet());	4✔
201
202	if (suspected.isEmpty()) {	3✔
203	return;	1✔
204	}
205
206	for (String workerId : suspected) {	10✔
207	Long skipUntil = nextVerifyTime.get(workerId);	6✔
208	if (skipUntil != null && currentTimeMillis() < skipUntil) {	2!
209	log.trace("Worker {} in backoff, skip (remaining={}ms)", workerId, skipUntil - currentTimeMillis());	×
210	continue;	×
211	}
212
213	boolean alive = sendPingAndWaitPong(workerId);	4✔
214	if (!alive) {	2✔
215	healthView.markVerificationFailed(workerId);	4✔
216
217	int attempt = healthView.getVerifyFailures(workerId);	5✔
218	if (attempt >= MAX_RETRY) {	3!
NEW 219	log.warn("Worker {} confirmed dead ({} failures), removing record", workerId, MAX_RETRY);	×
NEW 220	healthView.removeRecord(workerId);	×
NEW 221	nextVerifyTime.remove(workerId);	×
NEW 222	continue;	×
223	}
224
225	long backoffMs = computeBackoffMs(attempt);	4✔
226
227	nextVerifyTime.put(workerId, currentTimeMillis() + backoffMs);	9✔
228	log.warn("Worker {} verification failed (attempt={}, backoff={}ms)", workerId, attempt, backoffMs);	19✔
229	} else {	1✔
230	nextVerifyTime.remove(workerId);	5✔
231	healthView.recordPong(workerId);	4✔
232	}
233
234	if (!healthView.isClusterHealthy()) {	4!
235	log.warn("Cluster is unhealthy after verifying worker {}", workerId);	4✔
236	}
237	}	1✔
238	} catch (Exception e) {	1✔
239	log.error("Scheduled verifySuspectedWorkers failed", e);	4✔
240	}	1✔
241	}	1✔
242
243	/**
244	* <p>Uses a two-phase exponential strategy: <em>dense start, steep extension</em>.
245	* First 3 attempts grow slowly (half the naive power) so early retries cluster
246	* closely; attempts 4-5 grow aggressively (double the naive power) to spread
247	* late retries further apart. Capped at {@code verifyMaxBackoffMs}.
248	*
249	* <table>
250	* <caption>Multiplier over {@code verifyIntervalMs}</caption>
251	* <tr><th>Attempt</th><th>Shift</th><th>Multiplier</th></tr>
252	* <tr><td>1</td><td>0</td><td>1×</td></tr>
253	* <tr><td>2</td><td>1</td><td>2×</td></tr>
254	* <tr><td>3</td><td>2</td><td>4×</td></tr>
255	* <tr><td>4</td><td>5</td><td>32×</td></tr>
256	* <tr><td>5</td><td>6</td><td>64×</td></tr>
257	* </table>
258	*
259	* @param attempt the number of consecutive failures (1-based)
260	* @return the backoff duration in milliseconds
261	*/
262	protected long computeBackoffMs(int attempt) {
263	int shift = attempt <= 3 ? (attempt - 1) : (attempt + 1);	8!
264	return Math.min(verifyMaxBackoffMs, verifyIntervalMs * (1L << Math.min(shift, MAX_BACKOFF_SHIFT)));	12✔
265	}
266
267	/**
268	* Sends a point-to-point PING message to the given Worker's dedicated verification
269	* queue ({@code hotkey.verify.ping.{workerId}}) and waits for a PONG response
270	* via AMQP Direct reply-to ({@code amq.rabbitmq.reply-to}).
271	*
272	* <p>The PING carries the application instance ID in the
273	* {@link io.github.hyshmily.hotkey.constants.HotKeyConstants#AMQP_HEADER_VERIFY_APP_INSTANCE} header so the Worker
274	* can identify the requester. The reply timeout is set in the message properties.
275	*
276	* @param workerId the Worker to ping; must not be null
277	* @return {@code true} if a non-null PONG response was received within
278	* {@code pingTimeoutMs}; {@code false} if the request timed out
279	* ({@link AmqpTimeoutException}) or the queue is unreachable
280	*/
281	public boolean sendPingAndWaitPong(String workerId) {
282	MessageProperties props = new MessageProperties();	4✔
283	props.setHeader(AMQP_HEADER_VERIFY_TYPE, AMQP_HEADER_VERIFY_PING);	4✔
284	props.setHeader(AMQP_HEADER_VERIFY_APP_INSTANCE, appInstanceId);	5✔
285	props.setReplyTo("amq.rabbitmq.reply-to");	3✔
286
287	Message ping = new Message(new byte[0], props);	7✔
288
289	try {
290	rabbitTemplate.setReplyTimeout((int) pingTimeoutMs);	7✔
291	Message pong = rabbitTemplate.sendAndReceive("", QUEUE_VERIFY_PING_PREFIX + workerId, ping);	8✔
292	return pong != null;	6✔
293	} catch (AmqpTimeoutException e) {	1✔
294	return false;	2✔
295	}
296	}
297	}

Hyshmily / hotkey / 28425041198

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous