Copilot commented on code in PR #69363:
URL: https://github.com/apache/doris/pull/69363#discussion_r4332943979
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProvider.java:
##########
@@ -573,10 +578,22 @@ private ParsedMessage parseMessage(Message message) {
if (isCommitted(key, message.messageId(),
message.receiptHandle())) {
continue;
}
+ String path = "s3://" + bucket + "/" + key;
+ RemoteObject object;
+ try {
+ // Notifications can outlive an overwrite; refresh both scan
size and cache identity.
+ object = objectStorage.headObject(path);
Review Comment:
Building an S3 URI via string concatenation can produce invalid/ambiguous
URIs for keys containing characters like spaces, `?`, or `#` (URI reserved
characters). Since this class already imports `ObjectStorageUri`, consider
constructing the location via the repository’s URI helper (e.g.,
`ObjectStorageUri`) and using its normalized string form for both
`headObject(...)` and `TBrokerFileStatus` to avoid subtle parsing/encoding bugs.
##########
fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java:
##########
@@ -125,6 +125,31 @@ public void testOnceMarksLastBatchAndRecovers() throws
Exception {
}
}
+ @Test
+ public void testListingRefreshesFileSizeAndModificationTime() throws
Exception {
+ FileSystem fs = Mockito.mock(FileSystem.class);
+ try (MockedStatic<FileSystemFactory> factory =
Mockito.mockStatic(FileSystemFactory.class)) {
+ factory.when(() ->
FileSystemFactory.getFileSystem(Mockito.any(StorageAdapter.class))).thenReturn(fs);
+ for (S3SourceOffsetProvider provider : List.of(
+ new S3SourceOffsetProvider(), new
S3SourceOffsetProvider(ONCE_PROPS))) {
+ Mockito.when(fs.globListWithLimit(Mockito.any(Location.class),
Mockito.isNull(),
+ Mockito.anyLong(), Mockito.anyLong())).thenReturn(
+ new GlobListing(List.of(new
FileEntry(Location.of("s3://bucket/data/a.csv"),
+ 19, false, 1767225600000L, null)),
"bucket", "data/", "data/a.csv"),
+ new GlobListing(List.of(new
FileEntry(Location.of("s3://bucket/data/a.csv"),
+ 170, false, 1767225601000L, null)),
"bucket", "data/", "data/a.csv"));
Review Comment:
The epoch-millis literals make the test harder to read/verify. Consider
deriving these values from `Instant.parse(...)` (or introducing a named
constant like `Instant.parse(\"2026-01-01T00:00:00Z\").toEpochMilli()`) so it’s
clear what timestamps the test is asserting.
##########
fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java:
##########
@@ -125,6 +125,31 @@ public void testOnceMarksLastBatchAndRecovers() throws
Exception {
}
}
+ @Test
+ public void testListingRefreshesFileSizeAndModificationTime() throws
Exception {
+ FileSystem fs = Mockito.mock(FileSystem.class);
+ try (MockedStatic<FileSystemFactory> factory =
Mockito.mockStatic(FileSystemFactory.class)) {
+ factory.when(() ->
FileSystemFactory.getFileSystem(Mockito.any(StorageAdapter.class))).thenReturn(fs);
+ for (S3SourceOffsetProvider provider : List.of(
+ new S3SourceOffsetProvider(), new
S3SourceOffsetProvider(ONCE_PROPS))) {
+ Mockito.when(fs.globListWithLimit(Mockito.any(Location.class),
Mockito.isNull(),
+ Mockito.anyLong(), Mockito.anyLong())).thenReturn(
+ new GlobListing(List.of(new
FileEntry(Location.of("s3://bucket/data/a.csv"),
+ 19, false, 1767225600000L, null)),
"bucket", "data/", "data/a.csv"),
+ new GlobListing(List.of(new
FileEntry(Location.of("s3://bucket/data/a.csv"),
+ 170, false, 1767225601000L, null)),
"bucket", "data/", "data/a.csv"));
+
+ TBrokerFileStatus first = provider.getNextOffset(ONCE_PROPS,
TVF_PROPS).getFileStatuses().get(0);
+ Assertions.assertEquals(19, first.getSize());
+ Assertions.assertEquals(1767225600000L,
first.getModificationTime());
Review Comment:
The epoch-millis literals make the test harder to read/verify. Consider
deriving these values from `Instant.parse(...)` (or introducing a named
constant like `Instant.parse(\"2026-01-01T00:00:00Z\").toEpochMilli()`) so it’s
clear what timestamps the test is asserting.
##########
fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java:
##########
@@ -125,6 +125,31 @@ public void testOnceMarksLastBatchAndRecovers() throws
Exception {
}
}
+ @Test
+ public void testListingRefreshesFileSizeAndModificationTime() throws
Exception {
+ FileSystem fs = Mockito.mock(FileSystem.class);
+ try (MockedStatic<FileSystemFactory> factory =
Mockito.mockStatic(FileSystemFactory.class)) {
+ factory.when(() ->
FileSystemFactory.getFileSystem(Mockito.any(StorageAdapter.class))).thenReturn(fs);
+ for (S3SourceOffsetProvider provider : List.of(
+ new S3SourceOffsetProvider(), new
S3SourceOffsetProvider(ONCE_PROPS))) {
+ Mockito.when(fs.globListWithLimit(Mockito.any(Location.class),
Mockito.isNull(),
+ Mockito.anyLong(), Mockito.anyLong())).thenReturn(
+ new GlobListing(List.of(new
FileEntry(Location.of("s3://bucket/data/a.csv"),
+ 19, false, 1767225600000L, null)),
"bucket", "data/", "data/a.csv"),
+ new GlobListing(List.of(new
FileEntry(Location.of("s3://bucket/data/a.csv"),
+ 170, false, 1767225601000L, null)),
"bucket", "data/", "data/a.csv"));
+
+ TBrokerFileStatus first = provider.getNextOffset(ONCE_PROPS,
TVF_PROPS).getFileStatuses().get(0);
+ Assertions.assertEquals(19, first.getSize());
+ Assertions.assertEquals(1767225600000L,
first.getModificationTime());
+ // An uncommitted batch must pick up an overwrite on the next
attempt.
+ TBrokerFileStatus retried = provider.getNextOffset(ONCE_PROPS,
TVF_PROPS).getFileStatuses().get(0);
+ Assertions.assertEquals(170, retried.getSize());
+ Assertions.assertEquals(1767225601000L,
retried.getModificationTime());
Review Comment:
The epoch-millis literals make the test harder to read/verify. Consider
deriving these values from `Instant.parse(...)` (or introducing a named
constant like `Instant.parse(\"2026-01-01T00:00:00Z\").toEpochMilli()`) so it’s
clear what timestamps the test is asserting.
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProvider.java:
##########
@@ -573,10 +578,22 @@ private ParsedMessage parseMessage(Message message) {
if (isCommitted(key, message.messageId(),
message.receiptHandle())) {
continue;
}
+ String path = "s3://" + bucket + "/" + key;
+ RemoteObject object;
+ try {
+ // Notifications can outlive an overwrite; refresh both scan
size and cache identity.
+ object = objectStorage.headObject(path);
+ } catch (FileNotFoundException e) {
+ log.info("Skip missing S3 EVENT object: jobId={},
messageId={}, path={}",
+ jobId, message.messageId(), path);
+ continue;
Review Comment:
Logging missing objects at INFO may be noisy in environments where S3 EVENT
messages can be delayed and objects may be short-lived or overwritten
frequently. Consider reducing this to DEBUG (or adding throttling / structured
counters) so production logs aren't flooded while still keeping the skip
behavior observable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
