This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 15eb3f4e432 [fix](fe) Fix compilation after dependency upgrades
(#69232)
15eb3f4e432 is described below
commit 15eb3f4e432b1d4a39f89457956135da80b90940
Author: zy-kkk <[email protected]>
AuthorDate: Sun Oct 18 22:10:54 2026 +0800
[fix](fe) Fix compilation after dependency upgrades (#69232)
### What problem does this PR solve?
Issue Number: N/A
Related PR: #68862
Problem Summary:
After #68862, FE production and test sources fail to compile:
- `S3EventSourceOffsetProvider` still imports `S3EventNotification` and
`Jackson` from AWS SDK v1 after the `aws-java-sdk-s3` dependency was
removed.
- `DorisHttpTestCase` still imports `ServletWebServerApplicationContext`
from its old package after the Spring Boot 4 upgrade.
Replace the AWS SDK v1 notification model with private DTOs parsed by
the existing Jackson dependency. Model only the fields consumed by
Doris, and retain the existing Joda time parser, scalar coercion,
single-pass UTF-8 URL decoding, trailing-JSON rejection and SQS
acknowledgement boundaries. This removes the remaining v1 imports
without restoring the removed dependency or adding a notification
library.
Unknown and unused notification metadata is ignored. A malformed unused
field such as `userIdentity` no longer discards a message whose consumed
fields are valid; malformed consumed fields retain their existing
handling.
Update the HTTP test import to
`org.springframework.boot.web.server.servlet.context.ServletWebServerApplicationContext`
for Spring Boot 4.
---
.../job/offset/s3/S3EventSourceOffsetProvider.java | 88 ++++++++++++++++------
.../org/apache/doris/http/DorisHttpTestCase.java | 2 +-
.../offset/s3/S3EventSourceOffsetProviderTest.java | 74 +++++++++++++++++-
3 files changed, 138 insertions(+), 26 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProvider.java
index 63023f7d986..c4b6f20d5a6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProvider.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProvider.java
@@ -33,16 +33,18 @@ import
org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableComma
import org.apache.doris.persist.gson.GsonUtils;
import org.apache.doris.thrift.TBrokerFileStatus;
-import com.amazonaws.services.s3.event.S3EventNotification;
-import
com.amazonaws.services.s3.event.S3EventNotification.S3EventNotificationRecord;
-import com.amazonaws.util.json.Jackson;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonSetter;
+import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Preconditions;
import com.google.gson.JsonObject;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang3.StringUtils;
+import org.joda.time.DateTime;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.BatchResultErrorEntry;
@@ -59,6 +61,7 @@ import
software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;
import java.io.IOException;
+import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
@@ -77,6 +80,9 @@ import java.util.stream.Collectors;
*/
@Log4j2
public class S3EventSourceOffsetProvider implements SourceOffsetProvider {
+ private static final ObjectMapper EVENT_MAPPER = new ObjectMapper()
+ .enable(JsonParser.Feature.ALLOW_COMMENTS)
+ .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
private static final int SQS_BATCH_SIZE = 10;
private static final int MIN_VISIBILITY_SECONDS = 120;
private static final int VISIBILITY_INTERVAL_MULTIPLIER = 6;
@@ -506,7 +512,7 @@ public class S3EventSourceOffsetProvider implements
SourceOffsetProvider {
private ParsedMessage parseMessage(Message message) {
S3EventNotification notification;
try {
- JsonNode json = Jackson.getObjectMapper().reader()
+ JsonNode json = EVENT_MAPPER.reader()
.with(DeserializationFeature.FAIL_ON_TRAILING_TOKENS).readTree(message.body());
if (json == null || !json.isObject()) {
log.warn("Skip invalid S3 EVENT message: jobId={},
messageId={}, reason=Expected a JSON object",
@@ -516,37 +522,35 @@ public class S3EventSourceOffsetProvider implements
SourceOffsetProvider {
if ("s3:TestEvent".equals(json.path("Event").asText())) {
return ParsedMessage.skipped(message.receiptHandle());
}
- notification = Jackson.getObjectMapper().treeToValue(json,
S3EventNotification.class);
+ notification = EVENT_MAPPER.treeToValue(json,
S3EventNotification.class);
} catch (JsonProcessingException e) {
// Only decoding failures are skipped; remote operation failures
must retain the message.
log.warn("Skip invalid S3 EVENT message: jobId={}, messageId={},
reason={}",
jobId, message.messageId(), e.getMessage());
return ParsedMessage.skipped(message.receiptHandle());
}
- if (notification.getRecords() == null ||
notification.getRecords().isEmpty()) {
+ if (notification.records == null || notification.records.isEmpty()) {
log.warn("Skip invalid S3 EVENT message: jobId={}, messageId={},
reason=No S3 event records",
jobId, message.messageId());
return ParsedMessage.skipped(message.receiptHandle());
}
- List<String> files = new ArrayList<>(notification.getRecords().size());
+ List<String> files = new ArrayList<>(notification.records.size());
List<TBrokerFileStatus> fileStatuses = new ArrayList<>();
long eventTimestampSeconds = 0;
- for (int i = 0; i < notification.getRecords().size(); i++) {
- S3EventNotificationRecord record =
notification.getRecords().get(i);
+ for (int i = 0; i < notification.records.size(); i++) {
+ S3EventNotificationRecord record = notification.records.get(i);
String skipReason = null;
if (record == null) {
skipReason = "Null S3 event record";
- } else if (!"aws:s3".equals(record.getEventSource())
- || !StringUtils.startsWith(record.getEventName(),
"ObjectCreated:")) {
+ } else if (!"aws:s3".equals(record.eventSource)
+ || !StringUtils.startsWith(record.eventName,
"ObjectCreated:")) {
skipReason = "Unsupported S3 event";
- } else if (record.getS3() == null || record.getS3().getBucket() ==
null
- || record.getS3().getObject() == null) {
+ } else if (record.s3 == null || record.s3.bucket == null ||
record.s3.object == null) {
skipReason = "Incomplete S3 event";
- } else if
(StringUtils.isBlank(record.getS3().getBucket().getName())
- ||
StringUtils.isBlank(record.getS3().getObject().getKey())) {
+ } else if (StringUtils.isBlank(record.s3.bucket.name)
+ || StringUtils.isBlank(record.s3.object.key)) {
skipReason = "Missing S3 bucket or object key";
- } else if (record.getS3().getObject().getSizeAsLong() == null
- || record.getS3().getObject().getSizeAsLong() < 0) {
+ } else if (record.s3.object.size == null || record.s3.object.size
< 0) {
skipReason = "Invalid S3 object size";
}
if (skipReason != null) {
@@ -554,10 +558,10 @@ public class S3EventSourceOffsetProvider implements
SourceOffsetProvider {
jobId, message.messageId(), i, skipReason);
continue;
}
- String bucket = record.getS3().getBucket().getName();
+ String bucket = record.s3.bucket.name;
String key;
try {
- key = record.getS3().getObject().getUrlDecodedKey();
+ key = URLDecoder.decode(record.s3.object.key,
StandardCharsets.UTF_8);
if (!sourceBucket.equals(bucket) ||
!sourceKeyMatcher.matcher(key).matches()) {
continue;
}
@@ -572,9 +576,9 @@ public class S3EventSourceOffsetProvider implements
SourceOffsetProvider {
files.add(key);
// The object must remain unchanged until ingestion completes;
notifications identify keys, not versions.
fileStatuses.add(new TBrokerFileStatus("s3://" + bucket + "/" +
key, false,
- record.getS3().getObject().getSizeAsLong(), true));
- if (record.getEventTime() != null) {
- eventTimestampSeconds = Math.max(eventTimestampSeconds,
record.getEventTime().getMillis() / 1000);
+ record.s3.object.size, true));
+ if (record.eventTime != null) {
+ eventTimestampSeconds = Math.max(eventTimestampSeconds,
record.eventTime.getMillis() / 1000);
}
}
if (files.isEmpty()) {
@@ -853,6 +857,46 @@ public class S3EventSourceOffsetProvider implements
SourceOffsetProvider {
}
}
+ // Keep Jackson's scalar coercion and the original ISO date parser without
depending on AWS SDK v1.
+ private static class S3EventNotification {
+ @JsonProperty("Records")
+ private List<S3EventNotificationRecord> records;
+ }
+
+ private static class S3EventNotificationRecord {
+ @JsonProperty
+ private String eventSource;
+ @JsonProperty
+ private String eventName;
+ @JsonProperty
+ private S3Entity s3;
+ private DateTime eventTime;
+
+ @JsonSetter("eventTime")
+ private void setEventTime(String value) {
+ eventTime = value == null ? null : DateTime.parse(value);
+ }
+ }
+
+ private static class S3Entity {
+ @JsonProperty
+ private S3BucketEntity bucket;
+ @JsonProperty
+ private S3ObjectEntity object;
+ }
+
+ private static class S3BucketEntity {
+ @JsonProperty
+ private String name;
+ }
+
+ private static class S3ObjectEntity {
+ @JsonProperty
+ private String key;
+ @JsonProperty
+ private Long size;
+ }
+
private static class ParsedMessage {
private final String messageId;
private final String receiptHandle;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/http/DorisHttpTestCase.java
b/fe/fe-core/src/test/java/org/apache/doris/http/DorisHttpTestCase.java
index 4c06874d8d0..0d9337cdfd2 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/http/DorisHttpTestCase.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/http/DorisHttpTestCase.java
@@ -67,7 +67,7 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.mockito.Mockito;
-import
org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext;
+import
org.springframework.boot.web.server.servlet.context.ServletWebServerApplicationContext;
import java.io.File;
import java.io.IOException;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProviderTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProviderTest.java
index ec134b6b696..0c095d0678f 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProviderTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProviderTest.java
@@ -464,7 +464,7 @@ public class S3EventSourceOffsetProviderTest {
String noTimestamp = record("bucket", "logs/a.csv", 1)
.replace("\"eventTime\":\"2026-01-01T00:00:00.000Z\",", "");
Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
- .thenReturn(response(message("missing-time", noTimestamp)));
+ .thenReturn(response(message("missing-time", noTimestamp,
record("bucket", "logs/null.csv", 1, null))));
provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
provider.updateOffset(nextOffset());
provider.onTaskCommitted(1, 1);
@@ -599,7 +599,8 @@ public class S3EventSourceOffsetProviderTest {
String validEvent = message("invalid", record("bucket", "logs/a.csv",
1)).body();
String unsupportedEvent = validEvent.replace("ObjectCreated:Put",
"ObjectRemoved:Delete");
for (String body : Arrays.asList("", " ", "not-json", "null", "[]",
"{}", "{\"Records\":\"invalid\"}",
- "{\"Records\":[null]}", "{\"Event\":\"s3:TestEvent\"}",
+ "{\"Records\":null}", "{\"Records\":[]}",
"{\"Records\":[null]}", "{\"Records\":[42]}",
+ "{\"Event\":\"s3:TestEvent\"}",
validEvent + " {}", validEvent + " garbage",
unsupportedEvent)) {
Mockito.clearInvocations(sqsClient);
Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
@@ -610,12 +611,79 @@ public class S3EventSourceOffsetProviderTest {
Assertions.assertFalse(provider.hasMoreDataToConsume());
Assertions.assertNull(provider.getPersistInfo());
- ArgumentCaptor<DeleteMessageBatchRequest> deleted =
ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ ArgumentCaptor<DeleteMessageBatchRequest> deleted =
+ ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
Mockito.verify(sqsClient).deleteMessageBatch(deleted.capture());
Assertions.assertEquals(Collections.singletonList("invalid"),
deleteHandles(deleted.getValue()));
}
}
+ @Test
+ public void testEventDecodingPreservesSizeTimeAndUnknownFields() {
+ long size = (long) Integer.MAX_VALUE + 1;
+ String eventTime = "2026-01-01T08:00:00.987+08:00";
+ String validRecord = record("bucket",
"logs/a+b%2Bc%252F%E4%B8%AD.csv", size, eventTime)
+ .replace("\"size\":" + size, "\"size\":\"" + size + "\"")
+ .replace("\"s3\":{", "\"s3\":{\"futureField\":true,")
+ .replace("\"object\":{", "\"object\":{\"futureField\":[],");
+ String body = message("compatible", validRecord).body()
+ .replace("{\"Records\":", "{/* comment
*/\"futureField\":{},\"Records\":");
+
Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+
.thenReturn(response(Message.builder().messageId("compatible").receiptHandle("compatible")
+ .body(body).build()),
ReceiveMessageResponse.builder().build());
+
+ provider.fetchRemoteMeta(jobProperties(2), tvfProperties);
+
+ S3EventOffset offset = nextOffset();
+ Assertions.assertEquals(Collections.singletonList("logs/a
b+c%2F中.csv"), offset.getFiles());
+ Assertions.assertEquals(size,
offset.getFileStatuses().get(0).getSize());
+ provider.updateOffset(offset);
+ provider.onTaskCommitted(1, size);
+ Assertions.assertEquals(Instant.parse(eventTime).toEpochMilli() / 1000,
+ provider.getLastSourceEventTimestampSeconds());
+ }
+
+ @Test
+ public void testUnusedNotificationMetadataDoesNotPreventIngestion() {
+ String validRecord = record("bucket", "logs/a.csv", 1)
+ .replace("\"userIdentity\":{\"principalId\":\"test\"}",
"\"userIdentity\":42")
+ .replace("\"size\":1", "\"size\":1.9");
+
Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(message("unused-metadata", validRecord)),
+ ReceiveMessageResponse.builder().build());
+
+ provider.fetchRemoteMeta(jobProperties(2), tvfProperties);
+
+ S3EventOffset offset = nextOffset();
+ Assertions.assertEquals(Collections.singletonList("logs/a.csv"),
offset.getFiles());
+ Assertions.assertEquals(1, offset.getFileStatuses().get(0).getSize());
+ Mockito.verify(sqsClient,
Mockito.never()).deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class));
+ }
+
+ @Test
+ public void testMalformedConsumedFieldsDiscardWholeMessage() {
+ String valid = record("bucket", "logs/valid.csv", 1);
+ for (String malformed : Arrays.asList(record("bucket", "logs/bad.csv",
1, "invalid-time"),
+ valid.replace("\"size\":1", "\"size\":\"invalid\""),
+ valid.replace("\"size\":1", "\"size\":9223372036854775808"),
+ valid.replace("\"key\":\"logs/valid.csv\"", "\"key\":{}"))) {
+ Mockito.clearInvocations(sqsClient);
+
Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(message("malformed", valid,
malformed)),
+ ReceiveMessageResponse.builder().build());
+
+ Assertions.assertDoesNotThrow(() ->
provider.fetchRemoteMeta(jobProperties(2), tvfProperties));
+
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+ ArgumentCaptor<DeleteMessageBatchRequest> deleted =
+ ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ Mockito.verify(sqsClient).deleteMessageBatch(deleted.capture());
+ Assertions.assertEquals(Collections.singletonList("malformed"),
deleteHandles(deleted.getValue()));
+ Mockito.verify(sqsClient,
Mockito.never()).changeMessageVisibilityBatch(
+ Mockito.any(ChangeMessageVisibilityBatchRequest.class));
+ }
+ }
+
@Test
public void
testInvalidRecordsDoNotDiscardValidFilesAndDeletionWaitsForCommit() {
Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
