github-actions[bot] commented on code in PR #68876:
URL: https://github.com/apache/doris/pull/68876#discussion_r4237112652
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProviderFactory.java:
##########
@@ -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()) {
Review Comment:
[P1] Preserve one-time semantics for persisted ONCE jobs. An active job
stored before this rename is reconstructed by `gsonPostProcess()` without mode
validation. This factory no longer recognizes `ONCE`, so it chooses the no-arg
`S3SourceOffsetProvider` with `oneTimeMode=false`; `hasReachedEnd()` ignores
even a persisted `lastBatch=true`, and future objects can be ingested instead
of ending the job. An unrelated ALTER of stored `ONCE` or `LEXICAL` jobs also
fails merged-property validation. Migrate stored modes or accept their legacy
spellings on replay and ALTER.
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProviderFactory.java:
##########
@@ -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()) {
Review Comment:
[P1] Gate notification jobs while an older FE can become master. A new
`NOTIFICATION` job persists its mode as a string, but the previous factory has
no notification branch and falls through to the ordinary S3 lister during
replay. The old provider reads the event-offset JSON with no `endFile` cursor,
then lists the source glob from the beginning instead of consuming SQS. During
a rolling upgrade, leadership moving to an older FE can import preexisting
objects outside the notification batch. Prevent older electable FEs from
scheduling this mode.
##########
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
Review Comment:
[P2] Use an append-preserving target for the restart oracle. `UNIQUE
KEY(id)` folds a second committed load of the same S3 object into the existing
row, so the before/restored/final counts and `GROUP BY` output still pass if FE
recovery reimports the latest committed batch. Use a `DUPLICATE KEY` table or
an append-only audit target and assert per-file multiplicity; the other
notification suites use the same masking table model.
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3EventSourceOffsetProvider.java:
##########
@@ -0,0 +1,824 @@
+// 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.S3URI;
+import org.apache.doris.datasource.storage.StorageAdapter;
+import org.apache.doris.filesystem.FileSystem;
+import org.apache.doris.filesystem.properties.S3CompatibleFileSystemProperties;
+import org.apache.doris.filesystem.spi.ObjFileSystem;
+import org.apache.doris.filesystem.spi.S3CompatibleFileSystem;
+import org.apache.doris.fs.FileSystemFactory;
+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.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.regex.Pattern;
+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 Pattern 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;
+ // Recovered keys map to their latest receipts and are removed after
successful SQS deletion.
+ private transient Map<String, String> recoveredCommittedFiles;
+ // 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 FileSystem 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);
+ StorageAdapter storageProperties = StorageAdapter.of(copiedProps);
+ if
(!"S3".equals(storageProperties.getSpiProperties().providerName())) {
+ throw new JobException("S3 NOTIFICATION mode currently
supports AWS S3 only");
+ }
+ S3CompatibleFileSystemProperties s3Properties =
+ (S3CompatibleFileSystemProperties)
storageProperties.getSpiProperties();
+ String uri = storageProperties.validateAndGetUri(copiedProps);
+ S3URI s3Uri =
S3URI.create(storageProperties.validateAndNormalizeUri(uri));
Review Comment:
[P1] Preserve glob question marks when deriving the event matcher. For
`s3://bucket/logs/file?.csv`, `S3URI.create()` treats `?` as the query
delimiter and `getKey()` yields `logs/file`, while ordinary S3 listing treats
`?` as a one-character wildcard. An event for `logs/file1.csv` is then
considered unmatched and its SQS receipt is deleted, permanently losing that
file. Parse the bucket/key with the same raw-key rules as the glob lister and
cover this pattern.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
