Skip to content

Commit 252a0dd

Browse files
authored
Add catching and logging of exceptions for s3 scan worker (opensearch-project#3159)
Signed-off-by: Taylor Gray <tylgry@amazon.com>
1 parent f11d882 commit 252a0dd

2 files changed

Lines changed: 17 additions & 1 deletion

File tree

data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/S3ScanPartitionCreationSupplier.java

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -116,10 +116,14 @@ private List<PartitionIdentifier> listFilteredS3ObjectsForBucket(final List<Stri
116116
.map(objectKey -> PartitionIdentifier.builder().withPartitionKey(String.format(BUCKET_OBJECT_PARTITION_KEY_FORMAT, bucket, objectKey)).build())
117117
.collect(Collectors.toList()));
118118

119+
LOG.info("Found page of {} objects from bucket {}", listObjectsV2Response.keyCount(), bucket);
120+
119121
mostRecentLastModifiedTimestamp = getMostRecentLastModifiedTimestamp(listObjectsV2Response, mostRecentLastModifiedTimestamp);
120122
} while (listObjectsV2Response.isTruncated());
121123

122124
globalStateMap.put(bucket, Objects.nonNull(mostRecentLastModifiedTimestamp) ? mostRecentLastModifiedTimestamp.toString() : null);
125+
126+
LOG.info("Returning partitions for {} S3 objects from bucket {}", allPartitionIdentifiers.size(), bucket);
123127
return allPartitionIdentifiers;
124128
}
125129

data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/ScanObjectWorker.java

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ public class ScanObjectWorker implements Runnable{
4242
private static final Logger LOG = LoggerFactory.getLogger(ScanObjectWorker.class);
4343

4444
private static final int STANDARD_BACKOFF_MILLIS = 30_000;
45+
private static final int RETRY_BACKOFF_ON_EXCEPTION_MILLIS = 5_000;
4546

4647
static final int ACKNOWLEDGEMENT_SET_TIMEOUT_SECONDS = Integer.MAX_VALUE;
4748
static final String ACKNOWLEDGEMENT_SET_CALLBACK_METRIC_NAME = "acknowledgementSetCallbackCounter";
@@ -99,7 +100,18 @@ public ScanObjectWorker(final S3Client s3Client,
99100
@Override
100101
public void run() {
101102
while (!shouldStopProcessing) {
102-
startProcessingObject(STANDARD_BACKOFF_MILLIS);
103+
104+
try {
105+
startProcessingObject(STANDARD_BACKOFF_MILLIS);
106+
} catch (final Exception e) {
107+
LOG.error("Received an exception while processing S3 objects, backing off and retrying", e);
108+
try {
109+
Thread.sleep(RETRY_BACKOFF_ON_EXCEPTION_MILLIS);
110+
} catch (InterruptedException ex) {
111+
LOG.error("S3 Scan worker thread interrupted while backing off.", ex);
112+
}
113+
}
114+
103115
}
104116
}
105117

0 commit comments

Comments
 (0)