static void scheduleStaggered(BiConsumer<Long, Long> scheduler,
                                  Duration interval, Instant now,
                                  String hostname, String clusterHostnames) {
        long delayMillis = 0;
        long intervalMillis = interval.toMillis();
        List<String> hostnames = Stream.of(clusterHostnames.split(","))
                                       .map(hostPort -> hostPort.split(":")[0])
                                       .toList();
        if (hostnames.contains(hostname)) {
            long offset = hostnames.indexOf(hostname) * intervalMillis;
            intervalMillis *= hostnames.size();
            delayMillis = Math.floorMod(offset - now.toEpochMilli(), intervalMillis);
        }
        scheduler.accept(delayMillis, intervalMillis);
    }
@Test
    void testStaggering() {
        scheduleStaggered((delayMillis, intervalMillis) -> {
                              assertEquals(0, delayMillis);
                              assertEquals(10, intervalMillis);
                          },
                          Duration.ofMillis(10),
                          Instant.ofEpochMilli(27),
                          "host",
                          "nys:123,hark:123");

        scheduleStaggered((delayMillis, intervalMillis) -> {
                              assertEquals(3, delayMillis);
                              assertEquals(10, intervalMillis);
                          },
                          Duration.ofMillis(10),
                          Instant.ofEpochMilli(27),
                          "host",
                          "host:123");

        scheduleStaggered((delayMillis, intervalMillis) -> {
                              assertEquals(7, delayMillis);
                              assertEquals(20, intervalMillis);
                          },
                          Duration.ofMillis(10),
                          Instant.ofEpochMilli(13),
                          "host",
                          "host:123,:nys:321");

        scheduleStaggered((delayMillis, intervalMillis) -> {
                              assertEquals(17, delayMillis);
                              assertEquals(20, intervalMillis);
                          },
                          Duration.ofMillis(10),
                          Instant.ofEpochMilli(13),
                          "nys",
                          "host:123,nys:321");
    }