Update microsoft-365 source pagination logic to not skip time range

**Description**

This commit updates microsoft-365 source pagination to not skip
events in certain cases.

PaginationCrawler saves Item lastModifiedAt as next poll attempt's
startTime in the coordinator table. Currently m365 source sets
Instant.now as lastModifiedAt(nextPollAttemptTime), which could skip
new arrived events between current last event's contentCreated time
and current timestamp.

The lastModifiedAt is updated to be contentCreated time(original
implementation) when there is next page, and eventTime+1ms when there
is no new page. This can ensure no missing event in any scenario as
well as no duplicate event in common scenario.

Signed-off-by: Wenjie Yao <wjyao@amazon.com>

Author: wjyao-amazon

data-prepper-plugins/saas-source-plugins/microsoft-office365-source/src/main/java/org/opensearch/dataprepper/plugins/source/microsoft_office365/service/Office365Service.java

@@ -108,9 +108,8 @@ private void searchForNewLogs(final Instant timestamp,
                             break;
                         }
 
-                        addItemsToQueue(response.getItems(), contentType, itemInfoQueue);
                         nextPageUri = response.getNextPageUri();
-
+                        addItemsToQueue(response.getItems(), contentType, itemInfoQueue, nextPageUri == null);
                     } while (nextPageUri != null);
                 } catch (Exception e) {
                     log.error(NOISY, "Failed to fetch logs for time window {} to {} for content type {}. Will retry this window.",
@@ -140,15 +139,22 @@ private void searchForNewLogs(final Instant timestamp,
 
     private void addItemsToQueue(final List<Map<String, Object>> items,
                                  final String contentType,
-                                 final Queue<ItemInfo> itemInfoQueue) {
+                                 final Queue<ItemInfo> itemInfoQueue,
+                                 final boolean lastPage) {
         items.forEach(item -> {
+            Instant eventTime = Instant.parse((String) item.get(CONTENT_CREATED_KEY));
+            // If last page, add 1ms offset for next poll start time to avoid duplication when all events are processed.
+            // 1ms is m365's smallest time unit so polling data starting from next 1ms would not skip any event.
+            // If not last page, keep nextPollAttemptStartTime to be contentCreated time so to avoid data loss in a rare scenario
+            // where 1ms have multiple events and split by nextPageUri
+            Instant nextPollAttemptStartTime = lastPage ? eventTime.plusMillis(1) : eventTime;
             ItemInfo itemInfo = Office365ItemInfo.builder()
                     .itemId((String) item.get(CONTENT_ID_KEY))
                     .eventTime(Instant.parse((String) item.get(CONTENT_CREATED_KEY)))
                     .partitionKey(contentType + UUID.randomUUID())
                     .metadata(item)
                     .keyAttributes(Map.of(TYPE_KEY, contentType, CONTENT_URI_KEY, item.get(CONTENT_URI_KEY)))
-                    .lastModifiedAt(Instant.now()) // Used to track the time that it was imported
+                    .lastModifiedAt(nextPollAttemptStartTime)
                     .build();
             itemInfoQueue.add(itemInfo);
         });