docs: @Override
      public Collection<IQueueMessage<ID, DATA>> getOrphanMessages(long thresholdTimestampMs) {
          if (isEphemeralDisabled()) {
              return null;
          }
          BinaryJedisCommands jc = getBinaryJedisCommands();
          try {
              long now = System.currentTimeMillis();
              Collection<IQueueMessage<ID, DATA>> result = new HashSet<>();
              byte[] min = "0".getBytes();
              byte[] max = String.valueOf(now - thresholdTimestampMs).getBytes();
              Set<byte[]> fields = jc.zrangeByScore(getRedisSortedSetNameAsBytes(), min, max, 0, 100);
              for (byte[] field : fields) {
                  byte[] data = jc.hget(getRedisHashNameAsBytes(), field);
                  IQueueMessage<ID, DATA> msg = deserialize(data);
                  if (msg != null) {
                      result.add(msg);
                  }
              }
              return result;
          } finally {
              closeJedisCommands(jc);
          }
      }