public Pair<List<String>, String> getNextEventsFromQueue(SqsClient sqs,
                                                           Option<String> lastCheckpointStr,
                                                           List<Message> processedMessages) {
    processedMessages.clear();
    log.info("Reading messages....");
    try {
      log.info("Start Checkpoint : " + lastCheckpointStr);
      List<Map<String, Object>> eventRecords = getValidEvents(sqs, processedMessages);
      log.info("Number of valid events: " + eventRecords.size());
      List<String> filteredEventRecords = new ArrayList<>();
      long newCheckpointTime = eventRecords.stream()
          .mapToLong(eventRecord -> Date.from(Instant.from(
                  DateTimeFormatter.ISO_INSTANT.parse((String) eventRecord.get(S3_MODEL_EVENT_TIME))))
              .getTime()).max().orElse(lastCheckpointStr.map(Long::parseLong).orElse(0L));

      for (Map<String, Object> eventRecord : eventRecords) {
        filteredEventRecords.add(new ObjectMapper().writeValueAsString(eventRecord).replace("%3D", "=")
            .replace("%24", "$").replace("%A3", "£").replace("%23", "#").replace("%26", "&").replace("%3F", "?")
            .replace("%7E", "~").replace("%25", "%").replace("%2B", "+"));
      }
      // Return the old checkpoint if no messages to consume from queue.
      String newCheckpoint = newCheckpointTime == 0 ? lastCheckpointStr.orElse(null) : String.valueOf(newCheckpointTime);
      return new ImmutablePair<>(filteredEventRecords, newCheckpoint);
    } catch (JSONException | IOException e) {
      throw new HoodieException("Unable to read from SQS: ", e);
    }
  }
@Test
  public void testEventsFromQueueNoMessages() {
    S3EventsMetaSelector selector = new S3EventsMetaSelector(props);
    Map<String, String> attribute = new HashMap<>();
    attribute.put(SQS_ATTR_APPROX_MESSAGES, "0");
    when(sqs.getQueueAttributes(any(GetQueueAttributesRequest.class)))
        .thenReturn(
            GetQueueAttributesResponse.builder()
                .attributesWithStrings(attribute)
                .build());

    List<Message> processed = new ArrayList<>();
    Pair<List<String>, String> eventFromQueue =
        selector.getNextEventsFromQueue(sqs, Option.empty(), processed);

    assertEquals(0, eventFromQueue.getLeft().size());
    assertEquals(0, processed.size());
    assertNull(eventFromQueue.getRight());
  }