You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2026/10/14 06:45:16 UTC
(doris) branch branch-4.1 updated: branch-4.1: [feature](fe) Support S3 notification-based streaming ingestion #68876 (#68992)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new bea4d0d5dbd branch-4.1: [feature](fe) Support S3 notification-based streaming ingestion #68876 (#68992)
bea4d0d5dbd is described below
commit bea4d0d5dbd61dedd72886d23d4c2121c21fd13c
Author: wudi <wu...@selectdb.com>
AuthorDate: Wed Oct 14 14:45:06 2026 +0800
branch-4.1: [feature](fe) Support S3 notification-based streaming ingestion #68876 (#68992)
Cherry-picked from #68876
---
fe/fe-core/pom.xml | 4 +
.../java/org/apache/doris/common/util/S3Util.java | 39 +-
.../insert/streaming/StreamingInsertJob.java | 14 +-
.../insert/streaming/StreamingInsertTask.java | 19 +-
.../insert/streaming/StreamingJobProperties.java | 56 +-
.../doris/job/offset/SourceOffsetProvider.java | 15 +-
.../job/offset/SourceOffsetProviderFactory.java | 6 +-
.../s3/{S3Offset.java => S3EventOffset.java} | 58 +-
.../job/offset/s3/S3EventSourceOffsetProvider.java | 879 +++++++++++++++
.../org/apache/doris/job/offset/s3/S3Offset.java | 8 +-
.../job/offset/s3/S3SourceOffsetProvider.java | 91 +-
.../java/org/apache/doris/metric/MetricRepo.java | 19 +-
.../trees/expressions/functions/table/S3.java | 30 +-
.../trees/plans/commands/AlterJobCommand.java | 8 +-
.../commands/insert/InsertIntoTableCommand.java | 10 +
.../ExternalFileTableValuedFunction.java | 8 +
.../doris/tablefunction/S3TableValuedFunction.java | 11 +-
.../org/apache/doris/common/util/S3UtilTest.java | 55 +-
.../streaming/StreamingInsertJobLagTest.java | 15 +
.../StreamingInsertJobOffsetPersistenceTest.java | 4 +-
.../StreamingInsertJobStatusTransitionTest.java | 66 ++
.../streaming/StreamingInsertTaskAuditTest.java | 137 ++-
.../streaming/StreamingJobPropertiesTest.java | 65 +-
.../doris/job/offset/s3/S3EventOffsetTest.java | 56 +
.../offset/s3/S3EventSourceOffsetProviderTest.java | 1185 ++++++++++++++++++++
.../job/offset/s3/S3SourceOffsetProviderTest.java | 67 +-
.../java/org/apache/doris/metric/MetricsTest.java | 29 +
.../tablefunction/S3TableValuedFunctionTest.java | 147 +++
.../test_streaming_insert_job_s3_event_aksk.out | 20 +
...test_streaming_insert_job_s3_event_iam_role.out | 7 +
...st_streaming_insert_job_s3_event_restart_fe.out | 14 +
regression-test/framework/pom.xml | 4 +
.../test_streaming_insert_job_s3_event_aksk.groovy | 217 ++++
...t_streaming_insert_job_s3_event_iam_role.groovy | 128 +++
...streaming_insert_job_s3_event_restart_fe.groovy | 152 +++
.../test_streaming_insert_job_s3_once.groovy | 6 +-
36 files changed, 3508 insertions(+), 141 deletions(-)
diff --git a/fe/fe-core/pom.xml b/fe/fe-core/pom.xml
index 848a731d4ba..2b928de3bde 100644
--- a/fe/fe-core/pom.xml
+++ b/fe/fe-core/pom.xml
@@ -632,6 +632,10 @@ under the License.
<groupId>software.amazon.awssdk</groupId>
<artifactId>s3-transfer-manager</artifactId>
</dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>sqs</artifactId>
+ </dependency>
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>sts</artifactId>
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/util/S3Util.java b/fe/fe-core/src/main/java/org/apache/doris/common/util/S3Util.java
index f63f0be3c7a..7b2d213697d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/util/S3Util.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/util/S3Util.java
@@ -577,7 +577,7 @@ public class S3Util {
/**
* Convert range patterns to brace enumeration patterns for glob matching.
- * Parts containing negative numbers or non-numeric characters are skipped.
+ * Literal alternatives are preserved; ranges containing negative bounds are skipped.
* eg(valid):
* -> "file{1..3}" => "file{1,2,3}"
* -> "file_{1..3,4,5..6}" => "file_{1,2,3,4,5,6}"
@@ -594,13 +594,12 @@ public class S3Util {
while (braceMatcher.find()) {
String braceContent = braceMatcher.group(1);
- String[] parts = braceContent.split(",");
- List<Integer> allNumbers = new ArrayList<>();
+ String[] parts = braceContent.split(",", -1);
+ List<String> values = new ArrayList<>();
Pattern rangePattern = Pattern.compile("^(-?\\d+)\\.\\.(-?\\d+)$");
for (String part : parts) {
- part = part.trim();
- Matcher rangeMatcher = rangePattern.matcher(part);
+ Matcher rangeMatcher = rangePattern.matcher(part.trim());
if (rangeMatcher.matches()) {
int start = Integer.parseInt(rangeMatcher.group(1));
@@ -617,39 +616,35 @@ public class S3Util {
end = temp;
}
for (int i = start; i <= end; i++) {
- if (!allNumbers.contains(i)) {
- allNumbers.add(i);
+ String value = String.valueOf(i);
+ if (!values.contains(value)) {
+ values.add(value);
}
}
- } else if (part.matches("^\\d+$")) {
- // This is a single non-negative number like "4"
- int num = Integer.parseInt(part);
- if (!allNumbers.contains(num)) {
- allNumbers.add(num);
- }
} else {
- // Not a valid number or range (e.g., negative number, or contains non-numeric chars)
- // Just skip this part and continue processing other parts
- continue;
+ // Preserve literal file names, including leading zeros and non-numeric characters.
+ if (!values.contains(part)) {
+ values.add(part);
+ }
}
}
- // If no valid numbers found after filtering, keep original content
- if (allNumbers.isEmpty()) {
- braceMatcher.appendReplacement(result, "{" + braceContent + "}");
+ // If all alternatives were skipped, keep the original content.
+ if (values.isEmpty()) {
+ braceMatcher.appendReplacement(result, Matcher.quoteReplacement("{" + braceContent + "}"));
continue;
}
// Build comma-separated result
StringBuilder sb = new StringBuilder("{");
- for (int i = 0; i < allNumbers.size(); i++) {
+ for (int i = 0; i < values.size(); i++) {
if (i > 0) {
sb.append(",");
}
- sb.append(allNumbers.get(i));
+ sb.append(values.get(i));
}
sb.append("}");
- braceMatcher.appendReplacement(result, sb.toString());
+ braceMatcher.appendReplacement(result, Matcher.quoteReplacement(sb.toString()));
}
braceMatcher.appendTail(result);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
index 972be838e19..938e6b99a46 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
@@ -563,6 +563,7 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M
clearRunningStreamTask(status);
}
if (isFinalStatus()) {
+ closeOffsetProvider();
Env.getCurrentGlobalTransactionMgr().getCallbackFactory().removeCallback(getJobId());
}
log.info("Streaming insert job {} update status to {}", getJobId(), getJobStatus());
@@ -777,7 +778,7 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M
// when fe restart, offsetProvider.jobId may be null
Map<String, String> props = getProviderProps();
offsetProvider.ensureInitialized(getJobId(), props);
- offsetProvider.fetchRemoteMeta(props);
+ offsetProvider.fetchRemoteMeta(jobProperties, props);
} catch (Exception ex) {
log.warn("fetch remote meta failed, job id: {}", getJobId(), ex);
if (tryPauseJob(new FailureReason(InternalErrorCode.GET_REMOTE_DATA_ERROR,
@@ -1036,6 +1037,10 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M
return offsetProvider != null ? offsetProvider.getLastSourceEventTimestampSeconds() : 0;
}
+ public long getLagMessages() {
+ return offsetProvider != null ? offsetProvider.getLagMessages() : -1;
+ }
+
public long getLastTaskSuccessTimeSeconds() {
return lastTaskSuccessTime / 1000L;
}
@@ -1704,12 +1709,19 @@ public class StreamingInsertJob extends AbstractJob<StreamingJobSchedulerTask, M
return offsetProvider != null && offsetProvider.hasReachedEnd();
}
+ private void closeOffsetProvider() {
+ if (offsetProvider != null) {
+ offsetProvider.close();
+ }
+ }
+
/**
* 1. Clean offset info in ms (s3 tvf)
* 2. Clean chunk info in meta table (jdbc)
*/
public void cleanup() throws JobException {
log.info("cleanup streaming job {}", getJobId());
+ closeOffsetProvider();
// s3 tvf clean offset
if (tvfType != null && Config.isCloudMode()) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java
index f6589d730c2..ec923146c93 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java
@@ -32,11 +32,11 @@ import org.apache.doris.job.extensions.insert.InsertTask;
import org.apache.doris.job.offset.SourceOffsetProvider;
import org.apache.doris.load.loadv2.LoadJob;
import org.apache.doris.nereids.StatementContext;
-import org.apache.doris.nereids.analyzer.UnboundTVFRelation;
import org.apache.doris.nereids.glue.LogicalPlanAdapter;
import org.apache.doris.nereids.parser.NereidsParser;
import org.apache.doris.nereids.trees.plans.commands.info.BaseViewInfo;
import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand;
+import org.apache.doris.nereids.trees.plans.logical.LogicalTVFRelation;
import org.apache.doris.nereids.util.SqlLiteralUtils;
import org.apache.doris.qe.AuditLogHelper;
import org.apache.doris.qe.ConnectContext;
@@ -118,11 +118,13 @@ public class StreamingInsertTask extends AbstractStreamingTask {
log.info("streaming insert task {} get running offset: {}", taskId, runningOffset.toString());
InsertIntoTableCommand baseCommand = (InsertIntoTableCommand) new NereidsParser().parseSingle(sql);
baseCommand.setJobId(getTaskId());
- StmtExecutor baseStmtExecutor =
- new StmtExecutor(ctx, new LogicalPlanAdapter(baseCommand, ctx.getStatementContext()));
- baseCommand.initPlan(ctx, baseStmtExecutor, false);
- if (!baseCommand.getParsedPlan().isPresent()) {
- throw new JobException("Can not get Parsed plan");
+ if (!S3TableValuedFunction.NAME.equalsIgnoreCase(offsetProvider.getSourceType())) {
+ StmtExecutor baseStmtExecutor =
+ new StmtExecutor(ctx, new LogicalPlanAdapter(baseCommand, ctx.getStatementContext()));
+ baseCommand.initPlan(ctx, baseStmtExecutor, false);
+ if (!baseCommand.getParsedPlan().isPresent()) {
+ throw new JobException("Can not get Parsed plan");
+ }
}
this.taskCommand = offsetProvider.rewriteTvfParams(baseCommand, runningOffset, getTaskId());
this.taskCommand.setLabelName(Optional.of(labelName));
@@ -175,10 +177,11 @@ public class StreamingInsertTask extends AbstractStreamingTask {
private String buildAuditSql() {
TreeMap<Pair<Integer, Integer>, String> replacements = new TreeMap<>(new Pair.PairComparator<>());
new NereidsParser().parseForEncryption(sql, replacements);
- List<UnboundTVFRelation> tvfRelations = taskCommand.getAllTVFRelation();
+ List<LogicalTVFRelation> tvfRelations = taskCommand.getLogicalQuery()
+ .collectToList(LogicalTVFRelation.class::isInstance);
Preconditions.checkState(replacements.size() == 1 && tvfRelations.size() == 1,
"S3 streaming insert must contain exactly one TVF");
- String rewrittenProperties = tvfRelations.get(0).getProperties().getMap().entrySet().stream()
+ String rewrittenProperties = tvfRelations.get(0).getFunction().getTVFProperties().getMap().entrySet().stream()
.map(entry -> SqlLiteralUtils.quoteStringLiteral(entry.getKey()) + " = "
+ SqlLiteralUtils.quoteStringLiteral(entry.getValue()))
.collect(Collectors.joining(", "));
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java
index d0ea8ec6055..dba78d506c2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobProperties.java
@@ -26,6 +26,7 @@ import org.apache.doris.qe.VariableMgr;
import com.google.common.base.Strings;
import lombok.Data;
+import org.apache.commons.lang3.StringUtils;
import org.json.simple.JSONObject;
import java.util.ArrayList;
@@ -43,15 +44,19 @@ public class StreamingJobProperties implements JobProperties {
public static final String S3_MAX_BATCH_FILES_PROPERTY = "s3.max_batch_files";
public static final String S3_MAX_BATCH_BYTES_PROPERTY = "s3.max_batch_bytes";
public static final String S3_INGESTION_MODE_PROPERTY = "s3.ingestion_mode";
- public static final String S3_INGESTION_MODE_LEXICAL = "LEXICAL";
- public static final String S3_INGESTION_MODE_ONCE = "ONCE";
+ public static final String S3_EVENT_SOURCE_PROPERTY = "s3.event.source";
+ public static final String S3_SQS_QUEUE_URL_PROPERTY = "s3.sqs.queue_url";
+ public static final String S3_INGESTION_MODE_ORDERED_LIST = "ORDERED_LIST";
+ public static final String S3_INGESTION_MODE_NOTIFICATION = "NOTIFICATION";
+ public static final String S3_INGESTION_MODE_ONE_TIME = "ONE_TIME";
+ public static final String S3_EVENT_SOURCE_SQS = "SQS";
public static final String SESSION_VAR_PREFIX = "session.";
public static final String INTERNAL_KEY_PREFIX = "__";
public static final String OFFSET_PROPERTY = "offset";
public static final String COMPUTE_GROUP_PROPERTY = "compute_group";
public static final List<String> SUPPORT_STREAM_JOB_PROPS = Arrays.asList(MAX_INTERVAL_SECOND_PROPERTY,
S3_MAX_BATCH_FILES_PROPERTY, S3_MAX_BATCH_BYTES_PROPERTY, S3_INGESTION_MODE_PROPERTY,
- OFFSET_PROPERTY, COMPUTE_GROUP_PROPERTY);
+ S3_EVENT_SOURCE_PROPERTY, S3_SQS_QUEUE_URL_PROPERTY, OFFSET_PROPERTY, COMPUTE_GROUP_PROPERTY);
public static final long DEFAULT_MAX_INTERVAL_SECOND = 10;
public static final long DEFAULT_MAX_S3_BATCH_FILES = 256;
@@ -129,12 +134,31 @@ public class StreamingJobProperties implements JobProperties {
+ properties.get(StreamingJobProperties.S3_MAX_BATCH_BYTES_PROPERTY));
String ingestionMode = getS3IngestionMode();
- if (!S3_INGESTION_MODE_LEXICAL.equals(ingestionMode)
- && !S3_INGESTION_MODE_ONCE.equals(ingestionMode)) {
+ if (!S3_INGESTION_MODE_ORDERED_LIST.equals(ingestionMode)
+ && !S3_INGESTION_MODE_ONE_TIME.equals(ingestionMode)
+ && !S3_INGESTION_MODE_NOTIFICATION.equals(ingestionMode)) {
throw new AnalysisException("Unsupported s3.ingestion_mode: " + ingestionMode);
}
- if (S3_INGESTION_MODE_ONCE.equals(ingestionMode) && properties.containsKey(OFFSET_PROPERTY)) {
- throw new AnalysisException("offset is not supported when s3.ingestion_mode is ONCE");
+ if (S3_INGESTION_MODE_ONE_TIME.equals(ingestionMode) && properties.containsKey(OFFSET_PROPERTY)) {
+ throw new AnalysisException("offset is not supported when s3.ingestion_mode is ONE_TIME");
+ }
+
+ if (S3_INGESTION_MODE_NOTIFICATION.equals(ingestionMode)) {
+ if (!S3_EVENT_SOURCE_SQS.equalsIgnoreCase(getS3EventSource())) {
+ throw new AnalysisException("s3.event.source must be SQS when s3.ingestion_mode is NOTIFICATION");
+ }
+ if (StringUtils.isBlank(getS3SqsQueueUrl())) {
+ throw new AnalysisException("s3.sqs.queue_url is required when s3.ingestion_mode is NOTIFICATION");
+ }
+ if (properties.containsKey(OFFSET_PROPERTY)) {
+ throw new AnalysisException("offset is not supported when s3.ingestion_mode is NOTIFICATION");
+ }
+ if (maxIntervalSecond >= 7200) {
+ throw new AnalysisException("max_interval must be less than 7200 seconds for NOTIFICATION mode");
+ }
+ } else if (properties.containsKey(S3_EVENT_SOURCE_PROPERTY)
+ || properties.containsKey(S3_SQS_QUEUE_URL_PROPERTY)) {
+ throw new AnalysisException("s3.event.source and s3.sqs.queue_url require s3.ingestion_mode=NOTIFICATION");
}
// validate session variables
@@ -220,11 +244,23 @@ public class StreamingJobProperties implements JobProperties {
}
public String getS3IngestionMode() {
- return properties.getOrDefault(S3_INGESTION_MODE_PROPERTY, S3_INGESTION_MODE_LEXICAL)
+ return properties.getOrDefault(S3_INGESTION_MODE_PROPERTY, S3_INGESTION_MODE_ORDERED_LIST)
.trim().toUpperCase(Locale.ROOT);
}
- public boolean isS3OnceMode() {
- return S3_INGESTION_MODE_ONCE.equals(getS3IngestionMode());
+ public boolean isS3OneTimeMode() {
+ return S3_INGESTION_MODE_ONE_TIME.equals(getS3IngestionMode());
+ }
+
+ public boolean isS3NotificationMode() {
+ return S3_INGESTION_MODE_NOTIFICATION.equals(getS3IngestionMode());
+ }
+
+ public String getS3EventSource() {
+ return properties.get(S3_EVENT_SOURCE_PROPERTY);
+ }
+
+ public String getS3SqsQueueUrl() {
+ return properties.get(S3_SQS_QUEUE_URL_PROPERTY);
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java b/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java
index be70fdf83bb..e2c70b228ae 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java
@@ -44,6 +44,9 @@ public interface SourceOffsetProvider {
*/
default void ensureInitialized(Long jobId, Map<String, String> originTvfProps) throws JobException {}
+ /** Permanently close runtime resources and prevent subsequent metadata requests. */
+ default void close() {}
+
/**
* One-time initialization on fresh job creation (not on FE restart). Subclasses may
* initialize split progress, fetch initial splits, or open remote readers.
@@ -102,6 +105,11 @@ public interface SourceOffsetProvider {
*/
void fetchRemoteMeta(Map<String, String> properties) throws Exception;
+ default void fetchRemoteMeta(StreamingJobProperties jobProperties, Map<String, String> properties)
+ throws Exception {
+ fetchRemoteMeta(properties);
+ }
+
/**
* Whether there is more data to consume
*
@@ -211,7 +219,12 @@ public interface SourceOffsetProvider {
return -1;
}
- /** Get the source event timestamp at the committed offset as Unix seconds, or 0 if unavailable. */
+ /** Get the observed message backlog, or -1 if unavailable or not applicable. */
+ default long getLagMessages() {
+ return -1;
+ }
+
+ /** Get the source event timestamp of the latest committed batch as Unix seconds, or 0 if unavailable. */
default long getLastSourceEventTimestampSeconds() {
return 0;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProviderFactory.java b/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProviderFactory.java
index 631162d27fe..442ed187ba8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProviderFactory.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProviderFactory.java
@@ -20,6 +20,7 @@ package org.apache.doris.job.offset;
import org.apache.doris.job.exception.JobException;
import org.apache.doris.job.extensions.insert.streaming.StreamingJobProperties;
import org.apache.doris.job.offset.jdbc.JdbcTvfSourceOffsetProvider;
+import org.apache.doris.job.offset.s3.S3EventSourceOffsetProvider;
import org.apache.doris.job.offset.s3.S3SourceOffsetProvider;
import lombok.extern.log4j.Log4j2;
@@ -39,7 +40,10 @@ public class SourceOffsetProviderFactory {
public static SourceOffsetProvider createSourceOffsetProvider(
String sourceType, StreamingJobProperties jobProperties) {
try {
- if ("s3".equalsIgnoreCase(sourceType) && jobProperties.isS3OnceMode()) {
+ if ("s3".equalsIgnoreCase(sourceType) && jobProperties.isS3NotificationMode()) {
+ return new S3EventSourceOffsetProvider(jobProperties.getS3SqsQueueUrl());
+ }
+ if ("s3".equalsIgnoreCase(sourceType) && jobProperties.isS3OneTimeMode()) {
return new S3SourceOffsetProvider(jobProperties);
}
Class<? extends SourceOffsetProvider> cla = map.get(sourceType.toLowerCase());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3Offset.java b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventOffset.java
similarity index 51%
copy from fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3Offset.java
copy to fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventOffset.java
index 44df1756e02..47cad957499 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3Offset.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventOffset.java
@@ -19,23 +19,46 @@ package org.apache.doris.job.offset.s3;
import org.apache.doris.job.offset.Offset;
import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.thrift.TBrokerFileStatus;
import com.google.gson.annotations.SerializedName;
-import lombok.Getter;
-import lombok.Setter;
import org.apache.commons.lang3.StringUtils;
-@Getter
-@Setter
-public class S3Offset implements Offset {
- String startFile;
- @SerializedName("endFile")
- String endFile;
- // s3://bucket/path/{1.csv,2.csv}
- String fileLists;
- int fileNum;
- @SerializedName("lastBatch")
- boolean lastBatch;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class S3EventOffset implements Offset {
+ static final int TARGET_BATCH_SERIALIZED_BYTES = 64 * 1024;
+
+ @SerializedName("files")
+ private List<String> files;
+ private transient List<TBrokerFileStatus> fileStatuses;
+
+ public S3EventOffset() {
+ }
+
+ public S3EventOffset(List<String> files) {
+ this.files = new ArrayList<>(files);
+ }
+
+ public S3EventOffset(List<String> files, List<TBrokerFileStatus> fileStatuses) {
+ this(files);
+ this.fileStatuses = Collections.unmodifiableList(new ArrayList<>(fileStatuses));
+ }
+
+ public List<TBrokerFileStatus> getFileStatuses() {
+ return fileStatuses;
+ }
+
+ public List<String> getFiles() {
+ return files == null ? Collections.emptyList() : Collections.unmodifiableList(files);
+ }
+
+ public int serializedSize() {
+ return toSerializedJson().getBytes(StandardCharsets.UTF_8).length;
+ }
@Override
public String toSerializedJson() {
@@ -44,22 +67,21 @@ public class S3Offset implements Offset {
@Override
public boolean isEmpty() {
- return StringUtils.isEmpty(fileLists);
+ return files == null || files.isEmpty();
}
@Override
public boolean isValidOffset() {
- return StringUtils.isNotBlank(endFile);
+ return !isEmpty() && files.stream().noneMatch(StringUtils::isBlank);
}
@Override
public String showRange() {
- return "{\"startFileName\":\"" + startFile + "\",\"endFileName\":\"" + endFile + "\"}";
+ return toSerializedJson();
}
@Override
public String toString() {
- return "{\"startFileName\":\"" + startFile + "\","
- + "\"endFileName\":\"" + endFile + "\",\"fileNum\":" + fileNum + "}";
+ return toSerializedJson();
}
}
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
new file mode 100644
index 00000000000..1a949a9eb59
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProvider.java
@@ -0,0 +1,879 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.offset.s3;
+
+import org.apache.doris.common.util.S3Util;
+import org.apache.doris.datasource.property.storage.S3Properties;
+import org.apache.doris.datasource.property.storage.StorageProperties;
+import org.apache.doris.fs.FileSystemFactory;
+import org.apache.doris.fs.obj.S3ObjStorage;
+import org.apache.doris.fs.remote.ObjFileSystem;
+import org.apache.doris.fs.remote.RemoteFileSystem;
+import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob;
+import org.apache.doris.job.extensions.insert.streaming.StreamingJobProperties;
+import org.apache.doris.job.offset.Offset;
+import org.apache.doris.job.offset.SourceOffsetProvider;
+import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand;
+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.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.base.Preconditions;
+import com.google.gson.JsonObject;
+import lombok.extern.log4j.Log4j2;
+import org.apache.commons.lang3.StringUtils;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.sqs.SqsClient;
+import software.amazon.awssdk.services.sqs.model.BatchResultErrorEntry;
+import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityBatchRequest;
+import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityBatchRequestEntry;
+import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityBatchResponse;
+import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequest;
+import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequestEntry;
+import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchResponse;
+import software.amazon.awssdk.services.sqs.model.GetQueueAttributesRequest;
+import software.amazon.awssdk.services.sqs.model.Message;
+import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
+import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
+import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileSystems;
+import java.nio.file.PathMatcher;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/**
+ * Consumes S3 object-created notifications from SQS for streaming INSERT jobs.
+ */
+@Log4j2
+public class S3EventSourceOffsetProvider implements SourceOffsetProvider {
+ private static final int SQS_BATCH_SIZE = 10;
+ private static final int MIN_VISIBILITY_SECONDS = 120;
+ private static final int VISIBILITY_INTERVAL_MULTIPLIER = 6;
+
+ private final String queueUrl;
+ private transient Long jobId;
+ private transient String sourceBucket;
+ private transient PathMatcher sourceKeyMatcher;
+ // Only one task batch is retained, either ready for scheduling or assigned to a task.
+ private transient TaskBatch readyBatch;
+ private transient TaskBatch runningBatch;
+ // Includes both committed task messages and skipped messages whose deletion must be retried.
+ private transient List<String> receiptsToDelete = new ArrayList<>();
+ private transient S3EventOffset currentOffset;
+ // File -> message ID -> receipt; an expired receipt stays null until that message is received again.
+ private transient Map<String, Map<String, String>> unackedCommittedFiles;
+ // Display-only statistic; unavailable after restart until another batch commits.
+ private transient long lastSourceEventTimestampSeconds;
+ // Approximate visible backlog only; null means the queue statistic is unavailable.
+ private transient Long approximateNumberOfMessages;
+ private transient volatile boolean closed;
+ // S3 and SQS share a refreshable credentials provider, which must stay open until SQS closes.
+ private transient RemoteFileSystem fileSystem;
+ private transient SqsClient sqsClient;
+ private transient Map<String, String> clientProperties;
+
+ public S3EventSourceOffsetProvider(String queueUrl) {
+ this.queueUrl = queueUrl;
+ }
+
+ @Override
+ public String getSourceType() {
+ return "s3";
+ }
+
+ @Override
+ public synchronized void ensureInitialized(Long jobId, Map<String, String> originTvfProps) throws JobException {
+ try {
+ this.jobId = jobId;
+ Map<String, String> copiedProps = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ copiedProps.putAll(originTvfProps);
+ StorageProperties storageProperties = StorageProperties.createPrimary(copiedProps);
+ if (!(storageProperties instanceof S3Properties)) {
+ throw new JobException("S3 NOTIFICATION mode currently supports AWS S3 only");
+ }
+ S3Properties s3Properties = (S3Properties) storageProperties;
+ String uri = storageProperties.validateAndGetUri(copiedProps);
+ String normalizedUri = storageProperties.validateAndNormalizeUri(uri);
+ // Notification keys preserve literal '?' and '#'; URI query/fragment parsing would truncate them.
+ String[] bucketAndKey = normalizedUri.substring("s3://".length()).split("/", 2);
+ sourceBucket = bucketAndKey[0];
+ if (sourceKeyMatcher == null) {
+ String key = bucketAndKey.length == 1 ? "" : bucketAndKey[1].replaceFirst("^/+", "");
+ sourceKeyMatcher = FileSystems.getDefault()
+ .getPathMatcher("glob:" + Paths.get(S3Util.extendGlobs(key)));
+ }
+
+ if (StringUtils.isBlank(s3Properties.getRegion())) {
+ throw new JobException("s3.region is required for S3 NOTIFICATION mode");
+ }
+ } catch (JobException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new JobException("Failed to initialize S3 EVENT source: " + e.getMessage());
+ }
+ }
+
+ @Override
+ public synchronized S3EventOffset getNextOffset(
+ StreamingJobProperties jobProperties, Map<String, String> properties) {
+ if (runningBatch != null) {
+ return runningBatch.offset;
+ }
+ Preconditions.checkState(readyBatch != null, "No ready S3 event batch");
+ runningBatch = readyBatch;
+ readyBatch = null;
+ return runningBatch.offset;
+ }
+
+ @Override
+ public synchronized String getShowCurrentOffset() {
+ String fileName = lastFileName(currentOffset);
+ return fileName == null ? null : GsonUtils.GSON.toJson(Collections.singletonMap("fileName", fileName));
+ }
+
+ @Override
+ public synchronized long getLastSourceEventTimestampSeconds() {
+ return lastSourceEventTimestampSeconds;
+ }
+
+ @Override
+ public synchronized long getLagMessages() {
+ return approximateNumberOfMessages == null ? -1 : approximateNumberOfMessages;
+ }
+
+ @Override
+ public synchronized String getShowMaxOffset() {
+ // Only accepted files count as observed progress; overflow and skipped messages do not.
+ S3EventOffset observedOffset = readyBatch != null ? readyBatch.offset
+ : runningBatch != null ? runningBatch.offset : currentOffset;
+ String fileName = lastFileName(observedOffset);
+ JsonObject result = new JsonObject();
+ if (fileName != null) {
+ result.addProperty("fileName", fileName);
+ }
+ if (approximateNumberOfMessages != null) {
+ result.addProperty("lagMessages", approximateNumberOfMessages);
+ }
+ return result.size() == 0 ? null : GsonUtils.GSON.toJson(result);
+ }
+
+ private String lastFileName(S3EventOffset offset) {
+ if (offset == null || offset.isEmpty()) {
+ return null;
+ }
+ // Receive order is not lexical order; this is a display value, not a recovery cursor.
+ List<String> files = offset.getFiles();
+ return files.get(files.size() - 1);
+ }
+
+ @Override
+ public InsertIntoTableCommand rewriteTvfParams(
+ InsertIntoTableCommand originCommand, Offset runningOffset, long taskId) {
+ S3EventOffset offset = (S3EventOffset) runningOffset;
+ return S3SourceOffsetProvider.rewriteS3Tvf(originCommand, offset.getFileStatuses());
+ }
+
+ @Override
+ public synchronized void updateOffset(Offset offset) {
+ currentOffset = (S3EventOffset) offset;
+ }
+
+ @Override
+ public void close() {
+ RemoteFileSystem fileSystemToClose;
+ SqsClient sqsClientToClose;
+ synchronized (this) {
+ closed = true;
+ fileSystemToClose = fileSystem;
+ sqsClientToClose = sqsClient;
+ fileSystem = null;
+ sqsClient = null;
+ clientProperties = null;
+ }
+ closeClientResources(fileSystemToClose, sqsClientToClose);
+ }
+
+ private void closeClientResources(RemoteFileSystem fileSystemToClose, SqsClient sqsClientToClose) {
+ if (sqsClientToClose != null) {
+ try {
+ sqsClientToClose.close();
+ } catch (RuntimeException e) {
+ log.warn("Failed to close SQS client for job {}", jobId, e);
+ }
+ }
+ if (fileSystemToClose != null) {
+ try {
+ fileSystemToClose.close();
+ } catch (IOException | RuntimeException e) {
+ log.warn("Failed to close S3 filesystem for job {}", jobId, e);
+ }
+ }
+ }
+
+ @Override
+ public void fetchRemoteMeta(
+ StreamingJobProperties jobProperties, Map<String, String> properties) {
+ // The job scheduler serializes fetches; task callbacks still access batch state concurrently.
+ TaskBatch batch = null;
+ try {
+ SqsClient client;
+ List<String> receipts;
+ synchronized (this) {
+ if (closed) {
+ return;
+ }
+ Map<String, String> copiedProps = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ copiedProps.putAll(properties);
+ // A pre-pause fetch may have old properties after ALTER/RESUME.
+ if (sqsClient != null && !copiedProps.equals(clientProperties)) {
+ closeClientResources(fileSystem, sqsClient);
+ fileSystem = null;
+ sqsClient = null;
+ clientProperties = null;
+ }
+ if (sqsClient == null) {
+ try {
+ fileSystem = FileSystemFactory.get(StorageProperties.createPrimary(copiedProps));
+ S3Client s3Client = ((S3ObjStorage) ((ObjFileSystem) fileSystem).getObjStorage()).getClient();
+ sqsClient = createSqsClient(s3Client);
+ clientProperties = copiedProps;
+ } catch (RuntimeException e) {
+ closeClientResources(fileSystem, sqsClient);
+ fileSystem = null;
+ sqsClient = null;
+ throw e;
+ }
+ }
+ client = sqsClient;
+ batch = readyBatch != null ? readyBatch : runningBatch;
+ receipts = new ArrayList<>(batch == null ? receiptsToDelete : batch.receiptHandles);
+ }
+ int visibilityTimeoutSeconds = (int) Math.max(MIN_VISIBILITY_SECONDS,
+ VISIBILITY_INTERVAL_MULTIPLIER * jobProperties.getMaxIntervalSecond());
+ try {
+ if (batch != null) {
+ if (!receipts.isEmpty()) {
+ renewBatchVisibility(client, batch, receipts, visibilityTimeoutSeconds);
+ }
+ return;
+ }
+ if (!receipts.isEmpty()) {
+ deletePendingMessages(client, receipts);
+ }
+ receiveBatch(client, jobProperties, visibilityTimeoutSeconds);
+ } finally {
+ if (!closed) {
+ refreshApproximateNumberOfMessages(client);
+ }
+ }
+ } catch (RuntimeException e) {
+ synchronized (this) {
+ if (closed) {
+ return;
+ }
+ if (batch != null && batch != readyBatch && batch != runningBatch) {
+ log.warn("SQS visibility update failed after the S3 EVENT batch committed", e);
+ return;
+ }
+ }
+ throw e;
+ }
+ }
+
+ private void refreshApproximateNumberOfMessages(SqsClient sqsClient) {
+ Long count = null;
+ try {
+ String value = sqsClient.getQueueAttributes(GetQueueAttributesRequest.builder()
+ .queueUrl(queueUrl)
+ .attributeNames(QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES)
+ .build()).attributes().get(QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES);
+ count = Long.parseLong(value);
+ } catch (RuntimeException e) {
+ // Monitoring must not block ingestion or report a stale count as current backlog.
+ if (!closed) {
+ log.warn("Failed to fetch SQS pending message count for queue {}", queueUrl, e);
+ }
+ }
+ synchronized (this) {
+ if (!closed) {
+ approximateNumberOfMessages = count;
+ }
+ }
+ }
+
+ private SqsClient createSqsClient(S3Client s3Client) {
+ return SqsClient.builder()
+ .region(s3Client.serviceClientConfiguration().region())
+ .credentialsProvider(s3Client.serviceClientConfiguration().credentialsProvider())
+ .build();
+ }
+
+ @Override
+ public void fetchRemoteMeta(Map<String, String> properties) {
+ throw new UnsupportedOperationException("S3 EVENT metadata fetch requires streaming job properties");
+ }
+
+ private void renewBatchVisibility(
+ SqsClient sqsClient, TaskBatch batch, List<String> receipts, int visibilityTimeoutSeconds) {
+ boolean renewed = true;
+ Set<String> invalidReceipts = new HashSet<>();
+ try {
+ for (int start = 0; start < receipts.size(); start += SQS_BATCH_SIZE) {
+ List<String> chunk = receipts.subList(start, Math.min(start + SQS_BATCH_SIZE, receipts.size()));
+ List<BatchResultErrorEntry> failures =
+ changeMessageVisibility(sqsClient, chunk, visibilityTimeoutSeconds);
+ for (BatchResultErrorEntry failure : failures) {
+ if (isInvalidReceipt(failure)) {
+ invalidReceipts.add(chunk.get(Integer.parseInt(failure.id())));
+ } else {
+ renewed = false;
+ }
+ }
+ }
+ } catch (RuntimeException e) {
+ throw new IllegalStateException("Failed to renew visibility for the current S3 EVENT batch", e);
+ } finally {
+ if (!invalidReceipts.isEmpty()) {
+ synchronized (this) {
+ batch.receiptsExpired = true;
+ batch.receiptHandles.removeAll(invalidReceipts);
+ invalidateReceipts(batch.fileReceipts, invalidReceipts);
+ // Commit may have completed while the renewal RPC was in progress.
+ if (unackedCommittedFiles != null) {
+ invalidateReceipts(unackedCommittedFiles, invalidReceipts);
+ }
+ receiptsToDelete.removeAll(invalidReceipts);
+ }
+ }
+ }
+ if (!invalidReceipts.isEmpty()) {
+ throw new IllegalStateException("S3 EVENT batch receipts expired; receive the batch again before retrying");
+ }
+ if (!renewed) {
+ throw new IllegalStateException(
+ "Failed to renew visibility for some messages in the current S3 EVENT batch");
+ }
+ }
+
+ private void receiveBatch(
+ SqsClient sqsClient, StreamingJobProperties jobProperties,
+ int visibilityTimeoutSeconds) {
+ List<String> files = new ArrayList<>();
+ List<String> receipts = new ArrayList<>();
+ Map<String, Map<String, String>> fileReceipts = new HashMap<>();
+ List<TBrokerFileStatus> fileStatuses = new ArrayList<>();
+ Set<String> selectedFiles = new HashSet<>();
+ long fileBytes = 0;
+ long serializedSize = new S3EventOffset(Collections.emptyList()).serializedSize();
+ long batchEventTimestampSeconds = 0;
+ int receivedMessages = 0;
+ int skippedMessages = 0;
+ ReceiveMessageRequest request = ReceiveMessageRequest.builder()
+ .queueUrl(queueUrl)
+ .maxNumberOfMessages(SQS_BATCH_SIZE)
+ .waitTimeSeconds(1)
+ .visibilityTimeout(visibilityTimeoutSeconds)
+ .build();
+ // Cap consecutive receives per fetch; this does not interrupt an ongoing SQS request.
+ long maxFetchDurationNanos = Math.min(TimeUnit.SECONDS.toNanos(jobProperties.getMaxIntervalSecond()) / 3,
+ TimeUnit.SECONDS.toNanos(5));
+ long fetchEndTimeNanos = System.nanoTime() + maxFetchDurationNanos;
+
+ do {
+ ReceiveMessageResponse response = sqsClient.receiveMessage(request);
+ if (closed) {
+ return;
+ }
+ List<Message> messages = response.messages();
+ receivedMessages += messages.size();
+ if (messages.isEmpty()) {
+ break;
+ }
+
+ List<ParsedMessage> candidates = parseMessages(sqsClient, messages, receipts);
+ // Delete skipped messages before filtering them out or stopping at the batch limit.
+ // Messages selected for INSERT are deleted only after the task commits.
+ deleteSkippedMessages(sqsClient, candidates, receipts);
+ candidates.removeIf(message -> message.files.isEmpty());
+ skippedMessages += messages.size() - candidates.size();
+
+ // Thresholds stop the next receive; every selected message from this receive stays in this batch.
+ for (ParsedMessage candidate : candidates) {
+ for (int i = 0; i < candidate.files.size(); i++) {
+ String key = candidate.files.get(i);
+ fileReceipts.computeIfAbsent(key, ignored -> new HashMap<>())
+ .put(candidate.messageId, candidate.receiptHandle);
+ // SQS may deliver the same key more than once; scan it once but acknowledge every receipt.
+ if (selectedFiles.add(key)) {
+ TBrokerFileStatus file = candidate.fileStatuses.get(i);
+ files.add(key);
+ fileStatuses.add(file);
+ fileBytes = Math.addExact(fileBytes, file.getSize());
+ serializedSize += GsonUtils.GSON.toJson(key).getBytes(StandardCharsets.UTF_8).length + 1;
+ }
+ }
+ receipts.add(candidate.receiptHandle);
+ batchEventTimestampSeconds = Math.max(batchEventTimestampSeconds, candidate.eventTimestampSeconds);
+ }
+ if (files.size() >= jobProperties.getS3BatchFiles()
+ || fileBytes >= jobProperties.getS3BatchBytes()
+ || serializedSize >= S3EventOffset.TARGET_BATCH_SERIALIZED_BYTES) {
+ break;
+ }
+ } while (!closed && System.nanoTime() < fetchEndTimeNanos);
+
+ if (!files.isEmpty()) {
+ makeReadyBatch(files, fileStatuses, receipts, fileReceipts, batchEventTimestampSeconds);
+ }
+ if (receivedMessages > 0) {
+ log.info("S3 EVENT receive batch: queue={}, receivedMessages={}, skippedMessages={}, "
+ + "selectedMessages={}, selectedFiles={}",
+ queueUrl, receivedMessages, skippedMessages, receipts.size(), files.size());
+ }
+ }
+
+ private List<ParsedMessage> parseMessages(
+ SqsClient sqsClient, List<Message> messages, List<String> acceptedReceipts) {
+ try {
+ List<ParsedMessage> parsedMessages = new ArrayList<>(messages.size());
+ for (Message message : messages) {
+ if (StringUtils.isBlank(message.receiptHandle())) {
+ throw new IllegalArgumentException("SQS message has no receipt handle");
+ }
+ parsedMessages.add(parseMessage(message));
+ }
+ return parsedMessages;
+ } catch (RuntimeException e) {
+ // No batch will be published; make uncommitted messages available for retry.
+ List<String> handles = new ArrayList<>(acceptedReceipts);
+ handles.addAll(messages.stream().map(Message::receiptHandle)
+ .filter(StringUtils::isNotBlank).collect(Collectors.toList()));
+ releaseMessages(sqsClient, handles);
+ throw e;
+ }
+ }
+
+ private void deleteSkippedMessages(SqsClient sqsClient, List<ParsedMessage> messages,
+ List<String> acceptedReceipts) {
+ List<String> skippedReceipts = messages.stream()
+ .filter(message -> message.files.isEmpty())
+ .map(message -> message.receiptHandle)
+ .collect(Collectors.toList());
+ if (skippedReceipts.isEmpty()) {
+ return;
+ }
+ synchronized (this) {
+ receiptsToDelete.addAll(skippedReceipts);
+ }
+ try {
+ deletePendingMessages(sqsClient, skippedReceipts);
+ } catch (RuntimeException e) {
+ releaseMessages(sqsClient, collectReceiptsToRelease(acceptedReceipts, messages));
+ throw e;
+ }
+ }
+
+ private ParsedMessage parseMessage(Message message) {
+ S3EventNotification notification;
+ try {
+ JsonNode json = Jackson.getObjectMapper().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",
+ jobId, message.messageId());
+ return ParsedMessage.skipped(message.receiptHandle());
+ }
+ if ("s3:TestEvent".equals(json.path("Event").asText())) {
+ return ParsedMessage.skipped(message.receiptHandle());
+ }
+ notification = Jackson.getObjectMapper().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()) {
+ 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<TBrokerFileStatus> fileStatuses = new ArrayList<>();
+ long eventTimestampSeconds = 0;
+ for (int i = 0; i < notification.getRecords().size(); i++) {
+ S3EventNotificationRecord record = notification.getRecords().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:")) {
+ skipReason = "Unsupported S3 event";
+ } else if (record.getS3() == null || record.getS3().getBucket() == null
+ || record.getS3().getObject() == null) {
+ skipReason = "Incomplete S3 event";
+ } else if (StringUtils.isBlank(record.getS3().getBucket().getName())
+ || StringUtils.isBlank(record.getS3().getObject().getKey())) {
+ skipReason = "Missing S3 bucket or object key";
+ } else if (record.getS3().getObject().getSizeAsLong() == null
+ || record.getS3().getObject().getSizeAsLong() < 0) {
+ skipReason = "Invalid S3 object size";
+ }
+ if (skipReason != null) {
+ log.warn("Skip invalid S3 EVENT record: jobId={}, messageId={}, recordIndex={}, reason={}",
+ jobId, message.messageId(), i, skipReason);
+ continue;
+ }
+ String bucket = record.getS3().getBucket().getName();
+ String key;
+ try {
+ key = record.getS3().getObject().getUrlDecodedKey();
+ if (!sourceBucket.equals(bucket) || !sourceKeyMatcher.matches(Paths.get(key))) {
+ continue;
+ }
+ } catch (IllegalArgumentException e) {
+ log.warn("Skip invalid S3 EVENT record: jobId={}, messageId={}, recordIndex={}, reason={}",
+ jobId, message.messageId(), i, e.getMessage());
+ continue;
+ }
+ if (isCommitted(key, message.messageId(), message.receiptHandle())) {
+ continue;
+ }
+ 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);
+ }
+ }
+ if (files.isEmpty()) {
+ return ParsedMessage.skipped(message.receiptHandle());
+ }
+ return new ParsedMessage(message.messageId(), message.receiptHandle(),
+ files, fileStatuses, eventTimestampSeconds);
+ }
+
+ private synchronized boolean isCommitted(String file, String messageId, String receiptHandle) {
+ boolean committed = (unackedCommittedFiles != null && unackedCommittedFiles.containsKey(file))
+ || (currentOffset != null && currentOffset.getFiles().contains(file));
+ if (committed) {
+ if (unackedCommittedFiles == null) {
+ unackedCommittedFiles = new HashMap<>();
+ }
+ unackedCommittedFiles.computeIfAbsent(file, ignored -> new HashMap<>()).put(messageId, receiptHandle);
+ }
+ return committed;
+ }
+
+ private synchronized void makeReadyBatch(List<String> files, List<TBrokerFileStatus> fileStatuses,
+ List<String> receipts, Map<String, Map<String, String>> fileReceipts, long eventTimestampSeconds) {
+ if (closed) {
+ return;
+ }
+ Preconditions.checkState(!files.isEmpty() && !receipts.isEmpty(),
+ "Cannot create an empty S3 EVENT batch");
+ S3EventOffset offset = new S3EventOffset(files, fileStatuses);
+ readyBatch = new TaskBatch(offset, new ArrayList<>(receipts), fileReceipts, eventTimestampSeconds);
+ }
+
+ private List<String> collectReceiptsToRelease(
+ List<String> acceptedReceipts, List<ParsedMessage> parsedMessages) {
+ List<String> handles = new ArrayList<>(acceptedReceipts);
+ handles.addAll(parsedMessages.stream().filter(message -> !message.files.isEmpty())
+ .map(message -> message.receiptHandle).collect(Collectors.toList()));
+ return handles;
+ }
+
+ private void releaseMessages(SqsClient sqsClient, List<String> handles) {
+ if (handles.isEmpty()) {
+ return;
+ }
+ try {
+ for (int start = 0; start < handles.size(); start += SQS_BATCH_SIZE) {
+ List<String> chunk = handles.subList(start, Math.min(start + SQS_BATCH_SIZE, handles.size()));
+ changeMessageVisibility(sqsClient, chunk, 0);
+ }
+ } catch (RuntimeException e) {
+ log.warn("Failed to release messages from a failed S3 EVENT receive batch", e);
+ }
+ }
+
+ private List<BatchResultErrorEntry> changeMessageVisibility(
+ SqsClient sqsClient, List<String> handles, int visibilityTimeout) {
+ List<ChangeMessageVisibilityBatchRequestEntry> entries = new ArrayList<>(handles.size());
+ for (int i = 0; i < handles.size(); i++) {
+ entries.add(ChangeMessageVisibilityBatchRequestEntry.builder()
+ .id(String.valueOf(i))
+ .receiptHandle(handles.get(i))
+ .visibilityTimeout(visibilityTimeout)
+ .build());
+ }
+ ChangeMessageVisibilityBatchResponse response = sqsClient.changeMessageVisibilityBatch(
+ ChangeMessageVisibilityBatchRequest.builder()
+ .queueUrl(queueUrl)
+ .entries(entries)
+ .build());
+ logBatchFailures("ChangeMessageVisibilityBatch", response.failed());
+ if (visibilityTimeout == 0) {
+ log.info("S3 EVENT release messages: queue={}, requestedMessages={}, "
+ + "succeededMessages={}, failedMessages={}",
+ queueUrl, handles.size(), response.successful().size(), response.failed().size());
+ }
+ return response.failed();
+ }
+
+ private void deletePendingMessages(SqsClient sqsClient, List<String> pending) {
+ List<String> failed = new ArrayList<>();
+ for (int start = 0; start < pending.size(); start += SQS_BATCH_SIZE) {
+ int end = Math.min(start + SQS_BATCH_SIZE, pending.size());
+ List<String> chunk = pending.subList(start, end);
+ try {
+ DeleteMessageBatchResponse response = sqsClient.deleteMessageBatch(DeleteMessageBatchRequest.builder()
+ .queueUrl(queueUrl)
+ .entries(buildDeleteEntries(chunk))
+ .build());
+ logBatchFailures("DeleteMessageBatch", response.failed());
+ log.info("S3 EVENT delete messages: queue={}, requestedMessages={}, "
+ + "succeededMessages={}, failedMessages={}",
+ queueUrl, chunk.size(), response.successful().size(), response.failed().size());
+ synchronized (this) {
+ if (unackedCommittedFiles != null && !unackedCommittedFiles.isEmpty()) {
+ Set<String> deletedReceipts = response.successful().stream()
+ .map(entry -> chunk.get(Integer.parseInt(entry.id())))
+ .collect(Collectors.toSet());
+ unackedCommittedFiles.entrySet().removeIf(entry -> {
+ Map<String, String> messageReceipts = entry.getValue();
+ boolean removed = messageReceipts.values().removeAll(deletedReceipts);
+ return removed && messageReceipts.isEmpty();
+ });
+ }
+ }
+ failed.addAll(getRetryableReceiptHandles(chunk, response.failed()));
+ } catch (RuntimeException e) {
+ // Retain current and unattempted chunks; the request outcome is unknown.
+ failed.addAll(pending.subList(start, pending.size()));
+ synchronized (this) {
+ receiptsToDelete = failed;
+ }
+ throw new IllegalStateException("Failed to delete S3 EVENT messages", e);
+ }
+ }
+ synchronized (this) {
+ // Fetch is serialized and no task batch exists during deletion.
+ receiptsToDelete = failed;
+ }
+ if (!failed.isEmpty()) {
+ throw new IllegalStateException("Failed to delete some S3 EVENT messages");
+ }
+ }
+
+ private boolean isInvalidReceipt(BatchResultErrorEntry failure) {
+ // InvalidParameterValue also covers invalid visibility durations, so only match explicit receipt expiry.
+ return "ReceiptHandleIsInvalid".equals(failure.code())
+ || "MessageNotInflight".equals(failure.code())
+ || ("InvalidParameterValue".equals(failure.code())
+ && StringUtils.containsIgnoreCase(failure.message(), "receipt handle has expired"));
+ }
+
+ private void logBatchFailures(String operation, List<BatchResultErrorEntry> failures) {
+ for (BatchResultErrorEntry failure : failures) {
+ log.warn("SQS {} failed: id={}, code={}, senderFault={}",
+ operation, failure.id(), failure.code(), failure.senderFault());
+ if (isInvalidReceipt(failure)) {
+ // Keep committed files until a fresh receipt confirms deletion.
+ log.warn("SQS {} receipt is no longer usable: id={}; message deletion is not confirmed",
+ operation, failure.id());
+ }
+ }
+ }
+
+ private List<DeleteMessageBatchRequestEntry> buildDeleteEntries(List<String> handles) {
+ List<DeleteMessageBatchRequestEntry> entries = new ArrayList<>(handles.size());
+ for (int i = 0; i < handles.size(); i++) {
+ entries.add(DeleteMessageBatchRequestEntry.builder()
+ .id(String.valueOf(i))
+ .receiptHandle(handles.get(i))
+ .build());
+ }
+ return entries;
+ }
+
+ private List<String> getRetryableReceiptHandles(List<String> handles, List<BatchResultErrorEntry> failures) {
+ return failures.stream().filter(failure -> !isInvalidReceipt(failure))
+ .map(failure -> handles.get(Integer.parseInt(failure.id())))
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public synchronized boolean hasMoreDataToConsume() {
+ return readyBatch != null || runningBatch != null;
+ }
+
+ @Override
+ public synchronized String getPersistInfo() {
+ return currentOffset == null ? null : currentOffset.toSerializedJson();
+ }
+
+ @Override
+ public synchronized void restoreFromPersistInfo(String persistInfo) {
+ if (persistInfo == null) {
+ return;
+ }
+ try {
+ currentOffset = GsonUtils.GSON.fromJson(persistInfo, S3EventOffset.class);
+ } catch (Exception e) {
+ log.warn("Failed to restore S3 EVENT offset from persist info", e);
+ }
+ }
+
+ @Override
+ public synchronized void replayIfNeed(StreamingInsertJob job) {
+ if (currentOffset == null) {
+ restoreFromPersistInfo(job.getOffsetProviderPersist());
+ }
+ // Capture after transaction replay; subsequent commits and resumes must not replace this snapshot.
+ if (unackedCommittedFiles == null) {
+ unackedCommittedFiles = new HashMap<>();
+ if (currentOffset != null) {
+ for (String file : currentOffset.getFiles()) {
+ unackedCommittedFiles.put(file, new HashMap<>());
+ }
+ }
+ }
+ // The scheduler serializes this resume hook with fetches, after transaction progress is replayed.
+ TaskBatch batch = readyBatch != null ? readyBatch : runningBatch;
+ if (batch != null) {
+ boolean committed = currentOffset != null && currentOffset.getFiles().equals(batch.offset.getFiles());
+ if (committed) {
+ rememberCommittedBatch(batch);
+ receiptsToDelete.addAll(batch.receiptHandles);
+ }
+ readyBatch = null;
+ runningBatch = null;
+ log.info("S3 EVENT resume batch: jobId={}, committed={}, files={}, receipts={}",
+ jobId, committed, batch.offset.getFiles().size(), batch.receiptHandles.size());
+ }
+ }
+
+ @Override
+ public Offset deserializeOffset(String offset) {
+ return GsonUtils.GSON.fromJson(offset, S3EventOffset.class);
+ }
+
+ @Override
+ public Offset deserializeOffsetProperty(String offset) {
+ return null;
+ }
+
+ @Override
+ public void validateAlterOffset(String offset) throws JobException {
+ throw new JobException("offset cannot be altered when s3.ingestion_mode is NOTIFICATION");
+ }
+
+ @Override
+ public synchronized String getCommitOffsetJson(Offset runningOffset, long taskId, List<Long> scanBackendIds) {
+ Preconditions.checkState(runningBatch != null, "No running S3 EVENT batch");
+ Preconditions.checkState(!runningBatch.receiptsExpired, "S3 EVENT batch receipts expired before commit");
+ S3EventOffset offset = (S3EventOffset) runningOffset;
+ Preconditions.checkState(runningBatch.offset.getFiles().equals(offset.getFiles()),
+ "S3 EVENT running offset changed before commit");
+ return offset.toSerializedJson();
+ }
+
+ @Override
+ public synchronized void onTaskCommitted(long scannedRows, long loadBytes) {
+ Preconditions.checkState(runningBatch != null, "No running S3 EVENT batch to commit");
+ Preconditions.checkState(receiptsToDelete.isEmpty(), "Previous S3 EVENT message deletion is pending");
+ Preconditions.checkState(currentOffset != null
+ && currentOffset.getFiles().equals(runningBatch.offset.getFiles()),
+ "Committed S3 EVENT offset does not match the running batch");
+ rememberCommittedBatch(runningBatch);
+ receiptsToDelete = new ArrayList<>(runningBatch.receiptHandles);
+ lastSourceEventTimestampSeconds = runningBatch.eventTimestampSeconds;
+ runningBatch = null;
+ }
+
+ private void rememberCommittedBatch(TaskBatch batch) {
+ if (unackedCommittedFiles == null) {
+ unackedCommittedFiles = new HashMap<>();
+ }
+ batch.fileReceipts.forEach((file, receipts) ->
+ unackedCommittedFiles.computeIfAbsent(file, ignored -> new HashMap<>()).putAll(receipts));
+ }
+
+ private static void invalidateReceipts(Map<String, Map<String, String>> files, Set<String> invalidReceipts) {
+ files.values().forEach(receipts -> receipts.replaceAll(
+ (messageId, receipt) -> invalidReceipts.contains(receipt) ? null : receipt));
+ }
+
+ private static class TaskBatch {
+ private final S3EventOffset offset;
+ private final List<String> receiptHandles;
+ private final Map<String, Map<String, String>> fileReceipts;
+ private final long eventTimestampSeconds;
+ private boolean receiptsExpired;
+
+ private TaskBatch(S3EventOffset offset, List<String> receiptHandles,
+ Map<String, Map<String, String>> fileReceipts, long eventTimestampSeconds) {
+ this.offset = offset;
+ this.receiptHandles = receiptHandles;
+ this.fileReceipts = fileReceipts;
+ this.eventTimestampSeconds = eventTimestampSeconds;
+ }
+ }
+
+ private static class ParsedMessage {
+ private final String messageId;
+ private final String receiptHandle;
+ private final List<String> files;
+ private final List<TBrokerFileStatus> fileStatuses;
+ private final long eventTimestampSeconds;
+
+ private ParsedMessage(String messageId, String receiptHandle, List<String> files,
+ List<TBrokerFileStatus> fileStatuses, long eventTimestampSeconds) {
+ this.messageId = messageId;
+ this.receiptHandle = receiptHandle;
+ this.files = files;
+ this.fileStatuses = fileStatuses;
+ this.eventTimestampSeconds = eventTimestampSeconds;
+ }
+
+ private static ParsedMessage skipped(String receiptHandle) {
+ return new ParsedMessage(null, receiptHandle, Collections.emptyList(), Collections.emptyList(), 0);
+ }
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3Offset.java b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3Offset.java
index 44df1756e02..eccba403772 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3Offset.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3Offset.java
@@ -19,21 +19,23 @@ package org.apache.doris.job.offset.s3;
import org.apache.doris.job.offset.Offset;
import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.thrift.TBrokerFileStatus;
import com.google.gson.annotations.SerializedName;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.lang3.StringUtils;
+import java.util.List;
+
@Getter
@Setter
public class S3Offset implements Offset {
String startFile;
@SerializedName("endFile")
String endFile;
- // s3://bucket/path/{1.csv,2.csv}
- String fileLists;
int fileNum;
+ private transient List<TBrokerFileStatus> fileStatuses;
@SerializedName("lastBatch")
boolean lastBatch;
@@ -44,7 +46,7 @@ public class S3Offset implements Offset {
@Override
public boolean isEmpty() {
- return StringUtils.isEmpty(fileLists);
+ return fileStatuses == null || fileStatuses.isEmpty();
}
@Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java
index a89affb9536..02b22a95ed5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java
@@ -30,11 +30,14 @@ import org.apache.doris.job.offset.Offset;
import org.apache.doris.job.offset.SourceOffsetProvider;
import org.apache.doris.nereids.analyzer.UnboundTVFRelation;
import org.apache.doris.nereids.trees.expressions.Properties;
-import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.expressions.functions.table.S3;
import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand;
-import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalTVFRelation;
import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.Maps;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
@@ -42,25 +45,25 @@ import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang3.StringUtils;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.stream.Collectors;
@Log4j2
public class S3SourceOffsetProvider implements SourceOffsetProvider {
- private final boolean onceMode;
+ private final boolean oneTimeMode;
private volatile S3Offset noMoreFilesAfterOffset;
volatile S3Offset currentOffset;
volatile String maxEndFile;
public S3SourceOffsetProvider() {
- this.onceMode = false;
+ this.oneTimeMode = false;
}
public S3SourceOffsetProvider(StreamingJobProperties jobProperties) {
- this.onceMode = jobProperties.isS3OnceMode();
+ this.oneTimeMode = jobProperties.isS3OneTimeMode();
}
@Override
@@ -90,28 +93,17 @@ public class S3SourceOffsetProvider implements SourceOffsetProvider {
if (!rfiles.isEmpty()) {
String bucket = globListResult.getBucket();
- String prefix = globListResult.getPrefix();
-
String bucketBase = "s3://" + bucket + "/";
- // Get the path of the last directory
- int lastSlash = prefix.lastIndexOf('/');
- String basePrefix = (lastSlash >= 0) ? prefix.substring(0, lastSlash + 1) : "";
- String filePathBase = bucketBase + basePrefix;
- String joined = rfiles.stream()
- .map(path -> path.getName().replace(filePathBase, ""))
- .collect(Collectors.joining(","));
-
- String normalizedPrefix = basePrefix.endsWith("/")
- ? basePrefix.substring(0, basePrefix.length() - 1) : basePrefix;
- String finalFileLists = String.format("s3://%s/%s/{%s}", bucket, normalizedPrefix, joined);
String beginFile = rfiles.get(0).getName().replace(bucketBase, "");
String lastFile = rfiles.get(rfiles.size() - 1).getName().replace(bucketBase, "");
- offset.setFileLists(finalFileLists);
+ offset.setFileStatuses(rfiles.stream().map(file -> new TBrokerFileStatus(
+ file.getName(), file.isDirectory(), file.getSize(), !file.isDirectory()))
+ .collect(Collectors.toList()));
offset.setStartFile(beginFile);
offset.setEndFile(lastFile);
offset.setFileNum(rfiles.size());
maxEndFile = globListResult.getMaxFile();
- offset.setLastBatch(onceMode && lastFile.equals(globListResult.getMaxFile()));
+ offset.setLastBatch(oneTimeMode && lastFile.equals(globListResult.getMaxFile()));
} else {
throw new RuntimeException("No new files found in path: " + filePath);
}
@@ -146,29 +138,54 @@ public class S3SourceOffsetProvider implements SourceOffsetProvider {
public InsertIntoTableCommand rewriteTvfParams(InsertIntoTableCommand originCommand,
Offset runningOffset, long taskId) {
S3Offset offset = (S3Offset) runningOffset;
- Map<String, String> props = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
- // rewrite plan
- Plan rewritePlan = originCommand.getParsedPlan().get().rewriteUp(plan -> {
+ return rewriteS3Tvf(originCommand, offset.getFileStatuses());
+ }
+
+ // Render the same literal batch in task properties and audit SQL; execution uses file statuses.
+ public static String buildFileListUri(List<TBrokerFileStatus> files) {
+ Preconditions.checkArgument(!files.isEmpty(), "S3 task batch must contain files");
+ String[] paths = files.stream().map(TBrokerFileStatus::getPath).toArray(String[]::new);
+ String commonPrefix = StringUtils.getCommonPrefix(paths);
+ String baseUri = commonPrefix.substring(0, commonPrefix.lastIndexOf('/') + 1);
+ String joined = Arrays.stream(paths).map(path -> escapeGlob(path.substring(baseUri.length())))
+ .collect(Collectors.joining(","));
+ return escapeGlob(baseUri) + "{" + joined + "}";
+ }
+
+ // Selected object keys are literals; only the surrounding braces and separators are glob syntax.
+ private static String escapeGlob(String value) {
+ StringBuilder escaped = new StringBuilder(value.length());
+ for (int i = 0; i < value.length(); i++) {
+ char c = value.charAt(i);
+ if ("\\*?[]{},".indexOf(c) >= 0) {
+ escaped.append('\\');
+ }
+ escaped.append(c);
+ }
+ return escaped.toString();
+ }
+
+ static InsertIntoTableCommand rewriteS3Tvf(InsertIntoTableCommand originCommand,
+ List<TBrokerFileStatus> fileStatuses) {
+ Preconditions.checkNotNull(fileStatuses, "S3 task batch must contain exact file statuses");
+ String fileLists = buildFileListUri(fileStatuses);
+ return originCommand.rewriteQuery(plan -> {
if (plan instanceof UnboundTVFRelation) {
- UnboundTVFRelation originTvfRel = (UnboundTVFRelation) plan;
- Map<String, String> oriMap = originTvfRel.getProperties().getMap();
- props.putAll(oriMap);
- props.put("uri", offset.getFileLists());
- return new UnboundTVFRelation(
- originTvfRel.getRelationId(), originTvfRel.getFunctionName(), new Properties(props));
+ UnboundTVFRelation tvf = (UnboundTVFRelation) plan;
+ Map<String, String> props = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
+ props.putAll(tvf.getProperties().getMap());
+ props.put("uri", fileLists);
+ return new LogicalTVFRelation(tvf.getRelationId(), S3.withFiles(new Properties(props), fileStatuses),
+ ImmutableList.of());
}
return plan;
});
- InsertIntoTableCommand insertIntoTableCommand = new InsertIntoTableCommand((LogicalPlan) rewritePlan,
- Optional.empty(), Optional.empty(), Optional.empty(), true, Optional.empty());
- insertIntoTableCommand.setJobId(originCommand.getJobId());
- return insertIntoTableCommand;
}
@Override
public void updateOffset(Offset offset) {
this.currentOffset = (S3Offset) offset;
- this.currentOffset.setFileLists(null);
+ this.currentOffset.setFileStatuses(null);
}
@Override
@@ -193,7 +210,7 @@ public class S3SourceOffsetProvider implements SourceOffsetProvider {
? globListResult.getStatus().getErrMsg() : "null result";
throw new Exception("Failed to list S3 files: " + errMsg);
}
- if (onceMode) {
+ if (oneTimeMode) {
noMoreFilesAfterOffset = objects.isEmpty() ? offsetAtScan : null;
}
if (!objects.isEmpty() && StringUtils.isNotEmpty(globListResult.getMaxFile())) {
@@ -220,7 +237,7 @@ public class S3SourceOffsetProvider implements SourceOffsetProvider {
@Override
public boolean hasReachedEnd() {
S3Offset offset = currentOffset;
- return onceMode && offset != null && (offset.isLastBatch() || noMoreFilesAfterOffset == offset);
+ return oneTimeMode && offset != null && (offset.isLastBatch() || noMoreFilesAfterOffset == offset);
}
@Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java b/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java
index a89e5ed0e03..2e5484f0d41 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java
@@ -101,6 +101,7 @@ public final class MetricRepo {
public static final String STREAMING_JOB_PER_JOB_SUCCEED_TASK_COUNT = "streaming_job_per_job_succeed_task_count";
public static final String STREAMING_JOB_PER_JOB_FAILED_TASK_COUNT = "streaming_job_per_job_failed_task_count";
public static final String STREAMING_JOB_PER_JOB_LAG_BYTES = "streaming_job_per_job_lag_bytes";
+ public static final String STREAMING_JOB_PER_JOB_LAG_MESSAGES = "streaming_job_per_job_lag_messages";
public static final String STREAMING_JOB_PER_JOB_LAST_SOURCE_EVENT_TIMESTAMP_SECONDS =
"streaming_job_per_job_last_source_event_timestamp_seconds";
public static final String STREAMING_JOB_PER_JOB_LAST_TASK_SUCCESS_TIME_SECONDS =
@@ -1355,6 +1356,7 @@ public final class MetricRepo {
DORIS_METRIC_REGISTER.removeMetrics(STREAMING_JOB_PER_JOB_SUCCEED_TASK_COUNT);
DORIS_METRIC_REGISTER.removeMetrics(STREAMING_JOB_PER_JOB_FAILED_TASK_COUNT);
DORIS_METRIC_REGISTER.removeMetrics(STREAMING_JOB_PER_JOB_LAG_BYTES);
+ DORIS_METRIC_REGISTER.removeMetrics(STREAMING_JOB_PER_JOB_LAG_MESSAGES);
DORIS_METRIC_REGISTER.removeMetrics(STREAMING_JOB_PER_JOB_LAST_SOURCE_EVENT_TIMESTAMP_SECONDS);
DORIS_METRIC_REGISTER.removeMetrics(STREAMING_JOB_PER_JOB_LAST_TASK_SUCCESS_TIME_SECONDS);
@@ -1452,10 +1454,25 @@ public final class MetricRepo {
.addLabel(new MetricLabel("job_name", jobName));
DORIS_METRIC_REGISTER.addMetrics(lag);
+ long lagMessages = sJob.getLagMessages();
+ if (lagMessages >= 0) {
+ GaugeMetric<Long> messageLag = new GaugeMetric<Long>(
+ STREAMING_JOB_PER_JOB_LAG_MESSAGES, MetricUnit.NOUNIT,
+ "last observed approximate visible message backlog, excluding in-flight messages") {
+ @Override
+ public Long getValue() {
+ return lagMessages;
+ }
+ };
+ messageLag.addLabel(new MetricLabel("job_id", jobId))
+ .addLabel(new MetricLabel("job_name", jobName));
+ DORIS_METRIC_REGISTER.addMetrics(messageLag);
+ }
+
long lastSourceEventTimestampSeconds = sJob.getLastSourceEventTimestampSeconds();
GaugeMetric<Long> lastSourceEventTimestamp = new GaugeMetric<Long>(
STREAMING_JOB_PER_JOB_LAST_SOURCE_EVENT_TIMESTAMP_SECONDS, MetricUnit.SECONDS,
- "timestamp of the latest source binlog or WAL event recorded in the job's committed offset"
+ "source event timestamp of the latest committed batch"
+ " as Unix seconds, 0 means unavailable") {
@Override
public Long getValue() {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/S3.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/S3.java
index 3c4c924d7f4..34e89336baf 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/S3.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/S3.java
@@ -19,6 +19,7 @@ package org.apache.doris.nereids.trees.expressions.functions.table;
import org.apache.doris.catalog.FunctionSignature;
import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.Properties;
import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
import org.apache.doris.nereids.trees.plans.logical.SupportPruneNestedColumn;
@@ -26,13 +27,40 @@ import org.apache.doris.nereids.trees.plans.logical.SupportPruneNestedColumnForm
import org.apache.doris.nereids.types.coercion.AnyDataType;
import org.apache.doris.tablefunction.S3TableValuedFunction;
import org.apache.doris.tablefunction.TableValuedFunctionIf;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
import java.util.Map;
+import java.util.Objects;
/** s3 */
public class S3 extends TableValuedFunction implements SupportPruneNestedColumn {
+ private final List<TBrokerFileStatus> fileStatuses;
+
public S3(Properties properties) {
+ this(properties, null);
+ }
+
+ // Task-only inputs must not register another SQL function overload.
+ private S3(Properties properties, List<TBrokerFileStatus> fileStatuses) {
super("s3", properties);
+ this.fileStatuses = fileStatuses == null ? null : ImmutableList.copyOf(fileStatuses);
+ }
+
+ public static S3 withFiles(Properties properties, List<TBrokerFileStatus> fileStatuses) {
+ return new S3(properties, fileStatuses);
+ }
+
+ @Override
+ protected boolean extraEquals(Expression that) {
+ return super.extraEquals(that) && Objects.equals(fileStatuses, ((S3) that).fileStatuses);
+ }
+
+ @Override
+ public int computeHashCode() {
+ return Objects.hash(super.computeHashCode(), fileStatuses);
}
@Override
@@ -44,7 +72,7 @@ public class S3 extends TableValuedFunction implements SupportPruneNestedColumn
protected TableValuedFunctionIf toCatalogFunction() {
try {
Map<String, String> arguments = getTVFProperties().getMap();
- return new S3TableValuedFunction(arguments);
+ return new S3TableValuedFunction(arguments, fileStatuses);
} catch (Throwable t) {
throw new AnalysisException("Can not build s3(): " + t.getMessage(), t);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java
index e31af87bb36..fda03d8bd8f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterJobCommand.java
@@ -284,8 +284,12 @@ public class AlterJobCommand extends AlterCommand implements ForwardWithSync, Ne
mergedProperties.putAll(properties);
StreamingJobProperties updatedJobProperties = new StreamingJobProperties(mergedProperties);
updatedJobProperties.validate();
- if (!originJobProperties.getS3IngestionMode().equals(updatedJobProperties.getS3IngestionMode())) {
- throw new AnalysisException("s3.ingestion_mode cannot be altered");
+ if (!originJobProperties.getS3IngestionMode().equals(updatedJobProperties.getS3IngestionMode())
+ || !StringUtils.equalsIgnoreCase(originJobProperties.getS3EventSource(),
+ updatedJobProperties.getS3EventSource())
+ || !Objects.equals(originJobProperties.getS3SqsQueueUrl(),
+ updatedJobProperties.getS3SqsQueueUrl())) {
+ throw new AnalysisException("s3.ingestion_mode, s3.event.source and s3.sqs.queue_url cannot be altered");
}
String offset = properties.get(StreamingJobProperties.OFFSET_PROPERTY);
if (offset != null) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
index 97c58141256..2bd2ce18c4a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
@@ -103,6 +103,7 @@ import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
+import java.util.function.Function;
import java.util.function.Supplier;
/**
@@ -187,6 +188,15 @@ public class InsertIntoTableCommand extends Command implements NeedAuditEncrypti
return logicalQuery.orElse(originLogicalQuery);
}
+ /** Rewrite the original query before planning, preserving INSERT options and the WITH clause. */
+ public InsertIntoTableCommand rewriteQuery(Function<Plan, Plan> rewrite) {
+ InsertIntoTableCommand command = new InsertIntoTableCommand(
+ (LogicalPlan) originLogicalQuery.rewriteUp(rewrite), labelName, insertCtx,
+ cte, needNormalizePlan, branchName);
+ command.setJobId(jobId);
+ return command;
+ }
+
public Optional<Plan> getParsedPlan() {
return parsedPlan;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java
index 62d2d465544..41466e0648a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java
@@ -199,6 +199,10 @@ public abstract class ExternalFileTableValuedFunction extends TableValuedFunctio
}
protected void parseFile() throws AnalysisException {
+ parseFile(null);
+ }
+
+ protected void parseFile(List<TBrokerFileStatus> files) throws AnalysisException {
long startAt = System.currentTimeMillis();
String path = getFilePath();
BrokerDesc brokerDesc = getBrokerDesc();
@@ -209,6 +213,10 @@ public abstract class ExternalFileTableValuedFunction extends TableValuedFunctio
String endpoint = storageProperties.getEndpoint();
S3Util.validateAndTestEndpoint(endpoint);
}
+ if (files != null) {
+ fileStatuses.addAll(files);
+ return;
+ }
BrokerUtil.parseFile(path, brokerDesc, fileStatuses);
} catch (UserException e) {
throw new AnalysisException("parse file failed, err: " + e.getMessage(), e);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java
index 75e811f4b2c..91e79dda10e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java
@@ -25,6 +25,7 @@ import org.apache.doris.common.UserException;
import org.apache.doris.datasource.lance.metadata.LanceMetadataLoader;
import org.apache.doris.datasource.lance.metadata.LanceTableMetadata;
import org.apache.doris.datasource.property.storage.StorageProperties;
+import org.apache.doris.thrift.TBrokerFileStatus;
import org.apache.doris.thrift.TFileType;
import java.util.Collections;
@@ -46,6 +47,11 @@ public class S3TableValuedFunction extends ExternalFileTableValuedFunction {
public static final String NAME = "s3";
public S3TableValuedFunction(Map<String, String> properties) throws AnalysisException {
+ this(properties, null);
+ }
+
+ public S3TableValuedFunction(Map<String, String> properties, List<TBrokerFileStatus> files)
+ throws AnalysisException {
// 1. analyze common properties
Map<String, String> props = super.parseCommonProperties(properties);
try {
@@ -61,6 +67,9 @@ public class S3TableValuedFunction extends ExternalFileTableValuedFunction {
// Just check
// Fixme wait to be done #50320
// FileSystemFactory.get(storageProperties);
+ if (files != null) {
+ this.fileStatuses.addAll(files);
+ }
return;
}
if (isLanceFormat()) {
@@ -75,7 +84,7 @@ public class S3TableValuedFunction extends ExternalFileTableValuedFunction {
e);
}
} else {
- parseFile();
+ parseFile(files);
}
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/common/util/S3UtilTest.java b/fe/fe-core/src/test/java/org/apache/doris/common/util/S3UtilTest.java
index 6e61893ed3b..ed3cdd56328 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/common/util/S3UtilTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/common/util/S3UtilTest.java
@@ -20,6 +20,9 @@ package org.apache.doris.common.util;
import org.junit.Assert;
import org.junit.Test;
+import java.nio.file.FileSystems;
+import java.nio.file.PathMatcher;
+import java.nio.file.Paths;
import java.util.Arrays;
import java.util.List;
@@ -108,9 +111,9 @@ public class S3UtilTest {
@Test
public void testExtendGlobNumberRange_invalidCharacters() {
- // Test invalid characters {Refrain,1..3}
+ // Preserve literal file names alongside numeric ranges.
String input = "file_{Refrain,1..3}.csv";
- String expected = "file_{1,2,3}.csv";
+ String expected = "file_{Refrain,1,2,3}.csv";
String result = S3Util.extendGlobNumberRange(input);
Assert.assertEquals(expected, result);
}
@@ -224,6 +227,54 @@ public class S3UtilTest {
Assert.assertEquals(expected, result);
}
+ @Test
+ public void testExtendGlobNumberRange_preservesPaddedValues() {
+ for (String pattern : Arrays.asList("base/day/001", "base/day/{001}",
+ "base/day/{001,002}", "base/day/{001,1}", "base/day/{2147483648}")) {
+ Assert.assertEquals(pattern, S3Util.extendGlobs(pattern));
+ }
+ Assert.assertEquals("base/day/{001,1,2}",
+ S3Util.extendGlobs("base/day/{001,1..2}"));
+ }
+
+ @Test
+ public void testExtendGlobNumberRange_matchesPaddedFileNames() {
+ PathMatcher matcher = FileSystems.getDefault().getPathMatcher(
+ "glob:" + S3Util.extendGlobs("base/day/{001,002}"));
+ Assert.assertTrue(matcher.matches(Paths.get("base/day/001")));
+ Assert.assertTrue(matcher.matches(Paths.get("base/day/002")));
+ Assert.assertFalse(matcher.matches(Paths.get("base/day/1")));
+ Assert.assertFalse(matcher.matches(Paths.get("base/day/2")));
+ }
+
+ @Test
+ public void testExtendGlobNumberRange_preservesDistinctHeadPaths() {
+ Assert.assertEquals(Arrays.asList("base/day/001", "base/day/1"),
+ S3Util.expandBracePatterns(S3Util.extendGlobs("base/day/{001,1,001,1}")));
+ Assert.assertEquals(Arrays.asList("base/day/1"),
+ S3Util.expandBracePatterns(S3Util.extendGlobs("base/day/{1,1}")));
+ }
+
+ @Test
+ public void testExtendGlobNumberRange_preservesMixedFileNames() {
+ Assert.assertEquals("base/day/{001,abc}", S3Util.extendGlobs("base/day/{001,abc}"));
+ Assert.assertEquals("base/day/{abc,1,2}", S3Util.extendGlobs("base/day/{abc,1..2}"));
+ Assert.assertEquals(Arrays.asList("base/day/001", "base/day/abc.csv"),
+ S3Util.expandBracePatterns(S3Util.extendGlobs("base/day/{001,abc.csv,001,abc.csv}")));
+ PathMatcher matcher = FileSystems.getDefault().getPathMatcher(
+ "glob:" + S3Util.extendGlobs("base/day/{001,abc.csv}"));
+ Assert.assertTrue(matcher.matches(Paths.get("base/day/001")));
+ Assert.assertTrue(matcher.matches(Paths.get("base/day/abc.csv")));
+ }
+
+ @Test
+ public void testExtendGlobNumberRange_preservesLiteralCharacters() {
+ for (String pattern : Arrays.asList("base/day/{001,$file}", "base/day/{001,\\file}",
+ "base/day/{001, abc }", "base/day/{abc,}", "base/day/{$file}")) {
+ Assert.assertEquals(pattern, S3Util.extendGlobs(pattern));
+ }
+ }
+
@Test
public void testGetLongestPrefix_withGlobPattern() {
// Test getLongestPrefix with glob patterns
diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobLagTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobLagTest.java
index e1a64931c8b..a52a5825bf4 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobLagTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobLagTest.java
@@ -23,6 +23,7 @@ import org.apache.doris.job.base.TimerDefinition;
import org.apache.doris.job.cdc.split.BinlogSplit;
import org.apache.doris.job.offset.jdbc.JdbcOffset;
import org.apache.doris.job.offset.jdbc.JdbcSourceOffsetProvider;
+import org.apache.doris.job.offset.s3.S3EventSourceOffsetProvider;
import org.junit.Assert;
import org.junit.Test;
@@ -33,6 +34,20 @@ import java.util.Map;
public class StreamingInsertJobLagTest {
+ @Test
+ public void testMessageLagUsesOffsetProvider() {
+ StreamingInsertJob job = Deencapsulation.newInstance(StreamingInsertJob.class);
+ Assert.assertEquals(-1L, job.getLagMessages());
+ Deencapsulation.setField(job, "offsetProvider", new JdbcSourceOffsetProvider());
+ Assert.assertEquals(-1L, job.getLagMessages());
+
+ S3EventSourceOffsetProvider provider = new S3EventSourceOffsetProvider("test-queue");
+ Deencapsulation.setField(job, "offsetProvider", provider);
+ Assert.assertEquals(-1L, job.getLagMessages());
+ Deencapsulation.setField(provider, "approximateNumberOfMessages", 12L);
+ Assert.assertEquals(12L, job.getLagMessages());
+ }
+
@Test
public void testLastSourceEventTimestampUsesOffsetProvider() {
StreamingInsertJob job = Deencapsulation.newInstance(StreamingInsertJob.class);
diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java
index fde21328560..fce3a18b359 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobOffsetPersistenceTest.java
@@ -171,7 +171,7 @@ public class StreamingInsertJobOffsetPersistenceTest {
@Test
public void testS3OnceLastBatchFinishesOnlyAfterSuccess() throws Exception {
- StreamingJobProperties properties = new StreamingJobProperties(ImmutableMap.of("s3.ingestion_mode", "ONCE"));
+ StreamingJobProperties properties = new StreamingJobProperties(ImmutableMap.of("s3.ingestion_mode", "ONE_TIME"));
TestStreamingInsertJob failedJob = newJob(new S3SourceOffsetProvider(properties), 1017L);
NoopStreamingMultiTblTask failedTask =
(NoopStreamingMultiTblTask) Deencapsulation.getField(failedJob, "runningStreamTask");
@@ -241,7 +241,7 @@ public class StreamingInsertJobOffsetPersistenceTest {
@Test
public void testCloudReplayRefreshesPersistedOffset() {
- Map<String, String> properties = ImmutableMap.of("s3.ingestion_mode", "ONCE");
+ Map<String, String> properties = ImmutableMap.of("s3.ingestion_mode", "ONE_TIME");
S3SourceOffsetProvider provider = new S3SourceOffsetProvider(new StreamingJobProperties(properties));
StreamingInsertJob job = new StreamingInsertJob();
job.offsetProvider = provider;
diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java
index aedc3bd15c1..30bf7ce12e5 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJobStatusTransitionTest.java
@@ -17,14 +17,35 @@
package org.apache.doris.job.extensions.insert.streaming;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.jmockit.Deencapsulation;
import org.apache.doris.job.common.JobStatus;
import org.apache.doris.job.exception.JobException;
+import org.apache.doris.job.offset.SourceOffsetProvider;
+import org.apache.doris.job.offset.s3.S3EventSourceOffsetProvider;
+import org.apache.doris.transaction.GlobalTransactionMgrIface;
+import org.apache.doris.transaction.TxnStateCallbackFactory;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
public class StreamingInsertJobStatusTransitionTest {
+ private static StreamingInsertJob newJob(JobStatus status) {
+ StreamingInsertJob job = Deencapsulation.newInstance(StreamingInsertJob.class);
+ Deencapsulation.setField(job, "lock", new ReentrantReadWriteLock(true));
+ Deencapsulation.setField(job, "jobId", 8001L);
+ Deencapsulation.setField(job, "jobName", "test_job");
+ Deencapsulation.setField(job, "jobStatus", status);
+ return job;
+ }
+
+
@Test
public void testFinishedStatusCannotBeOverwritten() {
StreamingInsertJob job = new StreamingInsertJob();
@@ -33,4 +54,49 @@ public class StreamingInsertJobStatusTransitionTest {
Assertions.assertThrows(JobException.class, () -> job.updateJobStatus(JobStatus.PAUSED));
Assertions.assertEquals(JobStatus.FINISHED, job.getJobStatus());
}
+
+ @Test
+ public void testPauseAndResumeKeepOffsetProviderOpen() throws Exception {
+ StreamingInsertJob job = newJob(JobStatus.RUNNING);
+ SourceOffsetProvider provider = Mockito.mock(SourceOffsetProvider.class);
+ Deencapsulation.setField(job, "offsetProvider", provider);
+
+ job.updateJobStatus(JobStatus.PAUSED);
+ job.updateJobStatus(JobStatus.PENDING);
+
+ Mockito.verify(provider, Mockito.never()).close();
+ }
+
+ @Test
+ public void testFinalStatusesCloseOffsetProvider() throws Exception {
+ try (MockedStatic<Env> env = Mockito.mockStatic(Env.class)) {
+ GlobalTransactionMgrIface transactionMgr = Mockito.mock(GlobalTransactionMgrIface.class);
+ TxnStateCallbackFactory callbackFactory = Mockito.mock(TxnStateCallbackFactory.class);
+ env.when(Env::getCurrentGlobalTransactionMgr).thenReturn(transactionMgr);
+ Mockito.when(transactionMgr.getCallbackFactory()).thenReturn(callbackFactory);
+ for (JobStatus status : new JobStatus[] {JobStatus.STOPPED, JobStatus.FINISHED}) {
+ StreamingInsertJob job = newJob(JobStatus.RUNNING);
+ SourceOffsetProvider provider = Mockito.mock(SourceOffsetProvider.class);
+ Deencapsulation.setField(job, "offsetProvider", provider);
+
+ job.updateJobStatus(status);
+
+ Mockito.verify(provider).close();
+ }
+ }
+ }
+
+ @Test
+ public void testDropPendingJobPreventsLateMetadataFetch() throws Exception {
+ StreamingInsertJob job = newJob(JobStatus.PENDING);
+ S3EventSourceOffsetProvider provider = new S3EventSourceOffsetProvider("unused-queue");
+ Deencapsulation.setField(job, "offsetProvider", provider);
+
+ job.cleanup();
+ job.updateJobStatus(JobStatus.RUNNING);
+
+ // An old scheduler may still hold the removed job; it must not create clients with these invalid properties.
+ Assertions.assertDoesNotThrow(() -> provider.fetchRemoteMeta(
+ new StreamingJobProperties(Collections.emptyMap()), Collections.emptyMap()));
+ }
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java
index 520ad9c1005..29145af4174 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java
@@ -29,6 +29,7 @@ import org.apache.doris.job.extensions.insert.InsertTask;
import org.apache.doris.job.offset.Offset;
import org.apache.doris.job.offset.SourceOffsetProvider;
import org.apache.doris.job.offset.jdbc.JdbcTvfSourceOffsetProvider;
+import org.apache.doris.job.offset.s3.S3EventOffset;
import org.apache.doris.job.offset.s3.S3Offset;
import org.apache.doris.job.offset.s3.S3SourceOffsetProvider;
import org.apache.doris.nereids.StatementContext;
@@ -36,17 +37,23 @@ import org.apache.doris.nereids.analyzer.UnboundTVFRelation;
import org.apache.doris.nereids.glue.LogicalPlanAdapter;
import org.apache.doris.nereids.parser.NereidsParser;
import org.apache.doris.nereids.trees.expressions.Properties;
+import org.apache.doris.nereids.trees.expressions.functions.table.S3;
import org.apache.doris.nereids.trees.plans.RelationId;
import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand;
-import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalTVFRelation;
import org.apache.doris.plugin.AuditEvent;
import org.apache.doris.qe.AuditLogHelper;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.QueryState;
import org.apache.doris.qe.StmtExecutor;
import org.apache.doris.resource.workloadschedpolicy.WorkloadRuntimeStatusMgr;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.MockedConstruction;
@@ -55,11 +62,30 @@ import org.mockito.Mockito;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
public class StreamingInsertTaskAuditTest {
+ private ConnectContext previousContext;
+
+ @BeforeEach
+ public void setUp() {
+ previousContext = ConnectContext.get();
+ ConnectContext ctx = new ConnectContext();
+ ctx.setThreadLocalInfo();
+ ctx.setStatementContext(new StatementContext());
+ }
+
+ @AfterEach
+ public void tearDown() {
+ ConnectContext.remove();
+ if (previousContext != null) {
+ previousContext.setThreadLocalInfo();
+ }
+ }
+
private static final String ORIGIN_URI = "s3://bucket/input/*.csv";
private static final String RESOLVED_URI = "s3://bucket/input/{1.csv,2.csv}";
private static final String S3_SQL = "insert into target_table select * from s3("
@@ -88,6 +114,7 @@ public class StreamingInsertTaskAuditTest {
Assertions.assertEquals(StmtType.INSERT.name(), auditEvent.stmtType);
Assertions.assertFalse(auditEvent.stmt.contains(ORIGIN_URI));
Assertions.assertTrue(auditEvent.stmt.contains(RESOLVED_URI));
+ Assertions.assertFalse(auditEvent.stmt.contains("WITH LABEL"));
Assertions.assertFalse(auditEvent.stmt.contains("private-value"));
Assertions.assertEquals("OK", auditEvent.state);
Assertions.assertTrue(auditEvent.isInternal);
@@ -101,9 +128,27 @@ public class StreamingInsertTaskAuditTest {
Assertions.assertEquals(StmtType.INSERT.name(), auditEvent.stmtType);
Assertions.assertEquals("ERR", auditEvent.state);
Assertions.assertTrue(auditEvent.errorMessage.contains("insert failed"));
+ Assertions.assertFalse(auditEvent.stmt.contains("WITH LABEL"));
Assertions.assertTrue(auditEvent.isInternal);
}
+ @Test
+ public void testAuditPreservesOriginalInsertClauses() throws Exception {
+ for (String target : new String[] {
+ "target_table(c1)",
+ "`db`.`target table` PARTITION (p0) (c1)",
+ "target_table WITH LABEL old_label (c1)",
+ "target_table [shuffle]",
+ "target_table /* target comment */ (c1)"}) {
+ AuditEvent event = runTask(null, S3_SQL.replace("target_table", target),
+ new S3SourceOffsetProvider(), new S3Offset(), true);
+ Assertions.assertTrue(event.stmt.contains(target));
+ Assertions.assertFalse(event.stmt.contains("WITH LABEL `1_2`"));
+ Assertions.assertTrue(event.stmt.contains(RESOLVED_URI));
+ Assertions.assertFalse(event.stmt.contains("private-value"));
+ }
+ }
+
@Test
public void testCdcRunDoesNotSubmitAuditEvent() throws Exception {
String sql = "insert into target_table select * from cdc_stream("
@@ -130,13 +175,11 @@ public class StreamingInsertTaskAuditTest {
SourceOffsetProvider provider = Mockito.mock(SourceOffsetProvider.class);
Mockito.when(provider.getSourceType()).thenReturn("s3");
S3Offset offset = new S3Offset();
- offset.setFileLists(RESOLVED_URI);
Mockito.when(provider.getNextOffset(Mockito.eq(properties), Mockito.anyMap())).thenReturn(offset);
InsertIntoTableCommand baseCommand = Mockito.mock(InsertIntoTableCommand.class);
- Mockito.when(baseCommand.getParsedPlan()).thenReturn(Optional.of(Mockito.mock(LogicalPlan.class)));
InsertIntoTableCommand taskCommand = Mockito.mock(InsertIntoTableCommand.class);
- UnboundTVFRelation tvf = Mockito.mock(UnboundTVFRelation.class);
- Mockito.when(taskCommand.getAllTVFRelation()).thenReturn(Collections.singletonList(tvf));
+ Mockito.when(taskCommand.getLogicalQuery()).thenReturn(auditRelation());
+ offset.setFileStatuses(batchFiles());
Mockito.when(provider.rewriteTvfParams(Mockito.eq(baseCommand), Mockito.eq(offset), Mockito.anyLong()))
.thenReturn(taskCommand);
Mockito.doAnswer(invocation -> {
@@ -158,7 +201,7 @@ public class StreamingInsertTaskAuditTest {
MockedStatic<AuditLogHelper> audit = Mockito.mockStatic(AuditLogHelper.class)) {
insertTask.when(() -> InsertTask.makeConnectContext(UserIdentity.ROOT, "test_db")).thenReturn(ctx);
StreamingInsertTask task = new StreamingInsertTask(1L, 2L, S3_SQL, provider, "test_db", properties,
- Collections.emptyMap(), UserIdentity.ROOT, null);
+ ImmutableMap.of("URI", ORIGIN_URI), UserIdentity.ROOT, null);
// A retry must not audit files from an earlier attempt after preparation fails.
Deencapsulation.setField(task, "auditSql", "stale audit SQL from a previous attempt");
task.before();
@@ -168,7 +211,9 @@ public class StreamingInsertTaskAuditTest {
Mockito.verify(parsers.constructed().get(1)).parseForEncryption(Mockito.eq(S3_SQL), Mockito.anyMap());
task.run();
Assertions.assertEquals(QueryState.MysqlStateType.OK, state.getStateType());
- Mockito.verify(taskCommand).run(ctx, executors.constructed().get(1));
+ Assertions.assertEquals(1, executors.constructed().size());
+ Mockito.verify(baseCommand, Mockito.never()).initPlan(Mockito.any(), Mockito.any(), Mockito.anyBoolean());
+ Mockito.verify(taskCommand).run(ctx, executors.constructed().get(0));
audit.verifyNoInteractions();
}
}
@@ -179,28 +224,85 @@ public class StreamingInsertTaskAuditTest {
originProperties.put("URI", ORIGIN_URI);
UnboundTVFRelation originTvf = new UnboundTVFRelation(
new RelationId(1), "s3", new Properties(originProperties));
- InsertIntoTableCommand originCommand = Mockito.mock(InsertIntoTableCommand.class);
- Mockito.when(originCommand.getParsedPlan()).thenReturn(Optional.of(originTvf));
S3Offset offset = new S3Offset();
- offset.setFileLists(RESOLVED_URI);
Env env = Mockito.mock(Env.class);
Mockito.when(env.isMaster()).thenReturn(false);
try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ InsertIntoTableCommand originCommand = new InsertIntoTableCommand(originTvf,
+ Optional.of("original_label"), Optional.empty(), Optional.empty());
+ originCommand.setJobId(42L);
+ offset.setFileStatuses(batchFiles());
InsertIntoTableCommand rewritten = new S3SourceOffsetProvider()
.rewriteTvfParams(originCommand, offset, 1L);
Map<String, String> rewrittenProperties =
- rewritten.getAllTVFRelation().get(0).getProperties().getMap();
+ ((LogicalTVFRelation) rewritten.getLogicalQuery()).getFunction().getTVFProperties().getMap();
+ Assertions.assertEquals(Optional.of("original_label"), rewritten.getLabelName());
+ Assertions.assertEquals(42L, rewritten.getJobId());
Assertions.assertEquals(1, rewrittenProperties.size());
Assertions.assertEquals(RESOLVED_URI, rewrittenProperties.get("URI"));
}
}
+ @Test
+ public void testExactFileRewritePreservesWithClause() {
+ ConnectContext.get().setDatabase("test_db");
+ Env env = Mockito.mock(Env.class);
+ CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
+ InternalCatalog catalog = Mockito.mock(InternalCatalog.class);
+ Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
+ Mockito.when(catalogMgr.getCatalog(Mockito.anyString())).thenReturn(catalog);
+ Mockito.when(catalog.getName()).thenReturn("internal");
+ Mockito.when(env.isMaster()).thenReturn(false);
+ try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+ mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+ InsertIntoTableCommand command = (InsertIntoTableCommand) new NereidsParser().parseSingle(
+ "with input as (select 1 as c) insert into target_table select * from s3(\"uri\"=\""
+ + ORIGIN_URI + "\") cross join input");
+ S3Offset offset = new S3Offset();
+ offset.setFileStatuses(batchFiles());
+ InsertIntoTableCommand rewritten = new S3SourceOffsetProvider().rewriteTvfParams(command, offset, 1L);
+ Optional<?> originalCte = Deencapsulation.getField(command, "cte");
+ Optional<?> rewrittenCte = Deencapsulation.getField(rewritten, "cte");
+ Assertions.assertTrue(originalCte.isPresent());
+ Assertions.assertEquals(originalCte, rewrittenCte);
+ List<LogicalTVFRelation> relations = rewritten.getLogicalQuery()
+ .collectToList(LogicalTVFRelation.class::isInstance);
+ Assertions.assertEquals(1, relations.size());
+ Assertions.assertEquals(RESOLVED_URI,
+ relations.get(0).getFunction().getTVFProperties().getMap().get("uri"));
+ Assertions.assertEquals(batchFiles(),
+ Deencapsulation.getField(relations.get(0).getFunction(), "fileStatuses"));
+ }
+ }
+
+ @Test
+ public void testNotificationAuditUsesExactBatch() throws Exception {
+ SourceOffsetProvider provider = Mockito.mock(SourceOffsetProvider.class);
+ Mockito.when(provider.getSourceType()).thenReturn("s3");
+ AuditEvent event = runTask(null, S3_SQL, provider,
+ new S3EventOffset(ImmutableList.of("input/1.csv", "input/2.csv"), batchFiles()), true);
+ Assertions.assertTrue(event.stmt.contains(RESOLVED_URI));
+ Assertions.assertFalse(event.stmt.contains(ORIGIN_URI));
+ Assertions.assertFalse(event.stmt.contains("private-value"));
+ }
+
+ private static LogicalTVFRelation auditRelation() {
+ Map<String, String> properties = ImmutableMap.of(
+ "uri", RESOLVED_URI, "s3.secret_key", "private-value", "enclose", "\"");
+ return new LogicalTVFRelation(new RelationId(1), S3.withFiles(new Properties(properties), batchFiles()),
+ ImmutableList.of());
+ }
+
+ private static List<TBrokerFileStatus> batchFiles() {
+ return ImmutableList.of(new TBrokerFileStatus("s3://bucket/input/1.csv", false, 10, true),
+ new TBrokerFileStatus("s3://bucket/input/2.csv", false, 20, true));
+ }
+
private AuditEvent runS3Task(RuntimeException commandFailure) throws Exception {
S3Offset offset = new S3Offset();
- offset.setFileLists(RESOLVED_URI);
return runTask(commandFailure, S3_SQL, new S3SourceOffsetProvider(), offset, true);
}
@@ -223,13 +325,7 @@ public class StreamingInsertTaskAuditTest {
ctx.getState().setOk();
InsertIntoTableCommand command = Mockito.mock(InsertIntoTableCommand.class);
if (expectAudit) {
- Map<String, String> rewrittenTvfProps = new HashMap<>();
- rewrittenTvfProps.put("uri", RESOLVED_URI);
- rewrittenTvfProps.put("s3.secret_key", "private-value");
- rewrittenTvfProps.put("enclose", "\"");
- UnboundTVFRelation tvf = Mockito.mock(UnboundTVFRelation.class);
- Mockito.when(tvf.getProperties()).thenReturn(new Properties(rewrittenTvfProps));
- Mockito.when(command.getAllTVFRelation()).thenReturn(Collections.singletonList(tvf));
+ Mockito.when(command.getLogicalQuery()).thenReturn(auditRelation());
}
AtomicLong commandStartTime = new AtomicLong();
Mockito.doAnswer(invocation -> {
@@ -249,7 +345,8 @@ public class StreamingInsertTaskAuditTest {
StreamingInsertTask task = new StreamingInsertTask(
1L, 2L, sql, offsetProvider, "test_db", null,
- Collections.emptyMap(), UserIdentity.ROOT, null);
+ ImmutableMap.of("URI", ORIGIN_URI, "s3.secret_key", "private-value", "enclose", "\""),
+ UserIdentity.ROOT, null);
Deencapsulation.setField(task, "ctx", ctx);
Deencapsulation.setField(task, "taskCommand", command);
Deencapsulation.setField(task, "stmtExecutor", executor);
diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java
index 2e42c9f0c10..0e8c68bd74e 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJobPropertiesTest.java
@@ -22,6 +22,9 @@ import org.apache.doris.common.jmockit.Deencapsulation;
import org.apache.doris.job.common.JobStatus;
import org.apache.doris.job.exception.JobException;
import org.apache.doris.job.extensions.insert.InsertTask;
+import org.apache.doris.job.offset.SourceOffsetProviderFactory;
+import org.apache.doris.job.offset.s3.S3EventSourceOffsetProvider;
+import org.apache.doris.job.offset.s3.S3SourceOffsetProvider;
import org.apache.doris.nereids.trees.plans.commands.AlterJobCommand;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.SessionVariable;
@@ -33,17 +36,65 @@ import org.junit.Test;
import java.util.HashMap;
public class StreamingJobPropertiesTest {
+ @Test
+ public void testS3ProviderSelectionDefaultsToOrderedList() throws AnalysisException {
+ StreamingJobProperties orderedList = new StreamingJobProperties(new HashMap<>());
+ orderedList.validate();
+ Assert.assertTrue(SourceOffsetProviderFactory.createSourceOffsetProvider("s3", orderedList)
+ instanceof S3SourceOffsetProvider);
+
+ HashMap<String, String> props = new HashMap<>();
+ props.put("s3.ingestion_mode", "notification");
+ props.put("s3.event.source", "sqs");
+ props.put("s3.sqs.queue_url", "https://sqs.us-west-2.amazonaws.com/123456789012/events");
+ StreamingJobProperties event = new StreamingJobProperties(props);
+ event.validate();
+ Assert.assertTrue(SourceOffsetProviderFactory.createSourceOffsetProvider("s3", event)
+ instanceof S3EventSourceOffsetProvider);
+ }
+
+ @Test
+ public void testS3ModeNames() {
+ Assert.assertEquals("ORDERED_LIST", new StreamingJobProperties(ImmutableMap.of()).getS3IngestionMode());
+ Assert.assertEquals("ORDERED_LIST",
+ new StreamingJobProperties(ImmutableMap.of("s3.ingestion_mode", " ordered_list ")).getS3IngestionMode());
+ Assert.assertTrue(new StreamingJobProperties(
+ ImmutableMap.of("s3.ingestion_mode", "NOTIFICATION")).isS3NotificationMode());
+ Assert.assertTrue(new StreamingJobProperties(
+ ImmutableMap.of("s3.ingestion_mode", "ONE_TIME")).isS3OneTimeMode());
+ for (String mode : new String[] {"LEXICAL", "EVENT", "ONCE"}) {
+ Assert.assertThrows(AnalysisException.class,
+ () -> new StreamingJobProperties(ImmutableMap.of("s3.ingestion_mode", mode)).validate());
+ }
+ }
+
+ @Test
+ public void testNotificationRequiresQueueAndRejectsManualOffset() throws AnalysisException {
+ HashMap<String, String> props = new HashMap<>();
+ props.put("s3.ingestion_mode", "NOTIFICATION");
+ Assert.assertThrows(AnalysisException.class, () -> new StreamingJobProperties(props).validate());
+ props.put("s3.event.source", "SQS");
+ Assert.assertThrows(AnalysisException.class, () -> new StreamingJobProperties(props).validate());
+ props.put("s3.sqs.queue_url", "https://sqs.us-west-2.amazonaws.com/123456789012/events");
+ new StreamingJobProperties(props).validate();
+ props.put("offset", "logs/a.csv");
+ Assert.assertThrows(AnalysisException.class, () -> new StreamingJobProperties(props).validate());
+ props.remove("offset");
+ props.put("max_interval", "7200");
+ Assert.assertThrows(AnalysisException.class, () -> new StreamingJobProperties(props).validate());
+ }
+
@Test
- public void testS3OnceModeValidationAndAlter() throws Exception {
- StreamingJobProperties properties = new StreamingJobProperties(ImmutableMap.of("s3.ingestion_mode", " once "));
+ public void testS3OneTimeModeValidationAndAlter() throws Exception {
+ StreamingJobProperties properties = new StreamingJobProperties(ImmutableMap.of("s3.ingestion_mode", " one_time "));
properties.validate();
- Assert.assertTrue(properties.isS3OnceMode());
+ Assert.assertTrue(properties.isS3OneTimeMode());
Assert.assertThrows(AnalysisException.class,
() -> new StreamingJobProperties(ImmutableMap.of("s3.ingestion_mode", "invalid")).validate());
Assert.assertThrows(AnalysisException.class,
() -> new StreamingJobProperties(
- ImmutableMap.of("s3.ingestion_mode", "ONCE", "offset", "{\"fileName\":\"a.csv\"}"))
+ ImmutableMap.of("s3.ingestion_mode", "ONE_TIME", "offset", "{\"fileName\":\"a.csv\"}"))
.validate());
StreamingInsertJob job = new StreamingInsertJob();
@@ -51,7 +102,7 @@ public class StreamingJobPropertiesTest {
AlterJobCommand alterBatch = new AlterJobCommand("job", ImmutableMap.of("s3.max_batch_files", "1"),
null, null, null, ImmutableMap.of(), ImmutableMap.of());
Deencapsulation.invoke(alterBatch, "validateProps", job);
- AlterJobCommand alterMode = new AlterJobCommand("job", ImmutableMap.of("s3.ingestion_mode", "LEXICAL"),
+ AlterJobCommand alterMode = new AlterJobCommand("job", ImmutableMap.of("s3.ingestion_mode", "ORDERED_LIST"),
null, null, null, ImmutableMap.of(), ImmutableMap.of());
Assert.assertThrows(AnalysisException.class,
() -> Deencapsulation.invoke(alterMode, "validateProps", job));
@@ -62,9 +113,9 @@ public class StreamingJobPropertiesTest {
}
@Test
- public void testS3OnceModeRestoredWithLegacyOffset() throws Exception {
+ public void testS3OneTimeModeRestoredWithLegacyOffset() throws Exception {
StreamingInsertJob job = new StreamingInsertJob();
- Deencapsulation.setField(job, "properties", ImmutableMap.of("s3.ingestion_mode", "ONCE"));
+ Deencapsulation.setField(job, "properties", ImmutableMap.of("s3.ingestion_mode", "ONE_TIME"));
Deencapsulation.setField(job, "tvfType", "s3");
job.setOffsetProviderPersist("{\"endFile\":\"data/a.csv\"}");
job.gsonPostProcess();
diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3EventOffsetTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3EventOffsetTest.java
new file mode 100644
index 00000000000..6f464ac8247
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3EventOffsetTest.java
@@ -0,0 +1,56 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.offset.s3;
+
+import org.apache.doris.persist.gson.GsonUtils;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+public class S3EventOffsetTest {
+ @Test
+ public void testSerializedOffsetPreservesKeysAndCountsUtf8Bytes() {
+ List<String> files = Arrays.asList("logs/ä¸æ–‡.csv", "logs/a\"b\\c.csv");
+ S3EventOffset offset = new S3EventOffset(files);
+ String expected = "{\"files\":[\"logs/ä¸æ–‡.csv\",\"logs/a\\\"b\\\\c.csv\"]}";
+
+ Assertions.assertEquals(expected, offset.toSerializedJson());
+ Assertions.assertEquals(expected, offset.showRange());
+ Assertions.assertEquals(expected.getBytes(StandardCharsets.UTF_8).length, offset.serializedSize());
+ Assertions.assertEquals(files, GsonUtils.GSON.fromJson(expected, S3EventOffset.class).getFiles());
+ Assertions.assertTrue(offset.isValidOffset());
+ }
+
+ @Test
+ public void testTaskFilesCannotBeChangedThroughCallerList() {
+ List<String> files = new ArrayList<>(Collections.singletonList("logs/a.csv"));
+ S3EventOffset offset = new S3EventOffset(files);
+ files.add("logs/b.csv");
+
+ Assertions.assertEquals(Collections.singletonList("logs/a.csv"), offset.getFiles());
+ Assertions.assertThrows(UnsupportedOperationException.class, () -> offset.getFiles().add("logs/c.csv"));
+ Assertions.assertFalse(GsonUtils.GSON.fromJson("{}", S3EventOffset.class).isValidOffset());
+ Assertions.assertFalse(new S3EventOffset(Collections.singletonList(" ")).isValidOffset());
+ }
+}
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
new file mode 100644
index 00000000000..1bca291f532
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProviderTest.java
@@ -0,0 +1,1185 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.job.offset.s3;
+
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob;
+import org.apache.doris.job.extensions.insert.streaming.StreamingJobProperties;
+import org.apache.doris.persist.gson.GsonUtils;
+
+import org.apache.commons.lang3.StringUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
+import software.amazon.awssdk.identity.spi.AwsCredentialsIdentity;
+import software.amazon.awssdk.identity.spi.IdentityProvider;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3ClientBuilder;
+import software.amazon.awssdk.services.s3.S3ServiceClientConfiguration;
+import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
+import software.amazon.awssdk.services.sqs.SqsClient;
+import software.amazon.awssdk.services.sqs.SqsClientBuilder;
+import software.amazon.awssdk.services.sqs.model.BatchResultErrorEntry;
+import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityBatchRequest;
+import software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityBatchResponse;
+import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchRequest;
+import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchResponse;
+import software.amazon.awssdk.services.sqs.model.DeleteMessageBatchResultEntry;
+import software.amazon.awssdk.services.sqs.model.GetQueueAttributesRequest;
+import software.amazon.awssdk.services.sqs.model.GetQueueAttributesResponse;
+import software.amazon.awssdk.services.sqs.model.Message;
+import software.amazon.awssdk.services.sqs.model.QueueAttributeName;
+import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
+import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.StsClientBuilder;
+import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
+import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
+import software.amazon.awssdk.services.sts.model.AssumeRoleResponse;
+import software.amazon.awssdk.services.sts.model.Credentials;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+public class S3EventSourceOffsetProviderTest {
+ private static final String QUEUE_URL = "https://sqs.us-west-2.amazonaws.com/123456789012/events";
+
+ private S3EventSourceOffsetProvider provider;
+ private Map<String, String> tvfProperties;
+ private SqsClient sqsClient;
+ private SqsClientBuilder sqsBuilder;
+ private MockedStatic<SqsClient> sqsStatic;
+ private S3Client s3Client;
+ private S3ClientBuilder s3Builder;
+ private AwsCredentialsProvider s3CredentialsProvider;
+ private MockedStatic<S3Client> s3Static;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ tvfProperties = new HashMap<>();
+ tvfProperties.put("uri", "s3://bucket/logs/**");
+ tvfProperties.put("s3.endpoint", "s3.us-west-2.amazonaws.com");
+ tvfProperties.put("s3.region", "us-west-2");
+ tvfProperties.put("s3.access_key", "test-ak");
+ tvfProperties.put("s3.secret_key", "test-sk");
+ provider = new S3EventSourceOffsetProvider(QUEUE_URL);
+ provider.ensureInitialized(1L, tvfProperties);
+ sqsClient = Mockito.mock(SqsClient.class);
+ sqsBuilder = Mockito.mock(SqsClientBuilder.class, Mockito.RETURNS_SELF);
+ Mockito.when(sqsBuilder.build()).thenReturn(sqsClient);
+ sqsStatic = Mockito.mockStatic(SqsClient.class);
+ sqsStatic.when(SqsClient::builder).thenReturn(sqsBuilder);
+ s3Client = Mockito.mock(S3Client.class);
+ s3Builder = Mockito.mock(S3ClientBuilder.class, Mockito.RETURNS_SELF);
+ Mockito.when(s3Builder.build()).thenReturn(s3Client);
+ Mockito.when(s3Builder.credentialsProvider(Mockito.any(AwsCredentialsProvider.class)))
+ .thenAnswer(invocation -> {
+ s3CredentialsProvider = invocation.getArgument(0);
+ return s3Builder;
+ });
+ Mockito.when(s3Client.serviceClientConfiguration()).thenAnswer(invocation ->
+ S3ServiceClientConfiguration.builder().region(Region.US_WEST_2)
+ .credentialsProvider(s3CredentialsProvider).build());
+ s3Static = Mockito.mockStatic(S3Client.class);
+ s3Static.when(S3Client::builder).thenReturn(s3Builder);
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(ReceiveMessageResponse.builder().build());
+ Mockito.when(sqsClient.deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class)))
+ .thenAnswer(invocation -> {
+ DeleteMessageBatchRequest request = invocation.getArgument(0);
+ return DeleteMessageBatchResponse.builder().successful(request.entries().stream()
+ .map(entry -> DeleteMessageBatchResultEntry.builder().id(entry.id()).build())
+ .collect(Collectors.toList())).build();
+ });
+ Mockito.when(sqsClient.changeMessageVisibilityBatch(Mockito.any(ChangeMessageVisibilityBatchRequest.class)))
+ .thenReturn(ChangeMessageVisibilityBatchResponse.builder().build());
+ Mockito.when(sqsClient.getQueueAttributes(Mockito.any(GetQueueAttributesRequest.class)))
+ .thenReturn(GetQueueAttributesResponse.builder().attributes(Collections.singletonMap(
+ QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES, "0")).build());
+ }
+
+ @AfterEach
+ public void tearDown() {
+ provider.close();
+ s3Static.close();
+ sqsStatic.close();
+ }
+
+ @Test
+ public void testClientsAreReusedUntilClosed() {
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Mockito.verify(s3Builder).build();
+ Mockito.verify(sqsBuilder).build();
+ Mockito.verify(sqsClient, Mockito.never()).close();
+ Mockito.verify(s3Client, Mockito.never()).close();
+
+ provider.close();
+ provider.close();
+ InOrder closeOrder = Mockito.inOrder(sqsClient, s3Client);
+ closeOrder.verify(sqsClient).close();
+ closeOrder.verify(s3Client).close();
+ }
+
+ @Test
+ public void testChangedCredentialsReplaceClientsAndReuseReplacement() throws Exception {
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ tvfProperties.put("s3.access_key", "new-ak");
+ tvfProperties.put("s3.secret_key", "new-sk");
+ provider.ensureInitialized(1L, tvfProperties);
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Mockito.verify(s3Builder, Mockito.times(2)).build();
+ Mockito.verify(sqsBuilder, Mockito.times(2)).build();
+ Mockito.verify(sqsClient).close();
+ Mockito.verify(s3Client).close();
+ ArgumentCaptor<AwsCredentialsProvider> credentials = ArgumentCaptor.forClass(AwsCredentialsProvider.class);
+ Mockito.verify(sqsBuilder, Mockito.times(2)).credentialsProvider(
+ (IdentityProvider<? extends AwsCredentialsIdentity>) credentials.capture());
+ Assertions.assertEquals("test-ak", credentials.getAllValues().get(0).resolveCredentials().accessKeyId());
+ Assertions.assertEquals("new-ak", credentials.getAllValues().get(1).resolveCredentials().accessKeyId());
+ }
+
+ @Test
+ public void testCloseWaitsForClientCreation() throws Exception {
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ CountDownLatch closeStarted = new CountDownLatch(1);
+ AtomicReference<Future<?>> closeFuture = new AtomicReference<>();
+ Mockito.when(sqsBuilder.build()).thenAnswer(invocation -> {
+ closeFuture.set(executor.submit(() -> {
+ closeStarted.countDown();
+ provider.close();
+ }));
+ Assertions.assertTrue(closeStarted.await(5, TimeUnit.SECONDS));
+ Assertions.assertFalse(closeFuture.get().isDone());
+ Mockito.verify(s3Client, Mockito.never()).close();
+ return sqsClient;
+ });
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenAnswer(invocation -> {
+ closeFuture.get().get(5, TimeUnit.SECONDS);
+ return response(filesMessage("late", "logs/a.csv"));
+ });
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ closeFuture.get().get(5, TimeUnit.SECONDS);
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+ Mockito.verify(sqsClient).close();
+ Mockito.verify(s3Client).close();
+ provider.ensureInitialized(1L, tvfProperties);
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Mockito.verify(sqsBuilder).build();
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testCloseDuringReceiveDiscardsLateBatch() throws Exception {
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenAnswer(invocation -> {
+ executor.submit(() -> provider.close()).get(5, TimeUnit.SECONDS);
+ Mockito.verify(sqsClient).close();
+ return response(filesMessage("in-flight", "logs/a.csv"));
+ });
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+ Mockito.verify(sqsClient).close();
+ Mockito.verify(s3Client).close();
+ provider.ensureInitialized(1L, tvfProperties);
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Mockito.verify(sqsBuilder).build();
+ Mockito.verify(sqsClient, Mockito.never()).deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class));
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testCloseDuringReceiveSuppressesRequestFailure() {
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenAnswer(invocation -> {
+ provider.close();
+ throw new IllegalStateException("client closed");
+ });
+
+ Assertions.assertDoesNotThrow(() -> provider.fetchRemoteMeta(jobProperties(1), tvfProperties));
+
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+ Mockito.verify(sqsClient).close();
+ Mockito.verify(s3Client).close();
+ Mockito.verify(sqsClient, Mockito.never()).getQueueAttributes(Mockito.any(GetQueueAttributesRequest.class));
+ }
+
+ @Test
+ public void testClientCreationFailureReleasesFileSystemAndAllowsRetry() {
+ Mockito.when(sqsBuilder.build()).thenThrow(new IllegalStateException("client creation failed"))
+ .thenReturn(sqsClient);
+ Assertions.assertThrows(IllegalStateException.class,
+ () -> provider.fetchRemoteMeta(jobProperties(1), tvfProperties));
+ Mockito.verify(s3Client).close();
+
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Mockito.verify(s3Builder, Mockito.times(2)).build();
+ Mockito.verify(sqsBuilder, Mockito.times(2)).build();
+ }
+
+ @Test
+ public void testSqsCloseFailureStillClosesFileSystem() {
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Mockito.doThrow(new IllegalStateException("close failed")).when(sqsClient).close();
+ provider.close();
+ Mockito.verify(s3Client).close();
+ }
+
+ @Test
+ public void testNotificationMetadataDoesNotRequireHead() {
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(message("literal", record("bucket", "logs/a%2Cb%5B1%5D.csv", 123))));
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+
+ S3EventOffset offset = nextOffset();
+ Assertions.assertEquals(Collections.singletonList("logs/a,b[1].csv"), offset.getFiles());
+ Assertions.assertEquals("s3://bucket/logs/a,b[1].csv", offset.getFileStatuses().get(0).getPath());
+ Assertions.assertEquals(123, offset.getFileStatuses().get(0).getSize());
+ Assertions.assertFalse(offset.toSerializedJson().contains("fileStatuses"));
+ Mockito.verify(s3Client, Mockito.never()).headObject(Mockito.any(HeadObjectRequest.class));
+ Mockito.verify(sqsClient, Mockito.never()).deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class));
+ }
+
+ @Test
+ public void testNotificationMatchesFilesystemGlobSyntax() throws Exception {
+ tvfProperties.put("uri", "s3://bucket/logs/{001,abc,1..2}.csv");
+ provider = new S3EventSourceOffsetProvider(QUEUE_URL);
+ provider.ensureInitialized(1L, tvfProperties);
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("glob", "logs/001.csv", "logs/abc.csv",
+ "logs/1.csv", "logs/2.csv", "logs/3.csv", "logs//1.csv")));
+
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+
+ S3EventOffset offset = nextOffset();
+ Assertions.assertEquals(Arrays.asList("logs/001.csv", "logs/abc.csv",
+ "logs/1.csv", "logs/2.csv", "logs//1.csv"), offset.getFiles());
+ Assertions.assertEquals("s3://bucket/logs//1.csv", offset.getFileStatuses().get(4).getPath());
+ }
+
+ @Test
+ public void testNotificationPreservesRepeatedSlashesInObjectKey() throws Exception {
+ assertSelection("logs/*.csv", "logs//a.csv", true);
+ }
+
+ @Test
+ public void testNotificationSkipsNegativeRangeInMixedAlternatives() throws Exception {
+ assertSelection("logs/{x,-1..1}.csv", "logs/0.csv", false);
+ }
+
+ @Test
+ public void testNotificationPreservesSpacesInLiteralAlternatives() throws Exception {
+ assertSelection("logs/{1..2, abc }.csv", "logs/ abc .csv", true);
+ }
+
+ @Test
+ public void testNotificationUsesFilesystemCaretCharacterClass() throws Exception {
+ assertSelection("logs/[^a].csv", "logs/a.csv", true);
+ }
+
+ @Test
+ public void testQuestionMarkGlobPreservesMatchingNotificationsUntilCommit() throws Exception {
+ tvfProperties.put("uri", "s3://bucket/logs/file?.csv");
+ provider = new S3EventSourceOffsetProvider(QUEUE_URL);
+ provider.ensureInitialized(1L, tvfProperties);
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("keep", "logs/file1.csv"),
+ filesMessage("skip", "logs/file12.csv")), ReceiveMessageResponse.builder().build());
+
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+
+ Assertions.assertTrue(provider.hasMoreDataToConsume());
+ S3EventOffset offset = nextOffset();
+ Assertions.assertEquals(Collections.singletonList("logs/file1.csv"), offset.getFiles());
+ Assertions.assertEquals("s3://bucket/logs/file1.csv", offset.getFileStatuses().get(0).getPath());
+ ArgumentCaptor<DeleteMessageBatchRequest> deletes = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ Mockito.verify(sqsClient).deleteMessageBatch(deletes.capture());
+ Assertions.assertEquals(Collections.singletonList("skip"), deleteHandles(deletes.getValue()));
+
+ provider.updateOffset(offset);
+ provider.onTaskCommitted(1, 1);
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Mockito.verify(sqsClient, Mockito.times(2)).deleteMessageBatch(deletes.capture());
+ Assertions.assertEquals(Collections.singletonList("keep"), deleteHandles(deletes.getValue()));
+ }
+
+ @Test
+ public void testNotificationMatchesLiteralHashInSourceKey() throws Exception {
+ tvfProperties.put("uri", "s3://bucket/logs/file#1.csv");
+ provider = new S3EventSourceOffsetProvider(QUEUE_URL);
+ provider.ensureInitialized(1L, tvfProperties);
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("keep", "logs/file%231.csv")));
+
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+
+ Assertions.assertTrue(provider.hasMoreDataToConsume());
+ S3EventOffset offset = nextOffset();
+ Assertions.assertEquals(Collections.singletonList("logs/file#1.csv"), offset.getFiles());
+ Assertions.assertEquals("s3://bucket/logs/file#1.csv", offset.getFileStatuses().get(0).getPath());
+ Mockito.verify(sqsClient, Mockito.never()).deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class));
+ }
+
+ @Test
+ public void testDisplayTracksAcceptedAndCommittedBatches() {
+ Assertions.assertNull(provider.getShowCurrentOffset());
+ Assertions.assertNull(provider.getShowMaxOffset());
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("first", "logs/z.csv", "logs/a%22b.csv"),
+ filesMessage("overflow", "logs/overflow.csv")),
+ ReceiveMessageResponse.builder().build());
+ provider.fetchRemoteMeta(jobProperties(2), tvfProperties);
+ String endOffset = "{\"fileName\":\"logs/overflow.csv\",\"lagMessages\":0}";
+ Assertions.assertNull(provider.getShowCurrentOffset());
+ Assertions.assertEquals(endOffset, provider.getShowMaxOffset());
+ S3EventOffset offset = nextOffset();
+ Assertions.assertEquals(endOffset, provider.getShowMaxOffset());
+
+ Mockito.clearInvocations(sqsClient);
+ provider.updateOffset(provider.deserializeOffset(
+ provider.getCommitOffsetJson(offset, 1L, Collections.emptyList())));
+ provider.onTaskCommitted(3, 3);
+ Assertions.assertEquals("{\"fileName\":\"logs/overflow.csv\"}", provider.getShowCurrentOffset());
+ Assertions.assertEquals(endOffset, provider.getShowMaxOffset());
+ // Compact display must not truncate the durable file list or issue SQS calls.
+ Assertions.assertEquals("{\"files\":[\"logs/z.csv\",\"logs/a\\\"b.csv\",\"logs/overflow.csv\"]}",
+ provider.getPersistInfo());
+ Mockito.verifyNoInteractions(sqsClient);
+ provider.fetchRemoteMeta(jobProperties(2), tvfProperties);
+ Assertions.assertEquals(endOffset, provider.getShowMaxOffset());
+ }
+
+ @Test
+ public void testLagMessagesUsesVisibleCountAndSupportsLargeBacklogs() {
+ Assertions.assertEquals(-1L, provider.getLagMessages());
+ Map<QueueAttributeName, String> attributes = new HashMap<>();
+ attributes.put(QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES, "3000000000");
+ attributes.put(QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES_NOT_VISIBLE, "12");
+ attributes.put(QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES_DELAYED, "34");
+ Mockito.when(sqsClient.getQueueAttributes(Mockito.any(GetQueueAttributesRequest.class)))
+ .thenReturn(GetQueueAttributesResponse.builder().attributes(attributes).build());
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertNull(provider.getShowCurrentOffset());
+ Assertions.assertEquals("{\"lagMessages\":3000000000}", provider.getShowMaxOffset());
+ Assertions.assertEquals(3000000000L, provider.getLagMessages());
+ ArgumentCaptor<GetQueueAttributesRequest> request = ArgumentCaptor.forClass(GetQueueAttributesRequest.class);
+ Mockito.verify(sqsClient).getQueueAttributes(request.capture());
+ Assertions.assertEquals(QUEUE_URL, request.getValue().queueUrl());
+ Assertions.assertEquals(Collections.singletonList(QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES),
+ request.getValue().attributeNames());
+ }
+
+ @Test
+ public void testLagMessageFailureClearsStatisticWithoutBlockingBatch() {
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("first", "logs/a.csv")));
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertEquals("{\"fileName\":\"logs/a.csv\",\"lagMessages\":0}", provider.getShowMaxOffset());
+ Assertions.assertEquals(0L, provider.getLagMessages());
+ Mockito.when(sqsClient.getQueueAttributes(Mockito.any(GetQueueAttributesRequest.class)))
+ .thenThrow(new IllegalStateException("Access denied"))
+ .thenReturn(GetQueueAttributesResponse.builder().attributes(Collections.singletonMap(
+ QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES, "7")).build());
+
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertEquals("{\"fileName\":\"logs/a.csv\"}", provider.getShowMaxOffset());
+ Assertions.assertEquals(-1L, provider.getLagMessages());
+ Assertions.assertEquals(Collections.singletonList("logs/a.csv"), nextOffset().getFiles());
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertEquals("{\"fileName\":\"logs/a.csv\",\"lagMessages\":7}", provider.getShowMaxOffset());
+ Assertions.assertEquals(7L, provider.getLagMessages());
+ Mockito.verify(sqsClient, Mockito.times(1)).receiveMessage(Mockito.any(ReceiveMessageRequest.class));
+ }
+
+ @Test
+ public void testSourceEventTimestampTracksOnlyCommittedBatchAndIsNotPersisted() {
+ String latest = "2026-01-01T11:00:00.000Z";
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(message("first",
+ record("bucket", "logs/a.csv", 1, latest),
+ record("bucket", "logs/b.csv", 1, "2026-01-01T09:00:00.000Z"),
+ record("other", "logs/ignored.csv", 1, "2026-01-01T12:00:00.000Z"),
+ record("bucket", "other/ignored.csv", 1, "2026-01-01T12:00:00.000Z")),
+ message("second", record("bucket", "logs/c.csv", 1, "2026-01-01T11:30:00.000+02:00")),
+ message("overflow",
+ record("bucket", "logs/d.csv", 1, "2026-01-01T11:00:00.000Z"),
+ record("bucket", "logs/e.csv", 1, "2026-01-01T11:00:00.000Z"))));
+
+ Assertions.assertEquals(0L, provider.getLastSourceEventTimestampSeconds());
+ provider.fetchRemoteMeta(jobProperties(4), tvfProperties);
+ Assertions.assertEquals(0L, provider.getLastSourceEventTimestampSeconds());
+ S3EventOffset offset = nextOffset();
+ Assertions.assertEquals(0L, provider.getLastSourceEventTimestampSeconds());
+ String committedJson = provider.getCommitOffsetJson(offset, 1L, Collections.emptyList());
+ Assertions.assertEquals("{\"files\":[\"logs/a.csv\",\"logs/b.csv\",\"logs/c.csv\","
+ + "\"logs/d.csv\",\"logs/e.csv\"]}", committedJson);
+ provider.updateOffset(provider.deserializeOffset(committedJson));
+ provider.onTaskCommitted(5, 5);
+ Assertions.assertEquals(Instant.parse(latest).getEpochSecond(), provider.getLastSourceEventTimestampSeconds());
+ Mockito.verify(sqsClient, Mockito.never())
+ .changeMessageVisibilityBatch(Mockito.any(ChangeMessageVisibilityBatchRequest.class));
+
+ S3EventSourceOffsetProvider restored = new S3EventSourceOffsetProvider(QUEUE_URL);
+ restored.restoreFromPersistInfo(provider.getPersistInfo());
+ Assertions.assertEquals(provider.getShowCurrentOffset(), restored.getShowCurrentOffset());
+ Assertions.assertEquals(0L, restored.getLastSourceEventTimestampSeconds());
+
+ // A newly received batch must not replace the timestamp until it commits.
+ // Out-of-order notifications report that batch's time, not a historical maximum.
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("next", "logs/next.csv")));
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertEquals(Instant.parse(latest).getEpochSecond(), provider.getLastSourceEventTimestampSeconds());
+ provider.updateOffset(nextOffset());
+ provider.onTaskCommitted(1, 1);
+ Assertions.assertEquals(Instant.parse("2026-01-01T00:00:00Z").getEpochSecond(),
+ provider.getLastSourceEventTimestampSeconds());
+ }
+
+ @Test
+ public void testMissingSourceEventTimestampIsUnavailable() {
+ 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)));
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ provider.updateOffset(nextOffset());
+ provider.onTaskCommitted(1, 1);
+ Assertions.assertEquals(0L, provider.getLastSourceEventTimestampSeconds());
+ }
+
+ @Test
+ public void testFiltersRecordsAndDeletesOnlySkippedMessages() {
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(message("keep", record("bucket", "logs/a+b%2Bc.csv", 1),
+ record("other-bucket", "logs/wrong.csv", 1),
+ record("bucket", "other/wrong.csv", 1)),
+ Message.builder().receiptHandle("test-event")
+ .body("{\"Event\":\"s3:TestEvent\"}").build()),
+ ReceiveMessageResponse.builder().build());
+
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+
+ Assertions.assertEquals(Collections.singletonList("logs/a b+c.csv"), nextOffset().getFiles());
+ ArgumentCaptor<DeleteMessageBatchRequest> deleted = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ Mockito.verify(sqsClient).deleteMessageBatch(deleted.capture());
+ Assertions.assertEquals(Collections.singletonList("test-event"), deleteHandles(deleted.getValue()));
+ }
+
+ @Test
+ public void testDuplicateKeysAreScannedOnceAndAllReceiptsAreCommitted() {
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("first", "logs/a.csv", "logs/a.csv")),
+ response(filesMessage("second", "logs/a.csv", "logs/b.csv")));
+ provider.fetchRemoteMeta(jobProperties(2), tvfProperties);
+ S3EventOffset offset = nextOffset();
+ Assertions.assertEquals(Arrays.asList("logs/a.csv", "logs/b.csv"), offset.getFiles());
+ Assertions.assertEquals(2, offset.getFileStatuses().size());
+ Mockito.verify(sqsClient, Mockito.never()).deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class));
+ provider.updateOffset(provider.deserializeOffset(provider.getCommitOffsetJson(offset, 1L,
+ Collections.emptyList())));
+ provider.onTaskCommitted(2, 2);
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(ReceiveMessageResponse.builder().build());
+ provider.fetchRemoteMeta(jobProperties(2), tvfProperties);
+ ArgumentCaptor<DeleteMessageBatchRequest> deleted = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ Mockito.verify(sqsClient).deleteMessageBatch(deleted.capture());
+ Assertions.assertEquals(Arrays.asList("first", "second"), deleteHandles(deleted.getValue()));
+ }
+
+ @Test
+ public void testFileLimitAcceptsWholeReceive() {
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(message("first", record("bucket", "logs/a.csv", 1)),
+ message("overflow", record("bucket", "logs/b.csv", 1),
+ record("bucket", "logs/c.csv", 1))));
+
+ provider.fetchRemoteMeta(jobProperties(2), tvfProperties);
+
+ Assertions.assertEquals(Arrays.asList("logs/a.csv", "logs/b.csv", "logs/c.csv"), nextOffset().getFiles());
+ Mockito.verify(sqsClient, Mockito.never())
+ .changeMessageVisibilityBatch(Mockito.any(ChangeMessageVisibilityBatchRequest.class));
+ Mockito.verify(sqsClient, Mockito.never()).deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class));
+ }
+
+ @Test
+ public void testByteLimitAcceptsWholeReceive() {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("s3.max_batch_bytes", "104857600");
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(message("first", record("bucket", "logs/a.csv", 104857600)),
+ message("overflow", record("bucket", "logs/b.csv", 1))));
+
+ provider.fetchRemoteMeta(new StreamingJobProperties(properties), tvfProperties);
+
+ Assertions.assertEquals(Arrays.asList("logs/a.csv", "logs/b.csv"), nextOffset().getFiles());
+ Mockito.verify(sqsClient, Mockito.never())
+ .changeMessageVisibilityBatch(Mockito.any(ChangeMessageVisibilityBatchRequest.class));
+ Mockito.verify(sqsClient).receiveMessage(Mockito.any(ReceiveMessageRequest.class));
+ }
+
+ @Test
+ public void testSerializedLimitAcceptsWholeReceive() {
+ // Each key is below S3's key limit; each message is about 36 KiB of selected keys.
+ String[] firstKeys = longKeys("a", 40);
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("first", firstKeys),
+ filesMessage("overflow", longKeys("b", 40))));
+
+ provider.fetchRemoteMeta(jobProperties(256), tvfProperties);
+
+ S3EventOffset offset = nextOffset();
+ List<String> expected = new ArrayList<>(Arrays.asList(firstKeys));
+ expected.addAll(Arrays.asList(longKeys("b", 40)));
+ Assertions.assertEquals(expected, offset.getFiles());
+ Assertions.assertTrue(offset.serializedSize() > 64 * 1024);
+ Mockito.verify(sqsClient, Mockito.never())
+ .changeMessageVisibilityBatch(Mockito.any(ChangeMessageVisibilityBatchRequest.class));
+ }
+
+ @Test
+ public void testOversizedReceiveIsAcceptedWhole() {
+ String[] keys = longKeys("a", 80);
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("large", keys),
+ message("overflow", record("bucket", "logs/next.csv", 1))));
+
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+
+ S3EventOffset offset = nextOffset();
+ List<String> expected = new ArrayList<>(Arrays.asList(keys));
+ expected.add("logs/next.csv");
+ Assertions.assertEquals(expected, offset.getFiles());
+ Assertions.assertTrue(offset.serializedSize() > 64 * 1024);
+ Mockito.verify(sqsClient, Mockito.never())
+ .changeMessageVisibilityBatch(Mockito.any(ChangeMessageVisibilityBatchRequest.class));
+ }
+
+ @Test
+ public void testUnmatchedMessageIsDeletedWithoutAdvancingOffset() {
+ provider.restoreFromPersistInfo("{\"files\":[\"logs/committed.csv\"]}");
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("unmatched", "other/ignored.csv")),
+ ReceiveMessageResponse.builder().build());
+
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+ Assertions.assertEquals("{\"files\":[\"logs/committed.csv\"]}", provider.getPersistInfo());
+ ArgumentCaptor<DeleteMessageBatchRequest> deleted = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ Mockito.verify(sqsClient).deleteMessageBatch(deleted.capture());
+ Assertions.assertEquals(Collections.singletonList("unmatched"), deleteHandles(deleted.getValue()));
+ }
+
+ @Test
+ public void testInvalidMessagesAreDeletedWithoutPublishingBatch() {
+ 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\"}",
+ validEvent + " {}", validEvent + " garbage", unsupportedEvent)) {
+ Mockito.clearInvocations(sqsClient);
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(Message.builder().receiptHandle("invalid").body(body).build()),
+ ReceiveMessageResponse.builder().build());
+
+ provider.fetchRemoteMeta(jobProperties(2), tvfProperties);
+
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+ Assertions.assertNull(provider.getPersistInfo());
+ ArgumentCaptor<DeleteMessageBatchRequest> deleted = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ Mockito.verify(sqsClient).deleteMessageBatch(deleted.capture());
+ Assertions.assertEquals(Collections.singletonList("invalid"), deleteHandles(deleted.getValue()));
+ }
+ }
+
+ @Test
+ public void testInvalidRecordsDoNotDiscardValidFilesAndDeletionWaitsForCommit() {
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(message("mixed", record("bucket", "logs/a.csv", 1), "null",
+ record("bucket", "logs/deleted.csv", 1).replace("ObjectCreated:Put", "ObjectRemoved:Delete"),
+ "{\"eventSource\":\"aws:s3\",\"eventName\":\"ObjectCreated:Put\"}",
+ record("bucket", "", 1), record("bucket", "logs/invalid.csv", -1),
+ record("bucket", "logs/invalid%ZZ.csv", 1))),
+ ReceiveMessageResponse.builder().build());
+
+ provider.fetchRemoteMeta(jobProperties(2), tvfProperties);
+
+ S3EventOffset offset = nextOffset();
+ Assertions.assertEquals(Collections.singletonList("logs/a.csv"), offset.getFiles());
+ Mockito.verify(sqsClient, Mockito.never()).deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class));
+ provider.updateOffset(offset);
+ provider.onTaskCommitted(1, 1);
+ Mockito.verify(sqsClient, Mockito.never()).deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class));
+ provider.fetchRemoteMeta(jobProperties(2), tvfProperties);
+ ArgumentCaptor<DeleteMessageBatchRequest> deleted = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ Mockito.verify(sqsClient).deleteMessageBatch(deleted.capture());
+ Assertions.assertEquals(Collections.singletonList("mixed"), deleteHandles(deleted.getValue()));
+ }
+
+ @Test
+ public void testCommitDefersDeletionAndPartialFailureBlocksNextReceive() {
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(message("a", record("bucket", "logs/a.csv", 1)),
+ message("b", record("bucket", "logs/b.csv", 1))),
+ ReceiveMessageResponse.builder().build());
+ provider.fetchRemoteMeta(jobProperties(2), tvfProperties);
+ S3EventOffset offset = nextOffset();
+ Assertions.assertSame(offset, nextOffset());
+ Assertions.assertNull(provider.getPersistInfo());
+ Mockito.clearInvocations(sqsClient);
+
+ String committedJson = provider.getCommitOffsetJson(offset, 1L, Collections.emptyList());
+ provider.updateOffset(provider.deserializeOffset(committedJson));
+ provider.onTaskCommitted(2, 2);
+ Mockito.verifyNoInteractions(sqsClient);
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+ Assertions.assertEquals("{\"files\":[\"logs/a.csv\",\"logs/b.csv\"]}", provider.getPersistInfo());
+
+ Mockito.when(sqsClient.deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class)))
+ .thenReturn(DeleteMessageBatchResponse.builder().failed(BatchResultErrorEntry.builder()
+ .id("1").code("RequestThrottled").senderFault(false).build()).build(),
+ DeleteMessageBatchResponse.builder().build());
+ Assertions.assertThrows(IllegalStateException.class,
+ () -> provider.fetchRemoteMeta(jobProperties(2), tvfProperties));
+ Mockito.verify(sqsClient, Mockito.never()).receiveMessage(Mockito.any(ReceiveMessageRequest.class));
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+
+ provider.fetchRemoteMeta(jobProperties(2), tvfProperties);
+ ArgumentCaptor<DeleteMessageBatchRequest> deleted = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ InOrder order = Mockito.inOrder(sqsClient);
+ order.verify(sqsClient, Mockito.times(2)).deleteMessageBatch(deleted.capture());
+ order.verify(sqsClient).receiveMessage(Mockito.any(ReceiveMessageRequest.class));
+ Assertions.assertEquals(Arrays.asList("a", "b"), deleteHandles(deleted.getAllValues().get(0)));
+ Assertions.assertEquals(Collections.singletonList("b"), deleteHandles(deleted.getAllValues().get(1)));
+ }
+
+ @Test
+ public void testDeleteRequestExceptionRetainsUnattemptedChunks() {
+ Message[] first = IntStream.range(0, 10)
+ .mapToObj(i -> filesMessage("h" + i, "logs/" + i + ".csv")).toArray(Message[]::new);
+ Message[] second = IntStream.range(10, 13)
+ .mapToObj(i -> filesMessage("h" + i, "logs/" + i + ".csv")).toArray(Message[]::new);
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(first), response(second), ReceiveMessageResponse.builder().build());
+ provider.fetchRemoteMeta(jobProperties(13), tvfProperties);
+ Assertions.assertEquals(13, nextOffset().getFiles().size());
+ provider.updateOffset(nextOffset());
+ provider.onTaskCommitted(13, 13);
+ Mockito.clearInvocations(sqsClient);
+ RuntimeException failure = new IllegalStateException("connection failed");
+ Mockito.when(sqsClient.deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class)))
+ .thenThrow(failure).thenReturn(DeleteMessageBatchResponse.builder().build());
+
+ IllegalStateException thrown = Assertions.assertThrows(IllegalStateException.class,
+ () -> provider.fetchRemoteMeta(jobProperties(13), tvfProperties));
+ Assertions.assertSame(failure, thrown.getCause());
+ Mockito.verify(sqsClient, Mockito.never()).receiveMessage(Mockito.any(ReceiveMessageRequest.class));
+
+ provider.fetchRemoteMeta(jobProperties(13), tvfProperties);
+ ArgumentCaptor<DeleteMessageBatchRequest> deleted = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ Mockito.verify(sqsClient, Mockito.times(3)).deleteMessageBatch(deleted.capture());
+ Assertions.assertEquals(10, deleted.getAllValues().get(0).entries().size());
+ Assertions.assertEquals(deleteHandles(deleted.getAllValues().get(0)),
+ deleteHandles(deleted.getAllValues().get(1)));
+ Assertions.assertEquals(Arrays.asList("h10", "h11", "h12"), deleteHandles(deleted.getAllValues().get(2)));
+ }
+
+ @Test
+ public void testCommitCanProceedDuringRenewalAndExpiredReceiptIsNotDeleted() throws Exception {
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("old", "logs/a.csv")), ReceiveMessageResponse.builder().build());
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ S3EventOffset offset = nextOffset();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Mockito.when(sqsClient.changeMessageVisibilityBatch(Mockito.any(ChangeMessageVisibilityBatchRequest.class)))
+ .thenAnswer(invocation -> {
+ ChangeMessageVisibilityBatchRequest request = invocation.getArgument(0);
+ Assertions.assertTrue(request.entries().get(0).visibilityTimeout() > 10);
+ // Another thread must finish the commit while this SQS call is still in progress.
+ executor.submit(() -> {
+ Assertions.assertEquals("{\"files\":[\"logs/a.csv\"]}",
+ provider.getCommitOffsetJson(offset, 1L, Collections.emptyList()));
+ provider.updateOffset(offset);
+ provider.onTaskCommitted(1, 1);
+ }).get(5, TimeUnit.SECONDS);
+ return ChangeMessageVisibilityBatchResponse.builder().failed(BatchResultErrorEntry.builder()
+ .id("0").code("ReceiptHandleIsInvalid").senderFault(true).build()).build();
+ });
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Mockito.verify(sqsClient, Mockito.never()).deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class));
+ Mockito.verify(sqsClient, Mockito.times(2)).receiveMessage(Mockito.any(ReceiveMessageRequest.class));
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("c", "logs/c.csv")),
+ response(filesMessage("fresh-a", "logs/a.csv").toBuilder().messageId("id-old").build()),
+ ReceiveMessageResponse.builder().build());
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ provider.updateOffset(nextOffset());
+ provider.onTaskCommitted(1, 1);
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testResumeDiscardsReadyBatchAndReceivesFreshReceipt() {
+ assertResumeReceivesFreshReceipt(false);
+ }
+
+ @Test
+ public void testResumeDiscardsRunningBatchAndReceivesFreshReceipt() {
+ assertResumeReceivesFreshReceipt(true);
+ }
+
+ private void assertResumeReceivesFreshReceipt(boolean running) {
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("old", "logs/a.csv")),
+ response(filesMessage("fresh", "logs/a.csv")));
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ if (running) {
+ nextOffset();
+ }
+ provider.replayIfNeed(Mockito.mock(StreamingInsertJob.class));
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertEquals(Collections.singletonList("logs/a.csv"), nextOffset().getFiles());
+ provider.updateOffset(nextOffset());
+ provider.onTaskCommitted(1, 1);
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(ReceiveMessageResponse.builder().build());
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Mockito.verify(sqsClient, Mockito.never())
+ .changeMessageVisibilityBatch(Mockito.any(ChangeMessageVisibilityBatchRequest.class));
+ ArgumentCaptor<DeleteMessageBatchRequest> deleted = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ Mockito.verify(sqsClient).deleteMessageBatch(deleted.capture());
+ Assertions.assertEquals(Collections.singletonList("fresh"), deleteHandles(deleted.getValue()));
+ }
+
+ @Test
+ public void testExpiredReceiptPreventsCommit() {
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("old", "logs/a.csv"), filesMessage("valid", "logs/b.csv")));
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ S3EventOffset offset = nextOffset();
+ Mockito.when(sqsClient.changeMessageVisibilityBatch(Mockito.any(ChangeMessageVisibilityBatchRequest.class)))
+ .thenReturn(ChangeMessageVisibilityBatchResponse.builder().failed(BatchResultErrorEntry.builder()
+ .id("0").code("ReceiptHandleIsInvalid").build()).build());
+ Assertions.assertThrows(IllegalStateException.class,
+ () -> provider.fetchRemoteMeta(jobProperties(1), tvfProperties));
+ Assertions.assertThrows(IllegalStateException.class,
+ () -> provider.getCommitOffsetJson(offset, 1, Collections.emptyList()));
+ }
+
+ @Test
+ public void testExpiredCommittedReceiptSurvivesLaterCommit() {
+ provider.replayIfNeed(Mockito.mock(StreamingInsertJob.class));
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("b", "logs/b.csv")),
+ response(filesMessage("c", "logs/c.csv")),
+ response(filesMessage("fresh-b", "logs/b.csv").toBuilder().messageId("id-b").build()),
+ ReceiveMessageResponse.builder().build());
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ provider.updateOffset(nextOffset());
+ provider.onTaskCommitted(1, 1);
+ Mockito.when(sqsClient.deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class)))
+ .thenAnswer(invocation -> {
+ DeleteMessageBatchRequest request = invocation.getArgument(0);
+ if ("b".equals(request.entries().get(0).receiptHandle())) {
+ return DeleteMessageBatchResponse.builder().failed(BatchResultErrorEntry.builder()
+ .id("0").code("ReceiptHandleIsInvalid").build()).build();
+ }
+ return DeleteMessageBatchResponse.builder().successful(request.entries().stream()
+ .map(entry -> DeleteMessageBatchResultEntry.builder().id(entry.id()).build())
+ .collect(Collectors.toList())).build();
+ });
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertEquals(Collections.singletonList("logs/c.csv"), nextOffset().getFiles());
+ provider.updateOffset(nextOffset());
+ provider.onTaskCommitted(1, 1);
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+ Assertions.assertEquals("{\"files\":[\"logs/c.csv\"]}", provider.getPersistInfo());
+ }
+
+ @Test
+ public void testResumeAcknowledgesBatchWhoseCommitWasReplayed() {
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("committed", "logs/a.csv")),
+ ReceiveMessageResponse.builder().build());
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ S3EventOffset offset = nextOffset();
+ // Cloud replay can restore the commit before the task callback updates transient state.
+ provider.updateOffset(offset);
+ provider.replayIfNeed(Mockito.mock(StreamingInsertJob.class));
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertEquals("{\"files\":[\"logs/a.csv\"]}", provider.getPersistInfo());
+ ArgumentCaptor<DeleteMessageBatchRequest> deleted = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ Mockito.verify(sqsClient).deleteMessageBatch(deleted.capture());
+ Assertions.assertEquals(Collections.singletonList("committed"), deleteHandles(deleted.getValue()));
+ Mockito.verify(sqsClient, Mockito.never())
+ .changeMessageVisibilityBatch(Mockito.any(ChangeMessageVisibilityBatchRequest.class));
+ }
+
+ @Test
+ public void testPartialDeletionKeepsDuplicateKeyUntilEveryMessageIsAcknowledged() {
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("first", "logs/a.csv"),
+ filesMessage("second", "logs/a.csv", "logs/b.csv")),
+ response(filesMessage("c", "logs/c.csv")),
+ response(filesMessage("fresh", "logs/a.csv", "logs/b.csv")
+ .toBuilder().messageId("id-second").build()),
+ ReceiveMessageResponse.builder().build());
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertEquals(Arrays.asList("logs/a.csv", "logs/b.csv"), nextOffset().getFiles());
+ provider.updateOffset(nextOffset());
+ provider.onTaskCommitted(2, 2);
+ Mockito.when(sqsClient.deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class)))
+ .thenAnswer(invocation -> {
+ DeleteMessageBatchRequest request = invocation.getArgument(0);
+ if ("first".equals(request.entries().get(0).receiptHandle())) {
+ return DeleteMessageBatchResponse.builder()
+ .successful(DeleteMessageBatchResultEntry.builder().id("0").build())
+ .failed(BatchResultErrorEntry.builder().id("1")
+ .code("ReceiptHandleIsInvalid").build()).build();
+ }
+ return DeleteMessageBatchResponse.builder().successful(request.entries().stream()
+ .map(entry -> DeleteMessageBatchResultEntry.builder().id(entry.id()).build())
+ .collect(Collectors.toList())).build();
+ });
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertEquals(Collections.singletonList("logs/c.csv"), nextOffset().getFiles());
+ provider.updateOffset(nextOffset());
+ provider.onTaskCommitted(1, 1);
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+ // Successful acknowledgement removes the old keys rather than retaining them forever.
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("new-event", "logs/a.csv", "logs/b.csv")));
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertEquals(Arrays.asList("logs/a.csv", "logs/b.csv"), nextOffset().getFiles());
+ }
+
+ @Test
+ public void testInvalidVisibilityDurationDoesNotDiscardReceipt() {
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("running", "logs/a.csv")));
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ S3EventOffset offset = nextOffset();
+ Mockito.when(sqsClient.changeMessageVisibilityBatch(Mockito.any(ChangeMessageVisibilityBatchRequest.class)))
+ .thenReturn(ChangeMessageVisibilityBatchResponse.builder().failed(BatchResultErrorEntry.builder()
+ .id("0").code("InvalidParameterValue").message("Visibility timeout exceeds the maximum")
+ .senderFault(true).build()).build());
+
+ Assertions.assertThrows(IllegalStateException.class,
+ () -> provider.fetchRemoteMeta(jobProperties(1), tvfProperties));
+ Assertions.assertThrows(IllegalStateException.class,
+ () -> provider.fetchRemoteMeta(jobProperties(1), tvfProperties));
+ Assertions.assertSame(offset, nextOffset());
+ Mockito.verify(sqsClient, Mockito.times(2))
+ .changeMessageVisibilityBatch(Mockito.any(ChangeMessageVisibilityBatchRequest.class));
+ Mockito.verify(sqsClient).receiveMessage(Mockito.any(ReceiveMessageRequest.class));
+ }
+
+ @Test
+ public void testRecoveredFilesSurviveLaterCommitsAndResume() throws Exception {
+ StreamingInsertJob job = Mockito.mock(StreamingInsertJob.class);
+ provider.restoreFromPersistInfo("{\"files\":[\"logs/a.csv\"]}");
+ provider.replayIfNeed(job);
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("b", "logs/b.csv")),
+ response(filesMessage("c", "logs/c.csv")),
+ response(filesMessage("late-a", "logs/a.csv").toBuilder().messageId("id-a").build()),
+ ReceiveMessageResponse.builder().build());
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ provider.updateOffset(nextOffset());
+ provider.onTaskCommitted(1, 1);
+ provider.replayIfNeed(job);
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ provider.updateOffset(nextOffset());
+ provider.onTaskCommitted(1, 1);
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+ Assertions.assertEquals("{\"files\":[\"logs/c.csv\"]}", provider.getPersistInfo());
+ Assertions.assertEquals("{\"fileName\":\"logs/c.csv\"}", provider.getShowCurrentOffset());
+ ArgumentCaptor<DeleteMessageBatchRequest> deleted = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ Mockito.verify(sqsClient, Mockito.times(3)).deleteMessageBatch(deleted.capture());
+ Assertions.assertEquals(Collections.singletonList("late-a"), deleteHandles(deleted.getAllValues().get(2)));
+
+ provider.replayIfNeed(job);
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("overwrite-a", "logs/a.csv")));
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertEquals(Collections.singletonList("logs/a.csv"), nextOffset().getFiles());
+ }
+
+ @Test
+ public void testRecoverySnapshotUsesLatestReplayedOffsetAndDoesNotAccumulate() {
+ provider.restoreFromPersistInfo("{\"files\":[\"logs/image.csv\"]}");
+ // Transaction replay advances the image offset before the first scheduling pass.
+ provider.updateOffset(new S3EventOffset(Collections.singletonList("logs/a.csv")));
+ provider.replayIfNeed(Mockito.mock(StreamingInsertJob.class));
+ provider.updateOffset(new S3EventOffset(Collections.singletonList("logs/b.csv")));
+ provider.replayIfNeed(Mockito.mock(StreamingInsertJob.class));
+ provider.updateOffset(new S3EventOffset(Collections.singletonList("logs/c.csv")));
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("mixed", "logs/image.csv", "logs/a.csv",
+ "logs/b.csv", "logs/c.csv", "logs/new.csv")));
+
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+
+ Assertions.assertEquals(Arrays.asList("logs/image.csv", "logs/b.csv", "logs/new.csv"),
+ nextOffset().getFiles());
+ Assertions.assertEquals(3, nextOffset().getFileStatuses().size());
+ Mockito.verify(sqsClient, Mockito.never()).deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class));
+ Map<String, Map<String, String>> recovered = Deencapsulation.getField(provider, "unackedCommittedFiles");
+ Assertions.assertEquals(Collections.singletonMap("id-mixed", "mixed"), recovered.get("logs/a.csv"));
+ provider.updateOffset(nextOffset());
+ provider.onTaskCommitted(3, 3);
+ Assertions.assertEquals(Collections.singletonMap("id-mixed", "mixed"), recovered.get("logs/a.csv"));
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(ReceiveMessageResponse.builder().build());
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ ArgumentCaptor<DeleteMessageBatchRequest> deleted = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ Mockito.verify(sqsClient).deleteMessageBatch(deleted.capture());
+ Assertions.assertEquals(Collections.singletonList("mixed"), deleteHandles(deleted.getValue()));
+ Assertions.assertTrue(recovered.isEmpty());
+ }
+
+ @Test
+ public void testRecoveredFilesAreRemovedOnlyAfterSuccessfulDeletion() {
+ provider.restoreFromPersistInfo("{\"files\":[\"logs/a.csv\",\"logs/b.csv\",\"logs/unseen.csv\"]}");
+ provider.replayIfNeed(Mockito.mock(StreamingInsertJob.class));
+ provider.updateOffset(new S3EventOffset(Collections.singletonList("logs/current.csv")));
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("a", "logs/a.csv"), filesMessage("b", "logs/b.csv")));
+ Mockito.when(sqsClient.deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class)))
+ .thenReturn(DeleteMessageBatchResponse.builder()
+ .successful(DeleteMessageBatchResultEntry.builder().id("0").build())
+ .failed(BatchResultErrorEntry.builder().id("1").code("RequestThrottled").build()).build());
+
+ Assertions.assertThrows(IllegalStateException.class,
+ () -> provider.fetchRemoteMeta(jobProperties(1), tvfProperties));
+ Map<String, Map<String, String>> recovered = Deencapsulation.getField(provider, "unackedCommittedFiles");
+ Assertions.assertFalse(recovered.containsKey("logs/a.csv"));
+ Assertions.assertEquals(Collections.singletonMap("id-b", "b"), recovered.get("logs/b.csv"));
+ Assertions.assertTrue(recovered.containsKey("logs/unseen.csv"));
+
+ Mockito.when(sqsClient.deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class)))
+ .thenThrow(new IllegalStateException("delete failed"))
+ .thenReturn(DeleteMessageBatchResponse.builder()
+ .successful(DeleteMessageBatchResultEntry.builder().id("0").build()).build());
+ Assertions.assertThrows(IllegalStateException.class,
+ () -> provider.fetchRemoteMeta(jobProperties(1), tvfProperties));
+ Assertions.assertEquals(Collections.singletonMap("id-b", "b"), recovered.get("logs/b.csv"));
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(ReceiveMessageResponse.builder().build());
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+ Assertions.assertFalse(recovered.containsKey("logs/b.csv"));
+ Assertions.assertEquals(Collections.singleton("logs/unseen.csv"), recovered.keySet());
+ Assertions.assertEquals("{\"files\":[\"logs/current.csv\"]}", provider.getPersistInfo());
+ }
+
+ @Test
+ public void testFreshJobDoesNotCaptureLaterCommitAsRecoverySnapshot() {
+ StreamingInsertJob job = Mockito.mock(StreamingInsertJob.class);
+ provider.replayIfNeed(job);
+ provider.updateOffset(new S3EventOffset(Collections.singletonList("logs/a.csv")));
+ provider.replayIfNeed(job);
+ provider.updateOffset(new S3EventOffset(Collections.singletonList("logs/b.csv")));
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("old", "logs/a.csv")));
+
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+
+ Assertions.assertEquals(Collections.singletonList("logs/a.csv"), nextOffset().getFiles());
+ }
+
+ @Test
+ public void testRestoreKeepsCommittedFilesButNotReceiptHandles() {
+ provider.restoreFromPersistInfo("{\"files\":[\"logs/committed.csv\"]}");
+ StreamingInsertJob job = Mockito.mock(StreamingInsertJob.class);
+ Mockito.when(job.getOffsetProviderPersist()).thenReturn("{\"files\":[\"logs/older.csv\"]}");
+ provider.replayIfNeed(job);
+ Assertions.assertFalse(provider.hasMoreDataToConsume());
+ Assertions.assertEquals("{\"files\":[\"logs/committed.csv\"]}", provider.getPersistInfo());
+ Assertions.assertEquals("{\"fileName\":\"logs/committed.csv\"}", provider.getShowCurrentOffset());
+ Assertions.assertEquals("{\"fileName\":\"logs/committed.csv\"}", provider.getShowMaxOffset());
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("new-handle", "logs/committed.csv"),
+ filesMessage("next", "logs/next.csv")));
+
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+
+ Assertions.assertEquals(Collections.singletonList("logs/next.csv"), nextOffset().getFiles());
+ ArgumentCaptor<DeleteMessageBatchRequest> deleted = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ InOrder order = Mockito.inOrder(sqsClient);
+ order.verify(sqsClient).receiveMessage(Mockito.any(ReceiveMessageRequest.class));
+ order.verify(sqsClient).deleteMessageBatch(deleted.capture());
+ Assertions.assertEquals(Collections.singletonList("new-handle"), deleteHandles(deleted.getValue()));
+ }
+
+ @Test
+ public void testSqsUsesIamRoleWithoutStaticKeys() throws Exception {
+ tvfProperties.remove("s3.access_key");
+ tvfProperties.remove("s3.secret_key");
+ tvfProperties.put("s3.role_arn", "arn:aws:iam::123456789012:role/event-reader");
+ tvfProperties.put("s3.external_id", "event-external-id");
+ tvfProperties.put("s3.credentials_provider_type", "instance_profile");
+ StsClient stsClient = Mockito.mock(StsClient.class);
+ StsClientBuilder stsBuilder = Mockito.mock(StsClientBuilder.class, Mockito.RETURNS_SELF);
+ Mockito.when(stsBuilder.build()).thenReturn(stsClient);
+ Mockito.when(stsClient.assumeRole(Mockito.any(AssumeRoleRequest.class)))
+ .thenReturn(AssumeRoleResponse.builder().credentials(Credentials.builder()
+ .accessKeyId("role-ak").secretAccessKey("role-sk").sessionToken("role-token")
+ .expiration(Instant.now().plusSeconds(3600)).build()).build());
+ try (MockedStatic<StsClient> stsStatic = Mockito.mockStatic(StsClient.class)) {
+ stsStatic.when(StsClient::builder).thenReturn(stsBuilder);
+ provider = new S3EventSourceOffsetProvider(QUEUE_URL);
+ provider.ensureInitialized(1L, tvfProperties);
+ provider.fetchRemoteMeta(jobProperties(1), tvfProperties);
+
+ ArgumentCaptor<AwsCredentialsProvider> credentials = ArgumentCaptor.forClass(AwsCredentialsProvider.class);
+ Mockito.verify(sqsBuilder).credentialsProvider(
+ (IdentityProvider<? extends AwsCredentialsIdentity>) credentials.capture());
+ // Keep the real role provider: only STS and SQS network boundaries are mocked.
+ try (StsAssumeRoleCredentialsProvider roleProvider =
+ (StsAssumeRoleCredentialsProvider) credentials.getValue()) {
+ AwsSessionCredentials resolved = (AwsSessionCredentials) roleProvider.resolveCredentials();
+ Assertions.assertEquals("role-ak", resolved.accessKeyId());
+ Assertions.assertEquals("role-sk", resolved.secretAccessKey());
+ Assertions.assertEquals("role-token", resolved.sessionToken());
+ }
+ ArgumentCaptor<AssumeRoleRequest> assumed = ArgumentCaptor.forClass(AssumeRoleRequest.class);
+ Mockito.verify(stsClient).assumeRole(assumed.capture());
+ Assertions.assertEquals("arn:aws:iam::123456789012:role/event-reader", assumed.getValue().roleArn());
+ Assertions.assertEquals("event-external-id", assumed.getValue().externalId());
+ Mockito.verify(sqsBuilder).region(Region.US_WEST_2);
+ Mockito.verify(sqsClient, Mockito.never()).close();
+ }
+ }
+
+ private void assertSelection(String pattern, String key, boolean expected) throws Exception {
+ provider.close();
+ tvfProperties.put("uri", "s3://bucket/" + pattern);
+ provider = new S3EventSourceOffsetProvider(QUEUE_URL);
+ provider.ensureInitialized(1L, tvfProperties);
+ Mockito.when(sqsClient.receiveMessage(Mockito.any(ReceiveMessageRequest.class)))
+ .thenReturn(response(filesMessage("selection", key)), ReceiveMessageResponse.builder().build());
+
+ provider.fetchRemoteMeta(jobProperties(256), tvfProperties);
+
+ Assertions.assertEquals(expected, provider.hasMoreDataToConsume());
+ if (expected) {
+ S3EventOffset offset = nextOffset();
+ Assertions.assertEquals(Collections.singletonList(key), offset.getFiles());
+ Assertions.assertEquals("s3://bucket/" + key, offset.getFileStatuses().get(0).getPath());
+ Mockito.verify(sqsClient, Mockito.never()).deleteMessageBatch(Mockito.any(DeleteMessageBatchRequest.class));
+ } else {
+ ArgumentCaptor<DeleteMessageBatchRequest> deletes = ArgumentCaptor.forClass(DeleteMessageBatchRequest.class);
+ Mockito.verify(sqsClient).deleteMessageBatch(deletes.capture());
+ Assertions.assertEquals(Collections.singletonList("selection"), deleteHandles(deletes.getValue()));
+ }
+ }
+
+ private StreamingJobProperties jobProperties(int maxFiles) {
+ Map<String, String> properties = new HashMap<>();
+ properties.put("s3.max_batch_files", String.valueOf(maxFiles));
+ return new StreamingJobProperties(properties);
+ }
+
+ private S3EventOffset nextOffset() {
+ return provider.getNextOffset(jobProperties(256), tvfProperties);
+ }
+
+ private static List<String> deleteHandles(DeleteMessageBatchRequest request) {
+ Assertions.assertEquals(QUEUE_URL, request.queueUrl());
+ return request.entries().stream().map(entry -> entry.receiptHandle()).collect(Collectors.toList());
+ }
+
+ private static ReceiveMessageResponse response(Message... messages) {
+ return ReceiveMessageResponse.builder().messages(messages).build();
+ }
+
+ private static Message message(String handle, String... records) {
+ return Message.builder().messageId("id-" + handle).receiptHandle(handle)
+ .body("{\"Records\":[" + String.join(",", records) + "]}").build();
+ }
+
+ private static Message filesMessage(String handle, String... keys) {
+ return message(handle, Arrays.stream(keys).map(key -> record("bucket", key, 1)).toArray(String[]::new));
+ }
+
+ private static String[] longKeys(String prefix, int count) {
+ return IntStream.range(0, count).mapToObj(i -> "logs/" + prefix + i + StringUtils.repeat('x', 900) + ".csv")
+ .toArray(String[]::new);
+ }
+
+ private static String record(String bucket, String encodedKey, long size) {
+ return record(bucket, encodedKey, size, "2026-01-01T00:00:00.000Z");
+ }
+
+ private static String record(String bucket, String encodedKey, long size, String eventTime) {
+ return "{\"eventVersion\":\"2.1\",\"eventSource\":\"aws:s3\",\"awsRegion\":\"us-west-2\","
+ + "\"eventTime\":" + GsonUtils.GSON.toJson(eventTime) + ",\"eventName\":\"ObjectCreated:Put\","
+ + "\"userIdentity\":{\"principalId\":\"test\"},"
+ + "\"requestParameters\":{\"sourceIPAddress\":\"127.0.0.1\"},"
+ + "\"responseElements\":{\"x-amz-request-id\":\"request\",\"x-amz-id-2\":\"host\"},"
+ + "\"s3\":{\"s3SchemaVersion\":\"1.0\",\"configurationId\":\"test\","
+ + "\"bucket\":{\"name\":" + GsonUtils.GSON.toJson(bucket)
+ + ",\"ownerIdentity\":{\"principalId\":\"owner\"},\"arn\":\"arn:aws:s3:::" + bucket + "\"},"
+ + "\"object\":{\"key\":" + GsonUtils.GSON.toJson(encodedKey) + ",\"size\":" + size
+ + ",\"eTag\":\"etag\",\"sequencer\":\"001\"}}}";
+ }
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java
index a2f75c9aae7..897537366a5 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/job/offset/s3/S3SourceOffsetProviderTest.java
@@ -18,13 +18,17 @@
package org.apache.doris.job.offset.s3;
import org.apache.doris.backup.Status;
+import org.apache.doris.common.util.S3URI;
+import org.apache.doris.common.util.S3Util;
import org.apache.doris.datasource.property.storage.StorageProperties;
import org.apache.doris.fs.FileSystemFactory;
import org.apache.doris.fs.GlobListResult;
import org.apache.doris.fs.remote.RemoteFile;
import org.apache.doris.fs.remote.RemoteFileSystem;
import org.apache.doris.job.extensions.insert.streaming.StreamingJobProperties;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -32,15 +36,21 @@ import org.mockito.MockedStatic;
import org.mockito.Mockito;
import org.mockito.stubbing.Answer;
+import java.nio.file.FileSystems;
+import java.nio.file.PathMatcher;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
public class S3SourceOffsetProviderTest {
private static final Map<String, String> TVF_PROPS = ImmutableMap.of(
"uri", "s3://bucket/data/*.csv", "s3.endpoint", "s3.us-east-1.amazonaws.com",
"s3.region", "us-east-1", "s3.access_key", "ak", "s3.secret_key", "sk");
private static final StreamingJobProperties ONCE_PROPS = new StreamingJobProperties(
- ImmutableMap.of("s3.ingestion_mode", "ONCE", "s3.max_batch_files", "1"));
+ ImmutableMap.of("s3.ingestion_mode", "ONE_TIME", "s3.max_batch_files", "1"));
@Test
public void testOnceMarksLastBatchAndRecovers() throws Exception {
@@ -68,6 +78,10 @@ public class S3SourceOffsetProviderTest {
Assertions.assertTrue(provider.hasMoreDataToConsume());
S3Offset first = provider.getNextOffset(ONCE_PROPS, TVF_PROPS);
Assertions.assertEquals("data/a.csv", first.getEndFile());
+ Assertions.assertEquals("s3://bucket/data/a.csv", first.getFileStatuses().get(0).getPath());
+ Assertions.assertFalse(first.toSerializedJson().contains("fileStatuses"));
+ Assertions.assertFalse(first.isEmpty());
+ Assertions.assertTrue(provider.deserializeOffset(first.toSerializedJson()).isEmpty());
Assertions.assertFalse(first.isLastBatch());
Assertions.assertEquals("data/a.csv", provider.getNextOffset(ONCE_PROPS, TVF_PROPS).getEndFile());
provider.updateOffset(provider.deserializeOffset(first.toSerializedJson()));
@@ -148,4 +162,55 @@ public class S3SourceOffsetProviderTest {
return new GlobListResult(Status.OK, maxFile, "bucket", "data/");
};
}
+
+ @Test
+ void testOrdinaryFileListIsUnchanged() {
+ Assertions.assertEquals("s3://bucket/prefix/{001,normal.csv}",
+ S3SourceOffsetProvider.buildFileListUri(statuses("prefix/", Arrays.asList("001", "normal.csv"))));
+ }
+
+ @Test
+ void testFileListsAtBucketRootAndAcrossDirectories() {
+ Assertions.assertEquals("s3://bucket/{a.csv,b.csv}",
+ S3SourceOffsetProvider.buildFileListUri(statuses("", ImmutableList.of("a.csv", "b.csv"))));
+ Assertions.assertEquals("s3://bucket/logs/{a/1.csv,b/2.csv}",
+ S3SourceOffsetProvider.buildFileListUri(statuses("logs/", ImmutableList.of("a/1.csv", "b/2.csv"))));
+ Assertions.assertEquals("s3://bucket/logs/a/{1.csv}",
+ S3SourceOffsetProvider.buildFileListUri(statuses("logs/a/", ImmutableList.of("1.csv"))));
+ }
+
+ @Test
+ void testCommaFileDoesNotSelectOtherObjects() throws Exception {
+ assertLiteralFiles("prefix/", Arrays.asList("comma,a.csv", "normal.csv"),
+ Arrays.asList("prefix/comma", "prefix/a.csv", "prefix/comma,a.csv.bak"));
+ assertLiteralFiles("prefix/", Collections.singletonList("comma,a.csv"),
+ Arrays.asList("prefix/comma", "prefix/a.csv"));
+ }
+
+ @Test
+ void testLiteralGlobCharactersInDirectoriesAndFiles() throws Exception {
+ assertLiteralFiles("prefix/[dir],{part}/",
+ Arrays.asList("a[b].csv", "a{b,c}.csv", "a*.csv", "a\\b.csv", "$1.csv", "001"),
+ Arrays.asList("prefix/[dir],{part}/ab.csv", "prefix/[dir],{part}/other.csv",
+ "prefix/[dir],{part}/1", "prefix/d,part/001"));
+ }
+
+ private static List<TBrokerFileStatus> statuses(String prefix, List<String> files) {
+ return files.stream().map(file -> new TBrokerFileStatus("s3://bucket/" + prefix + file, false, 10, true))
+ .collect(Collectors.toList());
+ }
+
+ private void assertLiteralFiles(String prefix, List<String> files, List<String> otherKeys) throws Exception {
+ // Check the audit URI with the LIST matcher; execution uses the original file statuses.
+ String uri = S3SourceOffsetProvider.buildFileListUri(statuses(prefix, files));
+ String keyPattern = S3URI.create(uri).getKey();
+ PathMatcher matcher = FileSystems.getDefault()
+ .getPathMatcher("glob:" + Paths.get(S3Util.extendGlobs(keyPattern)));
+ for (String file : files) {
+ Assertions.assertTrue(matcher.matches(Paths.get(prefix + file)), "Missing file: " + file);
+ }
+ for (String key : otherKeys) {
+ Assertions.assertFalse(matcher.matches(Paths.get(key)), "Unexpected file: " + key);
+ }
+ }
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/metric/MetricsTest.java b/fe/fe-core/src/test/java/org/apache/doris/metric/MetricsTest.java
index 737647e1f84..f7359fa7843 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/metric/MetricsTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/metric/MetricsTest.java
@@ -33,6 +33,7 @@ import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob;
import org.apache.doris.job.manager.JobManager;
import org.apache.doris.job.offset.jdbc.JdbcOffset;
import org.apache.doris.job.offset.jdbc.JdbcSourceOffsetProvider;
+import org.apache.doris.job.offset.s3.S3EventSourceOffsetProvider;
import org.apache.doris.metric.Metric.MetricUnit;
import org.apache.doris.monitor.jvm.JvmService;
import org.apache.doris.monitor.jvm.JvmStats;
@@ -191,6 +192,34 @@ public class MetricsTest {
"doris_fe_streaming_job_per_job_last_task_success_time_seconds"
+ "{job_id=\"1787039821000\", job_name=\"streaming_metric_job\"} 1787039821"));
Assert.assertFalse(metricResult.contains("doris_fe_streaming_job_per_job_lag{"));
+ String messageLag = "doris_fe_streaming_job_per_job_lag_messages"
+ + "{job_id=\"1787039821000\", job_name=\"streaming_metric_job\"}";
+ Assert.assertFalse(metricResult.contains(messageLag));
+
+ S3EventSourceOffsetProvider eventProvider = new S3EventSourceOffsetProvider("test-queue");
+ Deencapsulation.setField(job, "offsetProvider", eventProvider);
+ MetricRepo.updateStreamingJobPerJobMetrics();
+ Assert.assertFalse(getPrometheusMetrics().contains(messageLag));
+
+ Deencapsulation.setField(eventProvider, "approximateNumberOfMessages", 12L);
+ Deencapsulation.setField(eventProvider, "lastSourceEventTimestampSeconds", 1787039800L);
+ MetricRepo.updateStreamingJobPerJobMetrics();
+ metricResult = getPrometheusMetrics();
+ Assert.assertTrue(metricResult.contains(messageLag + " 12"));
+ Assert.assertTrue(metricResult.contains(
+ "doris_fe_streaming_job_per_job_last_source_event_timestamp_seconds"
+ + "{job_id=\"1787039821000\", job_name=\"streaming_metric_job\"} 1787039800"));
+
+ Deencapsulation.setField(eventProvider, "approximateNumberOfMessages", null);
+ MetricRepo.updateStreamingJobPerJobMetrics();
+ Assert.assertFalse(getPrometheusMetrics().contains(messageLag));
+
+ Deencapsulation.setField(eventProvider, "approximateNumberOfMessages", 0L);
+ MetricRepo.updateStreamingJobPerJobMetrics();
+ Assert.assertTrue(getPrometheusMetrics().contains(messageLag + " 0"));
+ jobMap.remove(job.getJobId());
+ MetricRepo.updateStreamingJobPerJobMetrics();
+ Assert.assertFalse(getPrometheusMetrics().contains(messageLag));
} finally {
jobMap.remove(job.getJobId());
MetricRepo.updateStreamingJobPerJobMetrics();
diff --git a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/S3TableValuedFunctionTest.java b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/S3TableValuedFunctionTest.java
new file mode 100644
index 00000000000..2d79b5a776a
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/S3TableValuedFunctionTest.java
@@ -0,0 +1,147 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.tablefunction;
+
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.FeConstants;
+import org.apache.doris.common.util.S3Util;
+import org.apache.doris.fs.FileSystemFactory;
+import org.apache.doris.nereids.StatementContext;
+import org.apache.doris.nereids.trees.expressions.Properties;
+import org.apache.doris.nereids.trees.expressions.functions.table.S3;
+import org.apache.doris.proto.InternalService.PFetchTableSchemaResult;
+import org.apache.doris.proto.Types.PStatus;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.rpc.BackendServiceProxy;
+import org.apache.doris.system.Backend;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TStatusCode;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public class S3TableValuedFunctionTest {
+ private static final Map<String, String> PROPERTIES = ImmutableMap.of(
+ "uri", "s3://bucket/logs/*", "format", "csv", "s3.endpoint", "s3.us-west-2.amazonaws.com",
+ "s3.region", "us-west-2", "s3.access_key", "ak", "s3.secret_key", "sk");
+ private boolean runningUnitTest;
+ private ConnectContext previousContext;
+
+ @BeforeEach
+ public void setUp() {
+ runningUnitTest = FeConstants.runningUnitTest;
+ FeConstants.runningUnitTest = true;
+ previousContext = ConnectContext.get();
+ ConnectContext ctx = new ConnectContext();
+ ctx.setStatementContext(new StatementContext());
+ ctx.setThreadLocalInfo();
+ }
+
+ @AfterEach
+ public void tearDown() {
+ FeConstants.runningUnitTest = runningUnitTest;
+ ConnectContext.remove();
+ if (previousContext != null) {
+ previousContext.setThreadLocalInfo();
+ }
+ }
+
+ @Test
+ public void testExactFilesKeepLiteralPathsAndSkipListing() throws Exception {
+ FeConstants.runningUnitTest = false;
+ TBrokerFileStatus file = file("a,b[1]*?.csv", 123);
+ try (MockedStatic<S3Util> s3 = Mockito.mockStatic(S3Util.class);
+ MockedStatic<FileSystemFactory> filesystems = Mockito.mockStatic(FileSystemFactory.class)) {
+ S3TableValuedFunction tvf = new S3TableValuedFunction(PROPERTIES, Collections.singletonList(file));
+ Assertions.assertEquals(Collections.singletonList(file), tvf.getFileStatuses());
+ s3.verify(() -> S3Util.validateAndTestEndpoint("s3.us-west-2.amazonaws.com"));
+ filesystems.verifyNoInteractions();
+ }
+ }
+
+ @Test
+ public void testSchemaFailureDoesNotSkipFiles() throws Exception {
+ assertSchemaFailure(TStatusCode.NOT_FOUND);
+ assertSchemaFailure(TStatusCode.INTERNAL_ERROR);
+ }
+
+ @Test
+ public void testExactFilesAreScopedToTvf() throws Exception {
+ S3TableValuedFunction first = new S3TableValuedFunction(PROPERTIES,
+ Collections.singletonList(file("a.csv", 10)));
+ S3TableValuedFunction second = new S3TableValuedFunction(PROPERTIES,
+ Collections.singletonList(file("b.csv", 20)));
+ Assertions.assertEquals("s3://bucket/logs/a.csv", first.getFileStatuses().get(0).getPath());
+ Assertions.assertEquals("s3://bucket/logs/b.csv", second.getFileStatuses().get(0).getPath());
+ Assertions.assertTrue(new S3TableValuedFunction(PROPERTIES).getFileStatuses().isEmpty());
+ }
+
+ @Test
+ public void testExactFilesParticipateInExpressionEquality() {
+ S3 first = S3.withFiles(new Properties(PROPERTIES), Collections.singletonList(file("a.csv", 10)));
+ S3 same = S3.withFiles(new Properties(PROPERTIES), Collections.singletonList(file("a.csv", 10)));
+ S3 other = S3.withFiles(new Properties(PROPERTIES), Collections.singletonList(file("b.csv", 10)));
+ Assertions.assertEquals(first, same);
+ Assertions.assertEquals(first.hashCode(), same.hashCode());
+ Assertions.assertNotEquals(first, other);
+ Assertions.assertNotEquals(first, new S3(new Properties(PROPERTIES)));
+ }
+
+ private void assertSchemaFailure(TStatusCode code) throws Exception {
+ S3TableValuedFunction tvf = tvf();
+ BackendServiceProxy proxy = Mockito.mock(BackendServiceProxy.class);
+ Mockito.when(proxy.fetchTableStructureAsync(Mockito.any(), Mockito.any()))
+ .thenReturn(CompletableFuture.completedFuture(result(code)));
+ try (MockedStatic<BackendServiceProxy> service = Mockito.mockStatic(BackendServiceProxy.class)) {
+ service.when(BackendServiceProxy::getInstance).thenReturn(proxy);
+ Assertions.assertThrows(AnalysisException.class, tvf::getTableColumns);
+ Mockito.verify(proxy).fetchTableStructureAsync(Mockito.any(), Mockito.any());
+ Assertions.assertEquals(2, tvf.getFileStatuses().size());
+ }
+ }
+
+ private S3TableValuedFunction tvf() throws Exception {
+ S3TableValuedFunction tvf = Mockito.spy(new S3TableValuedFunction(PROPERTIES,
+ Arrays.asList(file("a.csv", 10), file("b.csv", 10))));
+ Backend backend = Mockito.mock(Backend.class);
+ Mockito.when(backend.getHost()).thenReturn("127.0.0.1");
+ Mockito.when(backend.getBrpcPort()).thenReturn(8060);
+ Mockito.doReturn(backend).when(tvf).getBackend();
+ return tvf;
+ }
+
+ private static TBrokerFileStatus file(String key, long size) {
+ return new TBrokerFileStatus("s3://bucket/logs/" + key, false, size, true);
+ }
+
+ private static PFetchTableSchemaResult result(TStatusCode code) {
+ return PFetchTableSchemaResult.newBuilder()
+ .setStatus(PStatus.newBuilder().setStatusCode(code.getValue()).addErrorMsgs("schema unavailable"))
+ .setColumnNums(0).build();
+ }
+}
diff --git a/regression-test/data/job_p0/streaming_job/test_streaming_insert_job_s3_event_aksk.out b/regression-test/data/job_p0/streaming_job/test_streaming_insert_job_s3_event_aksk.out
new file mode 100644
index 00000000000..429651b0252
--- /dev/null
+++ b/regression-test/data/job_p0/streaming_job/test_streaming_insert_job_s3_event_aksk.out
@@ -0,0 +1,20 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !first --
+1 first
+
+-- !second --
+1 first
+2 second
+
+-- !final --
+1 first 1
+2 second 1
+3 third 1
+4 fourth 1
+
+-- !event_comma --
+1 first
+2 second
+3 third
+4 fourth
+5 comma
diff --git a/regression-test/data/job_p0/streaming_job/test_streaming_insert_job_s3_event_iam_role.out b/regression-test/data/job_p0/streaming_job/test_streaming_insert_job_s3_event_iam_role.out
new file mode 100644
index 00000000000..5cbddfb9f10
--- /dev/null
+++ b/regression-test/data/job_p0/streaming_job/test_streaming_insert_job_s3_event_iam_role.out
@@ -0,0 +1,7 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !first --
+1 role_first
+
+-- !final --
+1 role_first 1
+2 role_second 1
diff --git a/regression-test/data/job_p0/streaming_job/test_streaming_insert_job_s3_event_restart_fe.out b/regression-test/data/job_p0/streaming_job/test_streaming_insert_job_s3_event_restart_fe.out
new file mode 100644
index 00000000000..017b53fba42
--- /dev/null
+++ b/regression-test/data/job_p0/streaming_job/test_streaming_insert_job_s3_event_restart_fe.out
@@ -0,0 +1,14 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !before --
+1 before_restart
+2 before_restart
+
+-- !restored --
+1 before_restart
+2 before_restart
+
+-- !final --
+1 before_restart 1
+2 before_restart 1
+3 after_restart 1
+4 after_restart 1
diff --git a/regression-test/framework/pom.xml b/regression-test/framework/pom.xml
index 51547d11faa..d5cbdc0670c 100644
--- a/regression-test/framework/pom.xml
+++ b/regression-test/framework/pom.xml
@@ -350,6 +350,10 @@ under the License.
</dependency>-->
<!---->
<!-- aws ram -->
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>sqs</artifactId>
+ </dependency>
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>iam</artifactId>
diff --git a/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_event_aksk.groovy b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_event_aksk.groovy
new file mode 100644
index 00000000000..6b519b621a8
--- /dev/null
+++ b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_event_aksk.groovy
@@ -0,0 +1,217 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import com.amazonaws.auth.AWSStaticCredentialsProvider
+import com.amazonaws.auth.BasicAWSCredentials
+import com.amazonaws.client.builder.AwsClientBuilder
+import com.amazonaws.services.s3.AmazonS3ClientBuilder
+import groovy.json.JsonOutput
+import org.awaitility.Awaitility
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.services.sqs.SqsClient
+import software.amazon.awssdk.services.sqs.model.GetQueueAttributesRequest
+import software.amazon.awssdk.services.sqs.model.QueueAttributeName
+import software.amazon.awssdk.services.sqs.model.SendMessageRequest
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_streaming_insert_job_s3_event_aksk") {
+ String queueUrl = getConf("s3EventQueueUrl", "")
+ if (!queueUrl) {
+ logger.info("Skip ${name}: s3EventQueueUrl is not configured")
+ return
+ }
+ String eventPrefix = getConf("s3EventPrefix", "")
+ def config = context.config
+ ["awsEndpoint", "awsRegion", "awsBucket", "awsAccessKey", "awsSecretKey"].each { key ->
+ if (!config[key]?.trim()) {
+ throw new IllegalArgumentException("${key} is required for ${name}")
+ }
+ }
+ if (!eventPrefix) {
+ throw new IllegalArgumentException("s3EventPrefix is required for ${name}")
+ }
+ def client = AmazonS3ClientBuilder.standard()
+ .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(config.awsEndpoint, config.awsRegion))
+ .withCredentials(new AWSStaticCredentialsProvider(
+ new BasicAWSCredentials(config.awsAccessKey, config.awsSecretKey))).build()
+ String bucket = config.awsBucket
+ String prefix = "${eventPrefix.replaceAll('/+$', '')}/${name}/${UUID.randomUUID()}"
+ String jobName = name
+ def sqsClient = SqsClient.builder().region(Region.of(config.awsRegion))
+ .credentialsProvider(StaticCredentialsProvider.create(
+ AwsBasicCredentials.create(config.awsAccessKey, config.awsSecretKey))).build()
+ def queueCounts = {
+ sqsClient.getQueueAttributes(GetQueueAttributesRequest.builder().queueUrl(queueUrl)
+ .attributeNames(QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES,
+ QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES_NOT_VISIBLE).build()).attributes()
+ }
+ def awaitQueueEmpty = {
+ // Wait for acknowledgement and eventual consistency before measuring a new backlog.
+ Awaitility.await().pollInSameThread().atMost(180, SECONDS).pollInterval(1, SECONDS).until {
+ def counts = queueCounts()
+ counts[QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES].toLong() == 0 &&
+ counts[QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES_NOT_VISIBLE].toLong() == 0
+ }
+ }
+ def upload = { String file, String content ->
+ String key = "${prefix}/${file}"
+ client.putObject(bucket, key, content)
+ }
+ def awaitRows = { int rows ->
+ Awaitility.await().pollInSameThread().atMost(300, SECONDS).pollInterval(1, SECONDS).until {
+ def job = sql("SELECT Status, ErrorMsg FROM jobs(\"type\"=\"insert\") WHERE Name='${jobName}'")
+ if (job && job[0][0] == "PAUSED") {
+ throw new IllegalStateException("S3 EVENT job paused: ${job[0][1]}")
+ }
+ sql("SELECT COUNT(*) FROM test_s3_event_aksk")[0][0] as long == rows
+ }
+ }
+ def awaitOffsets = { List<String> files ->
+ Awaitility.await().pollInSameThread().atMost(120, SECONDS).pollInterval(1, SECONDS).until {
+ def rows = sql("SELECT CurrentOffset, EndOffset, LastSourceEventTimestamp FROM jobs(\"type\"=\"insert\") WHERE Name='${jobName}'")
+ if (!rows || !rows[0][0] || !rows[0][1]) {
+ return false
+ }
+ def current = parseJson(rows[0][0])
+ def end = parseJson(rows[0][1])
+ // Queue counts are approximate; check their shape and range, not an exact backlog.
+ files.any { current == [fileName: "${prefix}/${it}".toString()] } &&
+ end.keySet() == ["fileName", "lagMessages"].toSet() &&
+ end.fileName == current.fileName && end.lagMessages instanceof Number &&
+ end.lagMessages >= 0 &&
+ rows[0][2]?.toString()?.isLong() &&
+ rows[0][2].toLong() > 0 && rows[0][2].toLong() < 10_000_000_000L
+ }
+ }
+ try {
+ sql "DROP JOB IF EXISTS WHERE jobname='${jobName}'"
+ sql "DROP TABLE IF EXISTS test_s3_event_aksk"
+ sql """CREATE TABLE test_s3_event_aksk (id INT, value STRING)
+ UNIQUE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES("replication_num"="1")"""
+
+ // Upload before CREATE JOB so TVF analysis has a real file to inspect.
+ upload("z.csv", "1,first\n")
+ upload("ignored.txt", "999,excluded\n")
+ sql """CREATE JOB ${jobName}
+ PROPERTIES(
+ "s3.ingestion_mode" = "NOTIFICATION",
+ "s3.event.source" = "SQS",
+ "s3.sqs.queue_url" = ${JsonOutput.toJson(queueUrl)},
+ "s3.max_batch_files" = "1",
+ "max_interval" = "1"
+ )
+ ON STREAMING DO INSERT INTO test_s3_event_aksk
+ SELECT * FROM S3(
+ "uri" = ${JsonOutput.toJson("s3://${bucket}/${prefix}/{z,a,b,c,comma*}.csv")},
+ "format" = "csv",
+ "provider" = "S3",
+ "column_separator" = ",",
+ "s3.endpoint" = ${JsonOutput.toJson(config.awsEndpoint)},
+ "s3.region" = ${JsonOutput.toJson(config.awsRegion)},
+ "s3.access_key" = ${JsonOutput.toJson(config.awsAccessKey)},
+ "s3.secret_key" = ${JsonOutput.toJson(config.awsSecretKey)}
+ )"""
+ awaitRows(1)
+ awaitOffsets(["z.csv"])
+ order_qt_first "SELECT * FROM test_s3_event_aksk"
+
+ // A smaller key uploaded after the first commit must still be discovered in NOTIFICATION mode.
+ upload("a.csv", "2,second\n")
+ awaitRows(2)
+ awaitOffsets(["a.csv"])
+ order_qt_second "SELECT * FROM test_s3_event_aksk"
+
+ awaitQueueEmpty()
+ sql "PAUSE JOB WHERE jobname='${jobName}'"
+ def pausedProgress = sql("SELECT CurrentOffset, LastSourceEventTimestamp FROM jobs(\"type\"=\"insert\") WHERE Name='${jobName}'")[0]
+ upload("b.csv", "3,third\n")
+ upload("c.csv", "4,fourth\n")
+ // Uploaded files must not advance the paused job; RESUME below verifies they are eventually consumed.
+ Awaitility.await().pollInSameThread().atMost(120, SECONDS).pollInterval(1, SECONDS).until {
+ def job = sql("SELECT Status, CurrentOffset, LastSourceEventTimestamp FROM jobs(\"type\"=\"insert\") WHERE Name='${jobName}'")
+ job[0][0] == "PAUSED" && job[0][1] == pausedProgress[0] && job[0][2] == pausedProgress[1] &&
+ (sql("SELECT COUNT(*) FROM test_s3_event_aksk")[0][0] as long) == 2
+ }
+ sql "RESUME JOB WHERE jobname='${jobName}'"
+ awaitRows(4)
+ // SQS does not guarantee the receive order of these two uploads.
+ awaitOffsets(["b.csv", "c.csv"])
+ order_qt_final "SELECT id, value, COUNT(*) FROM test_s3_event_aksk GROUP BY id, value"
+ awaitQueueEmpty()
+
+ // These decoys do not match the job glob. They must not be read when the comma key is rewritten.
+ upload("comma", "9001,wrong_first\n")
+ upload("other.csv", "9002,wrong_second\n")
+ upload("comma,other.csv", "5,comma\n")
+ awaitRows(5)
+ awaitOffsets(["comma,other.csv"])
+ awaitQueueEmpty()
+ order_qt_event_comma "SELECT * FROM test_s3_event_aksk"
+
+ def committedProgress = sql("SELECT Id, LastSourceEventTimestamp FROM jobs(\"type\"=\"insert\") WHERE Name='${jobName}'")[0]
+ String jobId = committedProgress[0].toString()
+ long committedEventTime = committedProgress[1].toLong()
+ Awaitility.await().pollInSameThread().atMost(120, SECONDS).pollInterval(1, SECONDS).until {
+ boolean metricsReady = false
+ httpTest {
+ endpoint getMasterIp() + ":" + getMasterPort("http")
+ uri "/metrics?type=json"
+ op "get"
+ check { code, body ->
+ assertEquals(200, code)
+ def metrics = parseJson(body).findAll {
+ it.tags?.job_id == jobId && it.tags?.job_name == jobName
+ }
+ def lag = metrics.find { it.tags.metric == "doris_fe_streaming_job_per_job_lag_messages" }
+ def timestamp = metrics.find {
+ it.tags.metric == "doris_fe_streaming_job_per_job_last_source_event_timestamp_seconds"
+ }
+ metricsReady = lag?.value != null && new BigDecimal(lag.value.toString()).signum() >= 0 &&
+ timestamp?.value != null &&
+ new BigDecimal(timestamp.value.toString()).compareTo(BigDecimal.valueOf(committedEventTime)) == 0
+ }
+ }
+ metricsReady
+ }
+
+ String committedOffset = sql("SELECT CurrentOffset FROM jobs(\"type\"=\"insert\") WHERE Name='${jobName}'")[0][0]
+ // A message without S3 records is acknowledged without advancing the job.
+ sql "PAUSE JOB WHERE jobname='${jobName}'"
+ sqsClient.sendMessage(SendMessageRequest.builder().queueUrl(queueUrl).messageBody("{}").build())
+ Awaitility.await().pollInSameThread().atMost(120, SECONDS).pollInterval(1, SECONDS).until {
+ queueCounts()[QueueAttributeName.APPROXIMATE_NUMBER_OF_MESSAGES].toLong() > 0
+ }
+ sql "RESUME JOB WHERE jobname='${jobName}'"
+ awaitQueueEmpty()
+ Awaitility.await().pollInSameThread().atMost(120, SECONDS).pollInterval(1, SECONDS).until {
+ def job = sql("SELECT Status, CurrentOffset FROM jobs(\"type\"=\"insert\") WHERE Name='${jobName}'")
+ job[0][0] == "RUNNING" && job[0][1] == committedOffset &&
+ (sql("SELECT COUNT(*) FROM test_s3_event_aksk")[0][0] as long) == 5
+ }
+ } finally {
+ try {
+ sql "DROP JOB IF EXISTS WHERE jobname='${jobName}'"
+ } finally {
+ sqsClient.close()
+ client.shutdown()
+ }
+ }
+}
diff --git a/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_event_iam_role.groovy b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_event_iam_role.groovy
new file mode 100644
index 00000000000..8dfb30d0d2b
--- /dev/null
+++ b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_event_iam_role.groovy
@@ -0,0 +1,128 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import com.amazonaws.auth.AWSStaticCredentialsProvider
+import com.amazonaws.auth.BasicAWSCredentials
+import com.amazonaws.client.builder.AwsClientBuilder
+import com.amazonaws.services.s3.AmazonS3ClientBuilder
+import groovy.json.JsonOutput
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_streaming_insert_job_s3_event_iam_role") {
+ String queueUrl = getConf("s3EventIamRoleQueueUrl", "")
+ if (!queueUrl) {
+ logger.info("Skip ${name}: s3EventIamRoleQueueUrl is not configured")
+ return
+ }
+ String eventPrefix = getConf("s3EventIamRolePrefix", "")
+ def config = context.config
+ ["awsEndpoint", "awsRegion", "awsBucket", "awsAccessKey", "awsSecretKey"].each { key ->
+ if (!config[key]?.trim()) {
+ throw new IllegalArgumentException("${key} is required for ${name}")
+ }
+ }
+ if (!eventPrefix) {
+ throw new IllegalArgumentException("s3EventIamRolePrefix is required for ${name}")
+ }
+ if (!config.awsRoleArn?.trim()) {
+ throw new IllegalArgumentException("awsRoleArn is required for ${name}")
+ }
+ // Upload with test AKSK; the Doris TVF uses only IAM Role properties.
+ def client = AmazonS3ClientBuilder.standard()
+ .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(config.awsEndpoint, config.awsRegion))
+ .withCredentials(new AWSStaticCredentialsProvider(
+ new BasicAWSCredentials(config.awsAccessKey, config.awsSecretKey))).build()
+ String bucket = config.awsBucket
+ String prefix = "${eventPrefix.replaceAll('/+$', '')}/${name}/${UUID.randomUUID()}"
+ String jobName = name
+ def upload = { String file, String content ->
+ String key = "${prefix}/${file}"
+ client.putObject(bucket, key, content)
+ }
+ def awaitRows = { int rows ->
+ Awaitility.await().pollInSameThread().atMost(300, SECONDS).pollInterval(1, SECONDS).until {
+ def job = sql("SELECT Status, ErrorMsg FROM jobs(\"type\"=\"insert\") WHERE Name='${jobName}'")
+ if (job && job[0][0] == "PAUSED") {
+ throw new IllegalStateException("S3 EVENT job paused: ${job[0][1]}")
+ }
+ sql("SELECT COUNT(*) FROM test_s3_event_iam_role")[0][0] as long == rows
+ }
+ }
+ def awaitOffsets = { String file ->
+ Awaitility.await().pollInSameThread().atMost(120, SECONDS).pollInterval(1, SECONDS).until {
+ def rows = sql("SELECT CurrentOffset, EndOffset, LastSourceEventTimestamp FROM jobs(\"type\"=\"insert\") WHERE Name='${jobName}'")
+ if (!rows || !rows[0][0] || !rows[0][1]) {
+ return false
+ }
+ def current = parseJson(rows[0][0])
+ def end = parseJson(rows[0][1])
+ // A numeric backlog also verifies GetQueueAttributes access through the assumed role.
+ current == [fileName: "${prefix}/${file}".toString()] &&
+ end.keySet() == ["fileName", "lagMessages"].toSet() &&
+ end.fileName == current.fileName && end.lagMessages instanceof Number &&
+ end.lagMessages >= 0 &&
+ rows[0][2]?.toString()?.isLong() &&
+ rows[0][2].toLong() > 0 && rows[0][2].toLong() < 10_000_000_000L
+ }
+ }
+ try {
+ sql "DROP JOB IF EXISTS WHERE jobname='${jobName}'"
+ sql "DROP TABLE IF EXISTS test_s3_event_iam_role"
+ sql """CREATE TABLE test_s3_event_iam_role (id INT, value STRING)
+ UNIQUE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES("replication_num"="1")"""
+ upload("z.csv", "1,role_first\n")
+ sql """CREATE JOB ${jobName}
+ PROPERTIES(
+ "s3.ingestion_mode" = "NOTIFICATION",
+ "s3.event.source" = "SQS",
+ "s3.sqs.queue_url" = ${JsonOutput.toJson(queueUrl)},
+ "s3.max_batch_files" = "1",
+ "max_interval" = "1"
+ )
+ ON STREAMING DO INSERT INTO test_s3_event_iam_role
+ SELECT * FROM S3(
+ "uri" = ${JsonOutput.toJson("s3://${bucket}/${prefix}/*.csv")},
+ "format" = "csv",
+ "provider" = "S3",
+ "column_separator" = ",",
+ "s3.endpoint" = ${JsonOutput.toJson(config.awsEndpoint)},
+ "s3.region" = ${JsonOutput.toJson(config.awsRegion)},
+ "s3.role_arn" = ${JsonOutput.toJson(config.awsRoleArn)}
+ ${config.awsExternalId ? ', "s3.external_id"=' + JsonOutput.toJson(config.awsExternalId) : ''}
+ )"""
+ awaitRows(1)
+ awaitOffsets("z.csv")
+ order_qt_first "SELECT * FROM test_s3_event_iam_role"
+
+ // The next batch also exercises message acknowledgement with the assumed role.
+ upload("a.csv", "2,role_second\n")
+ awaitRows(2)
+ awaitOffsets("a.csv")
+ order_qt_final "SELECT id, value, COUNT(*) FROM test_s3_event_iam_role GROUP BY id, value"
+ // Let the next fetchMeta acknowledge the final committed SQS message before dropping the job.
+ sleep(5_000)
+ } finally {
+ try {
+ sql "DROP JOB IF EXISTS WHERE jobname='${jobName}'"
+ } finally {
+ client.shutdown()
+ }
+ }
+}
diff --git a/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_event_restart_fe.groovy b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_event_restart_fe.groovy
new file mode 100644
index 00000000000..b3b412c6258
--- /dev/null
+++ b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_event_restart_fe.groovy
@@ -0,0 +1,152 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import com.amazonaws.auth.AWSStaticCredentialsProvider
+import com.amazonaws.auth.BasicAWSCredentials
+import com.amazonaws.client.builder.AwsClientBuilder
+import com.amazonaws.services.s3.AmazonS3ClientBuilder
+import groovy.json.JsonOutput
+import org.apache.doris.regression.suite.ClusterOptions
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_streaming_insert_job_s3_event_restart_fe", "docker") {
+ String queueUrl = getConf("s3EventQueueUrl", "")
+ if (!queueUrl) {
+ logger.info("Skip ${name}: s3EventQueueUrl is not configured")
+ return
+ }
+ String eventPrefix = getConf("s3EventPrefix", "")
+ def config = context.config
+ ["awsEndpoint", "awsRegion", "awsBucket", "awsAccessKey", "awsSecretKey"].each { key ->
+ if (!config[key]?.trim()) {
+ throw new IllegalArgumentException("${key} is required for ${name}")
+ }
+ }
+ if (!eventPrefix) {
+ throw new IllegalArgumentException("s3EventPrefix is required for ${name}")
+ }
+ def options = new ClusterOptions()
+ options.setFeNum(1)
+ options.setBeNum(1)
+ options.cloudMode = null
+ // Run separately from the ordinary AKSK case because both use the AKSK queue.
+ docker(options) {
+ def client = AmazonS3ClientBuilder.standard()
+ .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(config.awsEndpoint, config.awsRegion))
+ .withCredentials(new AWSStaticCredentialsProvider(
+ new BasicAWSCredentials(config.awsAccessKey, config.awsSecretKey))).build()
+ String bucket = config.awsBucket
+ String prefix = "${eventPrefix.replaceAll('/+$', '')}/${name}/${UUID.randomUUID()}"
+ String jobName = name
+ def upload = { String file, String content ->
+ String key = "${prefix}/${file}"
+ client.putObject(bucket, key, content)
+ }
+ def awaitRows = { int rows ->
+ Awaitility.await().pollInSameThread().atMost(300, SECONDS).pollInterval(1, SECONDS).until {
+ def job = sql("SELECT Status, ErrorMsg FROM jobs(\"type\"=\"insert\") WHERE Name='${jobName}'")
+ if (job && job[0][0] == "PAUSED") {
+ throw new IllegalStateException("S3 EVENT job paused: ${job[0][1]}")
+ }
+ sql("SELECT COUNT(*) FROM test_s3_event_restart_fe")[0][0] as long == rows
+ }
+ }
+ def awaitOffsets = { String file, boolean expectTimestamp = true ->
+ Awaitility.await().pollInSameThread().atMost(120, SECONDS).pollInterval(1, SECONDS).until {
+ def rows = sql("SELECT CurrentOffset, EndOffset, LastSourceEventTimestamp FROM jobs(\"type\"=\"insert\") WHERE Name='${jobName}'")
+ if (!rows || !rows[0][0] || !rows[0][1]) {
+ return false
+ }
+ def current = parseJson(rows[0][0])
+ def end = parseJson(rows[0][1])
+ // After restart, the file is restored and the queue statistic must be fetched again.
+ current == [fileName: "${prefix}/${file}".toString()] &&
+ end.keySet() == ["fileName", "lagMessages"].toSet() &&
+ end.fileName == current.fileName && end.lagMessages instanceof Number &&
+ end.lagMessages >= 0 &&
+ (expectTimestamp ? (rows[0][2]?.toString()?.isLong() &&
+ rows[0][2].toLong() > 0 && rows[0][2].toLong() < 10_000_000_000L) : !rows[0][2])
+ }
+ }
+ try {
+ sql "DROP JOB IF EXISTS WHERE jobname='${jobName}'"
+ sql "DROP TABLE IF EXISTS test_s3_event_restart_fe"
+ sql """CREATE TABLE test_s3_event_restart_fe (id INT, value STRING)
+ UNIQUE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES("replication_num"="1")"""
+ upload("z.csv", "1,before_restart\n")
+ sql """CREATE JOB ${jobName}
+ PROPERTIES(
+ "s3.ingestion_mode" = "NOTIFICATION",
+ "s3.event.source" = "SQS",
+ "s3.sqs.queue_url" = ${JsonOutput.toJson(queueUrl)},
+ "s3.max_batch_files" = "1",
+ "max_interval" = "1"
+ )
+ ON STREAMING DO INSERT INTO test_s3_event_restart_fe
+ SELECT * FROM S3(
+ "uri" = ${JsonOutput.toJson("s3://${bucket}/${prefix}/*.csv")},
+ "format" = "csv",
+ "provider" = "S3",
+ "column_separator" = ",",
+ "s3.endpoint" = ${JsonOutput.toJson(config.awsEndpoint)},
+ "s3.region" = ${JsonOutput.toJson(config.awsRegion)},
+ "s3.access_key" = ${JsonOutput.toJson(config.awsAccessKey)},
+ "s3.secret_key" = ${JsonOutput.toJson(config.awsSecretKey)}
+ )"""
+ awaitRows(1)
+ awaitOffsets("z.csv")
+ // A second committed task ensures the first batch's acknowledgement has completed.
+ upload("y.csv", "2,before_restart\n")
+ awaitRows(2)
+ awaitOffsets("y.csv")
+ def offsetSql = "SELECT CurrentOffset FROM jobs(\"type\"=\"insert\") WHERE Name='${jobName}'"
+ String committedOffset = sql(offsetSql)[0][0]
+ order_qt_before "SELECT * FROM test_s3_event_restart_fe"
+
+ cluster.restartFrontends()
+ Awaitility.await().pollInSameThread().atMost(120, SECONDS).pollInterval(1, SECONDS).ignoreExceptions().until {
+ context.reconnectFe()
+ sql("SELECT 1")[0][0] == 1
+ }
+ Awaitility.await().pollInSameThread().atMost(120, SECONDS).pollInterval(1, SECONDS).until {
+ def rows = sql(offsetSql)
+ rows && rows[0][0] == committedOffset
+ }
+ awaitOffsets("y.csv", false)
+ order_qt_restored "SELECT * FROM test_s3_event_restart_fe"
+
+ upload("a.csv", "3,after_restart\n")
+ awaitRows(3)
+ awaitOffsets("a.csv")
+ upload("b.csv", "4,after_restart\n")
+ awaitRows(4)
+ awaitOffsets("b.csv")
+ order_qt_final "SELECT id, value, COUNT(*) FROM test_s3_event_restart_fe GROUP BY id, value"
+ // Let the next fetchMeta acknowledge the final committed SQS message before dropping the job.
+ sleep(5_000)
+ } finally {
+ try {
+ sql "DROP JOB IF EXISTS WHERE jobname='${jobName}'"
+ } finally {
+ client.shutdown()
+ }
+ }
+ }
+}
diff --git a/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_once.groovy b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_once.groovy
index 799312f3de9..9699259cee9 100644
--- a/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_once.groovy
+++ b/regression-test/suites/job_p0/streaming_job/test_streaming_insert_job_s3_once.groovy
@@ -38,7 +38,7 @@ suite("test_streaming_insert_job_s3_once") {
sql """
CREATE JOB test_streaming_insert_job_s3_once
PROPERTIES (
- "s3.ingestion_mode" = "ONCE",
+ "s3.ingestion_mode" = "ONE_TIME",
"offset" = '{"fileName":"regression/load/data/example_0.csv"}'
)
ON STREAMING DO INSERT INTO test_streaming_insert_job_s3_once_tbl
@@ -53,13 +53,13 @@ suite("test_streaming_insert_job_s3_once") {
"s3.secret_key" = "${getS3SK()}"
)
"""
- exception "offset is not supported when s3.ingestion_mode is ONCE"
+ exception "offset is not supported when s3.ingestion_mode is ONE_TIME"
}
sql """
CREATE JOB test_streaming_insert_job_s3_once
PROPERTIES (
- "s3.ingestion_mode" = "ONCE",
+ "s3.ingestion_mode" = "ONE_TIME",
"s3.max_batch_files" = "1"
)
ON STREAMING DO INSERT INTO test_streaming_insert_job_s3_once_tbl
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org