This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 1e450e11f7 [Fix][Connector-V2][AmazonSqs] Preserve messages on
deserialization failure (#12038)
1e450e11f7 is described below
commit 1e450e11f7c8e3ef78d186b0a407148d5faed939
Author: Goutam Adwant <[email protected]>
AuthorDate: Fri Sep 4 02:53:49 2026 +0000
[Fix][Connector-V2][AmazonSqs] Preserve messages on deserialization failure
(#12038)
---
docs/en/connectors/source/AmazonSqs.md | 3 +
docs/zh/connectors/source/AmazonSqs.md | 3 +
.../amazonsqs/config/AmazonSqsSourceConfig.java | 3 +
.../amazonsqs/config/AmazonSqsSourceOptions.java | 6 +
.../deserialize/AmazonSqsDeserializer.java | 23 +-
.../AmazonSqsConnectorErrorCode.java} | 29 +-
.../amazonsqs/source/AmazonSqsSourceFactory.java | 14 +-
.../amazonsqs/source/AmazonSqsSourceReader.java | 8 +-
.../amazonsqs/AmazonSqsSourceFactoryTest.java | 6 +
.../source/AmazonSqsSourceReaderTest.java | 457 +++++++++++++++++++++
10 files changed, 529 insertions(+), 23 deletions(-)
diff --git a/docs/en/connectors/source/AmazonSqs.md
b/docs/en/connectors/source/AmazonSqs.md
index f571cb4551..443d8434ef 100644
--- a/docs/en/connectors/source/AmazonSqs.md
+++ b/docs/en/connectors/source/AmazonSqs.md
@@ -40,6 +40,7 @@ Each receive request asks SQS for up to 10 messages. If more
messages are waitin
| secret_access_key | String | No | - | AWS secret
access key. Set it together with `access_key_id` to use static credentials.
|
| format | String | No | json | Message body
format. Supported values are `json`, `text`, `canal_json`, and `debezium_json`.
|
| field_delimiter | String | No | , | Field
delimiter used when `format = text`.
|
+| ignore_parse_errors | Boolean | No | false | Whether to
skip messages that cannot be deserialized instead of failing the poll.
|
| delete_message | Boolean | No | false | Whether to
delete a message from the queue after it is read and deserialized successfully.
|
| message_group_id | String | No | - | Message
group ID option kept for compatibility. It is not required for normal SQS
reads. |
| debezium_record_include_schema | Boolean | No | true | Whether
Debezium JSON messages include a schema. This option is used only when `format
= debezium_json`. |
@@ -53,6 +54,8 @@ Each receive request asks SQS for up to 10 messages. If more
messages are waitin
- `text` splits each message body by `field_delimiter` and maps the values to
fields in `schema` order.
- `canal_json` reads Canal JSON messages. For details, see [Canal
JSON](../formats/canal-json.md).
- `debezium_json` reads Debezium JSON messages. For details, see [Debezium
JSON](../formats/debezium-json.md).
+- `ignore_parse_errors = false` fails the poll and retains an unreadable
message. When set to `true`, the source skips the message and continues
processing the batch.
+- When both `ignore_parse_errors` and `delete_message` are `true`, skipped
messages are deleted from SQS. Keep `delete_message = false` if skipped
messages should remain available for redelivery.
- `delete_message = true` removes consumed messages from SQS. Keep the default
`false` when you only want to inspect or copy messages without deleting them.
- `access_key_id` and `secret_access_key` are optional, but they must be
configured together when static AWS credentials are used.
- The source performs one receive request, with up to 10 messages, and then
finishes the bounded job.
diff --git a/docs/zh/connectors/source/AmazonSqs.md
b/docs/zh/connectors/source/AmazonSqs.md
index 159d5f4ecc..81f2f7db59 100644
--- a/docs/zh/connectors/source/AmazonSqs.md
+++ b/docs/zh/connectors/source/AmazonSqs.md
@@ -39,6 +39,7 @@ Amazon SQS 源连接器用于从一个 Amazon SQS 队列 URL 读取消息。连
| secret_access_key | String | 否 | - | AWS secret access
key。和 `access_key_id` 一起配置时使用静态凭证。
|
| format | String | 否 | json | 消息体格式。支持
`json`、`text`、`canal_json`、`debezium_json`。
|
| field_delimiter | String | 否 | , | 当 `format = text`
时使用的字段分隔符。
|
+| ignore_parse_errors | Boolean | 否 | false |
是否跳过无法解析的消息并继续处理，而不是让本次轮询失败。
|
| delete_message | Boolean | 否 | false |
读取并成功解析消息后，是否从队列中删除该消息。
|
| message_group_id | String | 否 | - | 为兼容保留的消息分组 ID
选项，普通 SQS 读取不需要配置。
|
| debezium_record_include_schema | Boolean | 否 | true | Debezium JSON
消息是否包含 schema。仅在 `format = debezium_json` 时使用。
|
@@ -52,6 +53,8 @@ Amazon SQS 源连接器用于从一个 Amazon SQS 队列 URL 读取消息。连
- `text`：按 `field_delimiter` 切分消息体，并按 `schema` 中字段顺序映射。
- `canal_json`：读取 Canal JSON 消息，详见 [Canal JSON](../formats/canal-json.md)。
- `debezium_json`：读取 Debezium JSON 消息，详见 [Debezium
JSON](../formats/debezium-json.md)。
+- `ignore_parse_errors = false` 会让本次轮询失败并保留无法解析的消息。设置为 `true`
时，源连接器会跳过该消息并继续处理本批次中的其他消息。
+- 当 `ignore_parse_errors` 和 `delete_message` 都为 `true` 时，跳过的消息会从 SQS
中删除。如果需要保留这些消息以便重新投递，请保持 `delete_message = false`。
- `delete_message = true` 会删除已经消费的 SQS 消息。如果只是检查或复制消息，建议保留默认值 `false`。
- `access_key_id` 和 `secret_access_key` 是可选项；如果使用静态 AWS 凭证，需要两个一起配置。
- 该源连接器只执行一次 receive 请求，最多读取 10 条消息，然后结束这个有界任务。
diff --git
a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/config/AmazonSqsSourceConfig.java
b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/config/AmazonSqsSourceConfig.java
index 58113a0d91..9e51dc907f 100644
---
a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/config/AmazonSqsSourceConfig.java
+++
b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/config/AmazonSqsSourceConfig.java
@@ -42,6 +42,8 @@ public class AmazonSqsSourceConfig implements Serializable {
private boolean deleteMessage;
+ private boolean ignoreParseErrors;
+
private Config schema;
public AmazonSqsSourceConfig(ReadonlyConfig config) {
@@ -51,6 +53,7 @@ public class AmazonSqsSourceConfig implements Serializable {
this.secretAccessKey =
config.get(AmazonSqsSourceOptions.SECRET_ACCESS_KEY);
this.messageGroupId =
config.get(AmazonSqsSourceOptions.MESSAGE_GROUP_ID);
this.deleteMessage = config.get(AmazonSqsSourceOptions.DELETE_MESSAGE);
+ this.ignoreParseErrors =
config.get(AmazonSqsSourceOptions.IGNORE_PARSE_ERRORS);
this.schema =
ReadonlyConfig.fromMap(config.get(AmazonSqsSourceOptions.SCHEMA)).toConfig();
;
}
diff --git
a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/config/AmazonSqsSourceOptions.java
b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/config/AmazonSqsSourceOptions.java
index d302f17661..d03acb2c6a 100644
---
a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/config/AmazonSqsSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/config/AmazonSqsSourceOptions.java
@@ -28,6 +28,12 @@ public class AmazonSqsSourceOptions extends
AmazonSqsBaseOptions {
.defaultValue(false)
.withDescription("Delete the message after it is consumed
if set true.");
+ public static final Option<Boolean> IGNORE_PARSE_ERRORS =
+ Options.key("ignore_parse_errors")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Optional flag to skip parse errors
instead of failing.");
+
public static final Option<String> MESSAGE_GROUP_ID =
Options.key("message_group_id")
.stringType()
diff --git
a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/deserialize/AmazonSqsDeserializer.java
b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/deserialize/AmazonSqsDeserializer.java
index 7322feda79..21314f20ed 100644
---
a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/deserialize/AmazonSqsDeserializer.java
+++
b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/deserialize/AmazonSqsDeserializer.java
@@ -19,23 +19,40 @@ package
org.apache.seatunnel.connectors.seatunnel.amazonsqs.deserialize;
import org.apache.seatunnel.api.serialization.DeserializationSchema;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import
org.apache.seatunnel.connectors.seatunnel.amazonsqs.exception.AmazonSqsConnectorErrorCode;
+import
org.apache.seatunnel.connectors.seatunnel.amazonsqs.exception.AmazonSqsConnectorException;
import java.io.IOException;
public class AmazonSqsDeserializer implements SeaTunnelRowDeserializer {
private final DeserializationSchema<SeaTunnelRow> deserializationSchema;
+ private final boolean ignoreParseErrors;
- public AmazonSqsDeserializer(DeserializationSchema<SeaTunnelRow>
deserializationSchema) {
+ public AmazonSqsDeserializer(
+ DeserializationSchema<SeaTunnelRow> deserializationSchema, boolean
ignoreParseErrors) {
this.deserializationSchema = deserializationSchema;
+ this.ignoreParseErrors = ignoreParseErrors;
}
@Override
public SeaTunnelRow deserializeRow(String row) {
try {
- return deserializationSchema.deserialize(row.getBytes());
+ SeaTunnelRow seaTunnelRow =
deserializationSchema.deserialize(row.getBytes());
+ if (seaTunnelRow == null && !ignoreParseErrors) {
+ throw new AmazonSqsConnectorException(
+ AmazonSqsConnectorErrorCode.DESERIALIZE_FAILED,
+ "Failed to deserialize Amazon SQS message");
+ }
+ return seaTunnelRow;
} catch (IOException e) {
- return null;
+ if (ignoreParseErrors) {
+ return null;
+ }
+ throw new AmazonSqsConnectorException(
+ AmazonSqsConnectorErrorCode.DESERIALIZE_FAILED,
+ "Failed to deserialize Amazon SQS message",
+ e);
}
}
}
diff --git
a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/deserialize/AmazonSqsDeserializer.java
b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/exception/AmazonSqsConnectorErrorCode.java
similarity index 53%
copy from
seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/deserialize/AmazonSqsDeserializer.java
copy to
seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/exception/AmazonSqsConnectorErrorCode.java
index 7322feda79..fa929814f4 100644
---
a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/deserialize/AmazonSqsDeserializer.java
+++
b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/exception/AmazonSqsConnectorErrorCode.java
@@ -15,27 +15,28 @@
* limitations under the License.
*/
-package org.apache.seatunnel.connectors.seatunnel.amazonsqs.deserialize;
+package org.apache.seatunnel.connectors.seatunnel.amazonsqs.exception;
-import org.apache.seatunnel.api.serialization.DeserializationSchema;
-import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
-import java.io.IOException;
+public enum AmazonSqsConnectorErrorCode implements SeaTunnelErrorCode {
+ DESERIALIZE_FAILED("AmazonSqs-01", "Deserialize Amazon SQS message
failed");
-public class AmazonSqsDeserializer implements SeaTunnelRowDeserializer {
+ private final String code;
+ private final String description;
- private final DeserializationSchema<SeaTunnelRow> deserializationSchema;
+ AmazonSqsConnectorErrorCode(String code, String description) {
+ this.code = code;
+ this.description = description;
+ }
- public AmazonSqsDeserializer(DeserializationSchema<SeaTunnelRow>
deserializationSchema) {
- this.deserializationSchema = deserializationSchema;
+ @Override
+ public String getCode() {
+ return code;
}
@Override
- public SeaTunnelRow deserializeRow(String row) {
- try {
- return deserializationSchema.deserialize(row.getBytes());
- } catch (IOException e) {
- return null;
- }
+ public String getDescription() {
+ return description;
}
}
diff --git
a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/source/AmazonSqsSourceFactory.java
b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/source/AmazonSqsSourceFactory.java
index aaeda80396..3ce5fcaf6b 100644
---
a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/source/AmazonSqsSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/source/AmazonSqsSourceFactory.java
@@ -51,6 +51,7 @@ import static
org.apache.seatunnel.connectors.seatunnel.amazonsqs.config.AmazonS
import static
org.apache.seatunnel.connectors.seatunnel.amazonsqs.config.AmazonSqsSourceOptions.DELETE_MESSAGE;
import static
org.apache.seatunnel.connectors.seatunnel.amazonsqs.config.AmazonSqsSourceOptions.FIELD_DELIMITER;
import static
org.apache.seatunnel.connectors.seatunnel.amazonsqs.config.AmazonSqsSourceOptions.FORMAT;
+import static
org.apache.seatunnel.connectors.seatunnel.amazonsqs.config.AmazonSqsSourceOptions.IGNORE_PARSE_ERRORS;
import static
org.apache.seatunnel.connectors.seatunnel.amazonsqs.config.AmazonSqsSourceOptions.MESSAGE_GROUP_ID;
import static
org.apache.seatunnel.connectors.seatunnel.amazonsqs.config.AmazonSqsSourceOptions.REGION;
import static
org.apache.seatunnel.connectors.seatunnel.amazonsqs.config.AmazonSqsSourceOptions.SECRET_ACCESS_KEY;
@@ -72,6 +73,7 @@ public class AmazonSqsSourceFactory implements
TableSourceFactory {
SECRET_ACCESS_KEY,
MESSAGE_GROUP_ID,
DELETE_MESSAGE,
+ IGNORE_PARSE_ERRORS,
FORMAT,
FIELD_DELIMITER,
DEBEZIUM_RECORD_INCLUDE_SCHEMA)
@@ -100,10 +102,13 @@ public class AmazonSqsSourceFactory implements
TableSourceFactory {
private DeserializationSchema<SeaTunnelRow> setDeserialization(
Config config, CatalogTable catalogTable) {
DeserializationSchema<SeaTunnelRow> deserializationSchema;
- MessageFormat format = ReadonlyConfig.fromConfig(config).get(FORMAT);
+ ReadonlyConfig readonlyConfig = ReadonlyConfig.fromConfig(config);
+ MessageFormat format = readonlyConfig.get(FORMAT);
+ boolean ignoreParseErrors = readonlyConfig.get(IGNORE_PARSE_ERRORS);
switch (format) {
case JSON:
- deserializationSchema = new
JsonDeserializationSchema(catalogTable, false, false);
+ deserializationSchema =
+ new JsonDeserializationSchema(catalogTable, false,
ignoreParseErrors);
break;
case TEXT:
String delimiter = DEFAULT_FIELD_DELIMITER;
@@ -119,7 +124,7 @@ public class AmazonSqsSourceFactory implements
TableSourceFactory {
case CANAL_JSON:
deserializationSchema =
CanalJsonDeserializationSchema.builder(catalogTable)
- .setIgnoreParseErrors(true)
+ .setIgnoreParseErrors(ignoreParseErrors)
.build();
break;
case DEBEZIUM_JSON:
@@ -128,7 +133,8 @@ public class AmazonSqsSourceFactory implements
TableSourceFactory {
includeSchema =
config.getBoolean(DEBEZIUM_RECORD_INCLUDE_SCHEMA.key());
}
deserializationSchema =
- new DebeziumJsonDeserializationSchema(catalogTable,
true, includeSchema);
+ new DebeziumJsonDeserializationSchema(
+ catalogTable, ignoreParseErrors,
includeSchema);
break;
default:
throw new SeaTunnelJsonFormatException(
diff --git
a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/source/AmazonSqsSourceReader.java
b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/source/AmazonSqsSourceReader.java
index 309b328ee1..5b3d361c31 100644
---
a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/source/AmazonSqsSourceReader.java
+++
b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/source/AmazonSqsSourceReader.java
@@ -57,7 +57,9 @@ public class AmazonSqsSourceReader extends
AbstractSingleSplitReader<SeaTunnelRo
SeaTunnelRowType seaTunnelRowType) {
this.context = context;
this.amazonSqsSourceConfig = amazonSqsSourceConfig;
- this.seaTunnelRowDeserializer = new
AmazonSqsDeserializer(deserializationSchema);
+ this.seaTunnelRowDeserializer =
+ new AmazonSqsDeserializer(
+ deserializationSchema,
amazonSqsSourceConfig.isIgnoreParseErrors());
}
@Override
@@ -107,7 +109,9 @@ public class AmazonSqsSourceReader extends
AbstractSingleSplitReader<SeaTunnelRo
for (Message message : messages) {
String messageBody = message.body();
SeaTunnelRow seaTunnelRow =
this.seaTunnelRowDeserializer.deserializeRow(messageBody);
- output.collect(seaTunnelRow);
+ if (seaTunnelRow != null) {
+ output.collect(seaTunnelRow);
+ }
// Delete the processed message
if (amazonSqsSourceConfig.isDeleteMessage()) {
diff --git
a/seatunnel-connectors-v2/connector-amazonsqs/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/AmazonSqsSourceFactoryTest.java
b/seatunnel-connectors-v2/connector-amazonsqs/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/AmazonSqsSourceFactoryTest.java
index e7fe20da9c..e25c3a7a57 100644
---
a/seatunnel-connectors-v2/connector-amazonsqs/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/AmazonSqsSourceFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-amazonsqs/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/AmazonSqsSourceFactoryTest.java
@@ -18,6 +18,7 @@
package org.apache.seatunnel.connectors.seatunnel.amazonsqs;
import org.apache.seatunnel.api.configuration.util.OptionRule;
+import
org.apache.seatunnel.connectors.seatunnel.amazonsqs.config.AmazonSqsSourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.amazonsqs.sink.AmazonSqsSinkFactory;
import
org.apache.seatunnel.connectors.seatunnel.amazonsqs.source.AmazonSqsSourceFactory;
@@ -32,6 +33,11 @@ public class AmazonSqsSourceFactoryTest {
AmazonSqsSourceFactory amazonSqsSourceFactory = new
AmazonSqsSourceFactory();
OptionRule sourceOptionRule = amazonSqsSourceFactory.optionRule();
Assertions.assertNotNull(sourceOptionRule);
+ Assertions.assertTrue(
+ sourceOptionRule
+ .getOptionalOptions()
+ .contains(AmazonSqsSourceOptions.IGNORE_PARSE_ERRORS));
+
Assertions.assertFalse(AmazonSqsSourceOptions.IGNORE_PARSE_ERRORS.defaultValue());
AmazonSqsSinkFactory amazonSqsSinkFactory = new AmazonSqsSinkFactory();
OptionRule sinkOptionRule = amazonSqsSinkFactory.optionRule();
diff --git
a/seatunnel-connectors-v2/connector-amazonsqs/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/source/AmazonSqsSourceReaderTest.java
b/seatunnel-connectors-v2/connector-amazonsqs/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/source/AmazonSqsSourceReaderTest.java
new file mode 100644
index 0000000000..ed7cb3d1ef
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-amazonsqs/src/test/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/source/AmazonSqsSourceReaderTest.java
@@ -0,0 +1,457 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.amazonsqs.source;
+
+import org.apache.seatunnel.api.common.metrics.MetricsContext;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.event.EventListener;
+import org.apache.seatunnel.api.serialization.DeserializationSchema;
+import org.apache.seatunnel.api.source.Boundedness;
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.source.SourceEvent;
+import org.apache.seatunnel.api.source.SourceReader;
+import org.apache.seatunnel.api.table.connector.TableSource;
+import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import
org.apache.seatunnel.connectors.seatunnel.amazonsqs.config.AmazonSqsSourceConfig;
+import
org.apache.seatunnel.connectors.seatunnel.amazonsqs.exception.AmazonSqsConnectorException;
+import
org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import software.amazon.awssdk.services.sqs.SqsClient;
+import software.amazon.awssdk.services.sqs.model.DeleteMessageRequest;
+import software.amazon.awssdk.services.sqs.model.DeleteMessageResponse;
+import software.amazon.awssdk.services.sqs.model.Message;
+import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+class AmazonSqsSourceReaderTest {
+
+ private static final SeaTunnelRowType ROW_TYPE =
+ new SeaTunnelRowType(
+ new String[] {"value"}, new SeaTunnelDataType[]
{BasicType.STRING_TYPE});
+
+ @Test
+ void shouldNotDeleteMessageWhenDeserializationThrows() {
+ RecordingSqsClient sqsClient = new
RecordingSqsClient(message("invalid", "receipt-1"));
+ RecordingReaderContext context = new RecordingReaderContext();
+ AmazonSqsSourceReader reader =
+ createReader(context, true, new
FailingDeserializationSchema(), sqsClient);
+ RecordingCollector collector = new RecordingCollector();
+
+ AmazonSqsConnectorException exception =
+ Assertions.assertThrows(
+ AmazonSqsConnectorException.class, () ->
reader.pollNext(collector));
+
+ Assertions.assertTrue(
+ exception.getMessage().contains("Failed to deserialize Amazon
SQS message"));
+ Assertions.assertInstanceOf(IOException.class, exception.getCause());
+ Assertions.assertEquals("invalid payload",
exception.getCause().getMessage());
+ Assertions.assertTrue(collector.records.isEmpty());
+ Assertions.assertTrue(sqsClient.deletedRequests.isEmpty());
+ Assertions.assertEquals(0, context.noMoreElementSignals);
+ }
+
+ @Test
+ void shouldNotDeleteMessageWhenDeserializerReturnsNull() {
+ RecordingSqsClient sqsClient = new
RecordingSqsClient(message("invalid", "receipt-1"));
+ RecordingReaderContext context = new RecordingReaderContext();
+ AmazonSqsSourceReader reader =
+ createReader(context, true, new NullDeserializationSchema(),
sqsClient);
+ RecordingCollector collector = new RecordingCollector();
+
+ AmazonSqsConnectorException exception =
+ Assertions.assertThrows(
+ AmazonSqsConnectorException.class, () ->
reader.pollNext(collector));
+
+ Assertions.assertTrue(
+ exception.getMessage().contains("Failed to deserialize Amazon
SQS message"));
+ Assertions.assertTrue(collector.records.isEmpty());
+ Assertions.assertTrue(sqsClient.deletedRequests.isEmpty());
+ Assertions.assertEquals(0, context.noMoreElementSignals);
+ }
+
+ @Test
+ void shouldDeleteMessageAfterSuccessfulCollectionWhenEnabled() throws
Exception {
+ RecordingSqsClient sqsClient = new
RecordingSqsClient(message("order-1", "receipt-1"));
+ RecordingReaderContext context = new RecordingReaderContext();
+ AmazonSqsSourceReader reader =
+ createReader(context, true, new TestDeserializationSchema(),
sqsClient);
+ RecordingCollector collector = new RecordingCollector();
+
+ reader.pollNext(collector);
+
+ Assertions.assertEquals(Collections.singletonList("order-1"),
collector.values());
+ Assertions.assertEquals(
+ Collections.singletonList("receipt-1"),
sqsClient.deletedReceiptHandles());
+ Assertions.assertEquals(1, context.noMoreElementSignals);
+ }
+
+ @Test
+ void shouldKeepMessageAfterSuccessfulCollectionWhenDeletionDisabled()
throws Exception {
+ RecordingSqsClient sqsClient = new
RecordingSqsClient(message("order-1", "receipt-1"));
+ RecordingReaderContext context = new RecordingReaderContext();
+ AmazonSqsSourceReader reader =
+ createReader(context, false, new TestDeserializationSchema(),
sqsClient);
+ RecordingCollector collector = new RecordingCollector();
+
+ reader.pollNext(collector);
+
+ Assertions.assertEquals(Collections.singletonList("order-1"),
collector.values());
+ Assertions.assertTrue(sqsClient.deletedRequests.isEmpty());
+ Assertions.assertEquals(1, context.noMoreElementSignals);
+ }
+
+ @Test
+ void shouldNotDeleteMessageWhenCollectionFails() {
+ RecordingSqsClient sqsClient = new
RecordingSqsClient(message("order-1", "receipt-1"));
+ RecordingReaderContext context = new RecordingReaderContext();
+ AmazonSqsSourceReader reader =
+ createReader(context, true, new TestDeserializationSchema(),
sqsClient);
+ RecordingCollector collector = new RecordingCollector("collector
rejected row");
+
+ IllegalStateException exception =
+ Assertions.assertThrows(
+ IllegalStateException.class, () ->
reader.pollNext(collector));
+
+ Assertions.assertTrue(exception.getMessage().contains("collector
rejected row"));
+ Assertions.assertTrue(sqsClient.deletedRequests.isEmpty());
+ Assertions.assertEquals(0, context.noMoreElementSignals);
+ }
+
+ @Test
+ void shouldStopBatchAtFailedMessageWithoutDeletingItOrLaterMessages() {
+ RecordingSqsClient sqsClient =
+ new RecordingSqsClient(
+ message("order-1", "receipt-1"),
+ message("invalid", "receipt-2"),
+ message("order-3", "receipt-3"));
+ RecordingReaderContext context = new RecordingReaderContext();
+ AmazonSqsSourceReader reader =
+ createReader(context, true, new
SelectiveFailingDeserializationSchema(), sqsClient);
+ RecordingCollector collector = new RecordingCollector();
+
+ AmazonSqsConnectorException exception =
+ Assertions.assertThrows(
+ AmazonSqsConnectorException.class, () ->
reader.pollNext(collector));
+
+ Assertions.assertTrue(
+ exception.getMessage().contains("Failed to deserialize Amazon
SQS message"));
+ Assertions.assertEquals(Collections.singletonList("order-1"),
collector.values());
+ Assertions.assertEquals(
+ Collections.singletonList("receipt-1"),
sqsClient.deletedReceiptHandles());
+ Assertions.assertEquals(0, context.noMoreElementSignals);
+ }
+
+ @Test
+ void shouldSkipFailedMessageAndContinueWhenParseErrorsIgnored() throws
Exception {
+ RecordingSqsClient sqsClient =
+ new RecordingSqsClient(
+ message("order-1", "receipt-1"),
+ message("invalid", "receipt-2"),
+ message("order-3", "receipt-3"));
+ RecordingReaderContext context = new RecordingReaderContext();
+ AmazonSqsSourceReader reader =
+ createReader(
+ context,
+ true,
+ true,
+ new SelectiveFailingDeserializationSchema(),
+ sqsClient);
+ RecordingCollector collector = new RecordingCollector();
+
+ reader.pollNext(collector);
+
+ Assertions.assertEquals(Arrays.asList("order-1", "order-3"),
collector.values());
+ Assertions.assertEquals(
+ Arrays.asList("receipt-1", "receipt-2", "receipt-3"),
+ sqsClient.deletedReceiptHandles());
+ Assertions.assertEquals(1, context.noMoreElementSignals);
+ }
+
+ @Test
+ void shouldKeepIgnoredNullMessageWhenDeletionDisabled() throws Exception {
+ RecordingSqsClient sqsClient = new
RecordingSqsClient(message("invalid", "receipt-1"));
+ RecordingReaderContext context = new RecordingReaderContext();
+ AmazonSqsSourceReader reader =
+ createReader(context, false, true, new
NullDeserializationSchema(), sqsClient);
+ RecordingCollector collector = new RecordingCollector();
+
+ reader.pollNext(collector);
+
+ Assertions.assertTrue(collector.records.isEmpty());
+ Assertions.assertTrue(sqsClient.deletedRequests.isEmpty());
+ Assertions.assertEquals(1, context.noMoreElementSignals);
+ }
+
+ @Test
+ void shouldApplyIgnoreParseErrorsToJsonMessagesCreatedByFactory() throws
Exception {
+ RecordingSqsClient sqsClient =
+ new RecordingSqsClient(
+ message("invalid", "receipt-1"),
+ message("{\"value\":\"order-2\"}", "receipt-2"));
+ RecordingReaderContext context = new RecordingReaderContext();
+
+ Map<String, Object> fields = new HashMap<>();
+ fields.put("value", "string");
+ Map<String, Object> schema = new HashMap<>();
+ schema.put("fields", fields);
+ Map<String, Object> options = new HashMap<>();
+ options.put("url",
"https://sqs.us-east-1.amazonaws.com/123456789012/orders";);
+ options.put("region", "us-east-1");
+ options.put("schema", schema);
+ options.put("delete_message", true);
+ options.put("ignore_parse_errors", true);
+
+ TableSource<?, ?, ?> tableSource =
+ new AmazonSqsSourceFactory()
+ .createSource(
+ new TableSourceFactoryContext(
+ ReadonlyConfig.fromMap(options),
+
Thread.currentThread().getContextClassLoader()));
+ AmazonSqsSource source = (AmazonSqsSource) tableSource.createSource();
+ AmazonSqsSourceReader reader =
+ (AmazonSqsSourceReader) source.createReader(new
SingleSplitReaderContext(context));
+ reader.sqsClient = sqsClient.client;
+ RecordingCollector collector = new RecordingCollector();
+
+ reader.pollNext(collector);
+
+ Assertions.assertEquals(Collections.singletonList("order-2"),
collector.values());
+ Assertions.assertEquals(
+ Arrays.asList("receipt-1", "receipt-2"),
sqsClient.deletedReceiptHandles());
+ Assertions.assertEquals(1, context.noMoreElementSignals);
+ }
+
+ private static AmazonSqsSourceReader createReader(
+ RecordingReaderContext context,
+ boolean deleteMessage,
+ DeserializationSchema<SeaTunnelRow> deserializationSchema,
+ RecordingSqsClient sqsClient) {
+ return createReader(context, deleteMessage, false,
deserializationSchema, sqsClient);
+ }
+
+ private static AmazonSqsSourceReader createReader(
+ RecordingReaderContext context,
+ boolean deleteMessage,
+ boolean ignoreParseErrors,
+ DeserializationSchema<SeaTunnelRow> deserializationSchema,
+ RecordingSqsClient sqsClient) {
+ AmazonSqsSourceConfig config =
+ new AmazonSqsSourceConfig(
+
"https://sqs.us-east-1.amazonaws.com/123456789012/orders";,
+ "us-east-1",
+ null,
+ null,
+ null,
+ deleteMessage,
+ ignoreParseErrors,
+ null);
+ AmazonSqsSourceReader reader =
+ new AmazonSqsSourceReader(
+ new SingleSplitReaderContext(context),
+ config,
+ deserializationSchema,
+ ROW_TYPE);
+ reader.sqsClient = sqsClient.client;
+ return reader;
+ }
+
+ private static Message message(String body, String receiptHandle) {
+ return
Message.builder().body(body).receiptHandle(receiptHandle).build();
+ }
+
+ private static class TestDeserializationSchema implements
DeserializationSchema<SeaTunnelRow> {
+
+ @Override
+ public SeaTunnelRow deserialize(byte[] message) throws IOException {
+ return new SeaTunnelRow(new Object[] {new String(message,
StandardCharsets.UTF_8)});
+ }
+
+ @Override
+ public SeaTunnelDataType<SeaTunnelRow> getProducedType() {
+ return ROW_TYPE;
+ }
+ }
+
+ private static final class FailingDeserializationSchema extends
TestDeserializationSchema {
+ @Override
+ public SeaTunnelRow deserialize(byte[] message) throws IOException {
+ throw new IOException("invalid payload");
+ }
+ }
+
+ private static final class NullDeserializationSchema extends
TestDeserializationSchema {
+ @Override
+ public SeaTunnelRow deserialize(byte[] message) throws IOException {
+ return null;
+ }
+ }
+
+ private static final class SelectiveFailingDeserializationSchema
+ extends TestDeserializationSchema {
+ @Override
+ public SeaTunnelRow deserialize(byte[] message) throws IOException {
+ String value = new String(message, StandardCharsets.UTF_8);
+ if (value.equals("invalid")) {
+ throw new IOException("invalid payload");
+ }
+ return super.deserialize(message);
+ }
+ }
+
+ private static final class RecordingCollector implements
Collector<SeaTunnelRow> {
+ private final List<SeaTunnelRow> records = new ArrayList<>();
+ private final String failureMessage;
+
+ private RecordingCollector() {
+ this(null);
+ }
+
+ private RecordingCollector(String failureMessage) {
+ this.failureMessage = failureMessage;
+ }
+
+ @Override
+ public void collect(SeaTunnelRow record) {
+ if (failureMessage != null) {
+ throw new IllegalStateException(failureMessage);
+ }
+ records.add(record);
+ }
+
+ @Override
+ public Object getCheckpointLock() {
+ return this;
+ }
+
+ private List<String> values() {
+ List<String> values = new ArrayList<>();
+ for (SeaTunnelRow record : records) {
+ values.add(record.getField(0).toString());
+ }
+ return values;
+ }
+ }
+
+ private static final class RecordingReaderContext implements
SourceReader.Context {
+ private int noMoreElementSignals;
+
+ @Override
+ public int getIndexOfSubtask() {
+ return 0;
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.BOUNDED;
+ }
+
+ @Override
+ public void signalNoMoreElement() {
+ noMoreElementSignals++;
+ }
+
+ @Override
+ public void sendSplitRequest() {}
+
+ @Override
+ public void sendSourceEventToEnumerator(SourceEvent sourceEvent) {}
+
+ @Override
+ public MetricsContext getMetricsContext() {
+ return null;
+ }
+
+ @Override
+ public EventListener getEventListener() {
+ return null;
+ }
+ }
+
+ private static final class RecordingSqsClient implements InvocationHandler
{
+ private final SqsClient client;
+ private final List<Message> messages;
+ private final List<DeleteMessageRequest> deletedRequests = new
ArrayList<>();
+
+ private RecordingSqsClient(Message... messages) {
+ this.messages =
Collections.unmodifiableList(Arrays.asList(messages));
+ this.client =
+ (SqsClient)
+ Proxy.newProxyInstance(
+ SqsClient.class.getClassLoader(),
+ new Class<?>[] {SqsClient.class},
+ this);
+ }
+
+ @Override
+ public Object invoke(Object proxy, Method method, Object[] args) {
+ if (method.getDeclaringClass() == Object.class) {
+ return invokeObjectMethod(proxy, method, args);
+ }
+ if (method.getName().equals("receiveMessage")) {
+ return
ReceiveMessageResponse.builder().messages(messages).build();
+ }
+ if (method.getName().equals("deleteMessage")) {
+ deletedRequests.add((DeleteMessageRequest) args[0]);
+ return DeleteMessageResponse.builder().build();
+ }
+ if (method.getName().equals("close")) {
+ return null;
+ }
+ throw new UnsupportedOperationException(method.getName());
+ }
+
+ private Object invokeObjectMethod(Object proxy, Method method,
Object[] args) {
+ if (method.getName().equals("toString")) {
+ return "RecordingSqsClient";
+ }
+ if (method.getName().equals("hashCode")) {
+ return System.identityHashCode(proxy);
+ }
+ if (method.getName().equals("equals")) {
+ return proxy == args[0];
+ }
+ throw new UnsupportedOperationException(method.getName());
+ }
+
+ private List<String> deletedReceiptHandles() {
+ List<String> receiptHandles = new ArrayList<>();
+ for (DeleteMessageRequest request : deletedRequests) {
+ receiptHandles.add(request.receiptHandle());
+ }
+ return receiptHandles;
+ }
+ }
+}
