code: @Override
      public Collection<IQueueMessage<ID, DATA>> getOrphanMessages(long thresholdTimestampMs) {
          if (isEphemeralDisabled()) {
              return null;
          }
          Collection<IQueueMessage<ID, DATA>> result = new ArrayList<>();
          Consumer<Document> processor = (doc) -> result.add(fromDocument(doc));
          Date threshold = new Date(System.currentTimeMillis() - thresholdTimestampMs);
          getCollection().find(Filters.and(Filters.ne(COLLECTION_FIELD_EPHEMERAL_KEY, null),
                  Filters.lte(COLLECTION_FIELD_QUEUE_TIME, threshold))).forEach(processor);
          return result;
      }
docs: {@inheritDoc}