28637103998

Committed 03 Jul 2026 03:48AM UTC coverage: 90.399% (-0.2%) from 90.63%

Build # 28637103998

Build Type

push

github

Committed by

Hyshmily

Commit Message

fix: add packaging to gitignore, fix flaky jitter test, update docs references

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

Coverage Stats

1264 of 1461 branches covered (86.52%)

Branch coverage included in aggregate %.

3604 of 3924 relevant lines covered (91.85%)

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

Hyshmily / hotkey / 28637103998

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