This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12084-aa24c7e9e59ab0d8cba785bbb974fa83ebfd7164 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 83454b590dbf9442faecb28974b815fbb6ae2b82 Author: Goutam Adwant <[email protected]> AuthorDate: Fri Sep 4 21:47:44 2026 +0000 [Fix][Connector-V2][AmazonSqs] Support multi-row CDC deserialization (#12084) --- docs/en/connectors/source/AmazonSqs.md | 2 + docs/zh/connectors/source/AmazonSqs.md | 2 + .../deserialize/AmazonSqsDeserializer.java | 69 +++++- .../deserialize/SeaTunnelRowDeserializer.java | 10 + .../amazonsqs/source/AmazonSqsSource.java | 14 +- .../amazonsqs/source/AmazonSqsSourceFactory.java | 14 +- .../amazonsqs/source/AmazonSqsSourceReader.java | 22 +- .../source/AmazonSqsSourceReaderTest.java | 241 ++++++++++++++++++++- 8 files changed, 359 insertions(+), 15 deletions(-) diff --git a/docs/en/connectors/source/AmazonSqs.md b/docs/en/connectors/source/AmazonSqs.md index 443d8434ef..44efc844b8 100644 --- a/docs/en/connectors/source/AmazonSqs.md +++ b/docs/en/connectors/source/AmazonSqs.md @@ -54,8 +54,10 @@ 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). +- A `canal_json` or `debezium_json` message can emit multiple rows. For example, an update emits its before and after rows. - `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. +- For multi-row messages, deletion happens only after every row has been collected. A collection failure keeps the SQS message 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 81f2f7db59..c518aab808 100644 --- a/docs/zh/connectors/source/AmazonSqs.md +++ b/docs/zh/connectors/source/AmazonSqs.md @@ -53,8 +53,10 @@ 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)。 +- 一条 `canal_json` 或 `debezium_json` 消息可能产生多行。例如，更新事件会产生更新前和更新后的行。 - `ignore_parse_errors = false` 会让本次轮询失败并保留无法解析的消息。设置为 `true` 时，源连接器会跳过该消息并继续处理本批次中的其他消息。 - 当 `ignore_parse_errors` 和 `delete_message` 都为 `true` 时，跳过的消息会从 SQS 中删除。如果需要保留这些消息以便重新投递，请保持 `delete_message = false`。 +- 对于产生多行的消息，只有在所有行都成功收集后才会删除消息。收集失败时，SQS 消息会保留以便重新投递。 - `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/deserialize/AmazonSqsDeserializer.java b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/deserialize/AmazonSqsDeserializer.java index 21314f20ed..45ba43d70d 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 @@ -18,27 +18,46 @@ package org.apache.seatunnel.connectors.seatunnel.amazonsqs.deserialize; import org.apache.seatunnel.api.serialization.DeserializationSchema; +import org.apache.seatunnel.api.source.Collector; import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.common.exception.CommonErrorCode; +import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException; +import org.apache.seatunnel.connectors.seatunnel.amazonsqs.config.MessageFormat; import org.apache.seatunnel.connectors.seatunnel.amazonsqs.exception.AmazonSqsConnectorErrorCode; import org.apache.seatunnel.connectors.seatunnel.amazonsqs.exception.AmazonSqsConnectorException; import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; public class AmazonSqsDeserializer implements SeaTunnelRowDeserializer { private final DeserializationSchema<SeaTunnelRow> deserializationSchema; private final boolean ignoreParseErrors; + private final MessageFormat format; public AmazonSqsDeserializer( DeserializationSchema<SeaTunnelRow> deserializationSchema, boolean ignoreParseErrors) { + this(deserializationSchema, ignoreParseErrors, MessageFormat.JSON); + } + + public AmazonSqsDeserializer( + DeserializationSchema<SeaTunnelRow> deserializationSchema, + boolean ignoreParseErrors, + MessageFormat format) { this.deserializationSchema = deserializationSchema; this.ignoreParseErrors = ignoreParseErrors; + this.format = format; } @Override public SeaTunnelRow deserializeRow(String row) { + byte[] message = row.getBytes(StandardCharsets.UTF_8); try { - SeaTunnelRow seaTunnelRow = deserializationSchema.deserialize(row.getBytes()); + SeaTunnelRow seaTunnelRow = deserializationSchema.deserialize(message); if (seaTunnelRow == null && !ignoreParseErrors) { throw new AmazonSqsConnectorException( AmazonSqsConnectorErrorCode.DESERIALIZE_FAILED, @@ -55,4 +74,52 @@ public class AmazonSqsDeserializer implements SeaTunnelRowDeserializer { e); } } + + @Override + public List<SeaTunnelRow> deserializeRows(String row) { + if (format == MessageFormat.CANAL_JSON || format == MessageFormat.DEBEZIUM_JSON) { + return deserializeMultipleRows(row.getBytes(StandardCharsets.UTF_8)); + } + return SeaTunnelRowDeserializer.super.deserializeRows(row); + } + + private List<SeaTunnelRow> deserializeMultipleRows(byte[] message) { + List<SeaTunnelRow> rows = new ArrayList<>(); + try { + deserializationSchema.deserialize(message, new BufferingCollector(rows)); + return rows; + } catch (IOException e) { + if (ignoreParseErrors) { + return Collections.emptyList(); + } + throw new AmazonSqsConnectorException( + AmazonSqsConnectorErrorCode.DESERIALIZE_FAILED, + "Failed to deserialize Amazon SQS message", + e); + } catch (SeaTunnelRuntimeException e) { + if (ignoreParseErrors + && CommonErrorCode.JSON_OPERATION_FAILED.equals(e.getSeaTunnelErrorCode())) { + return Collections.emptyList(); + } + throw e; + } + } + + private static final class BufferingCollector implements Collector<SeaTunnelRow> { + private final List<SeaTunnelRow> rows; + + private BufferingCollector(List<SeaTunnelRow> rows) { + this.rows = rows; + } + + @Override + public void collect(SeaTunnelRow row) { + rows.add(Objects.requireNonNull(row, "Deserialization schema emitted a null row")); + } + + @Override + public Object getCheckpointLock() { + return this; + } + } } diff --git a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/deserialize/SeaTunnelRowDeserializer.java b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/deserialize/SeaTunnelRowDeserializer.java index e837324343..3d271d43e0 100644 --- a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/deserialize/SeaTunnelRowDeserializer.java +++ b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/deserialize/SeaTunnelRowDeserializer.java @@ -19,7 +19,17 @@ package org.apache.seatunnel.connectors.seatunnel.amazonsqs.deserialize; import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import java.util.Collections; +import java.util.List; + public interface SeaTunnelRowDeserializer { SeaTunnelRow deserializeRow(String row); + + default List<SeaTunnelRow> deserializeRows(String row) { + SeaTunnelRow seaTunnelRow = deserializeRow(row); + return seaTunnelRow == null + ? Collections.emptyList() + : Collections.singletonList(seaTunnelRow); + } } diff --git a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/source/AmazonSqsSource.java b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/source/AmazonSqsSource.java index ee0d078ab1..98a0e2e506 100644 --- a/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/source/AmazonSqsSource.java +++ b/seatunnel-connectors-v2/connector-amazonsqs/src/main/java/org/apache/seatunnel/connectors/seatunnel/amazonsqs/source/AmazonSqsSource.java @@ -23,6 +23,7 @@ import org.apache.seatunnel.api.source.SupportColumnProjection; import org.apache.seatunnel.api.table.catalog.CatalogTable; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.connectors.seatunnel.amazonsqs.config.AmazonSqsSourceConfig; +import org.apache.seatunnel.connectors.seatunnel.amazonsqs.config.MessageFormat; import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader; import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitSource; import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext; @@ -39,14 +40,24 @@ public class AmazonSqsSource extends AbstractSingleSplitSource<SeaTunnelRow> private AmazonSqsSourceConfig amazonSqsSourceConfig; private DeserializationSchema<SeaTunnelRow> deserializationSchema; private CatalogTable catalogTable; + private MessageFormat format; public AmazonSqsSource( AmazonSqsSourceConfig amazonSqsSourceConfig, CatalogTable catalogTable, DeserializationSchema<SeaTunnelRow> deserializationSchema) { + this(amazonSqsSourceConfig, catalogTable, deserializationSchema, MessageFormat.JSON); + } + + AmazonSqsSource( + AmazonSqsSourceConfig amazonSqsSourceConfig, + CatalogTable catalogTable, + DeserializationSchema<SeaTunnelRow> deserializationSchema, + MessageFormat format) { this.amazonSqsSourceConfig = amazonSqsSourceConfig; this.catalogTable = catalogTable; this.deserializationSchema = deserializationSchema; + this.format = format; } @Override @@ -71,6 +82,7 @@ public class AmazonSqsSource extends AbstractSingleSplitSource<SeaTunnelRow> readerContext, amazonSqsSourceConfig, deserializationSchema, - catalogTable.getSeaTunnelRowType()); + catalogTable.getSeaTunnelRowType(), + format); } } 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 3ce5fcaf6b..2441e61247 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 @@ -84,14 +84,16 @@ public class AmazonSqsSourceFactory implements TableSourceFactory { public <T, SplitT extends SourceSplit, StateT extends Serializable> TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) { CatalogTable catalogTable = CatalogTableUtil.buildWithConfig(context.getOptions()); + MessageFormat format = context.getOptions().get(FORMAT); DeserializationSchema<SeaTunnelRow> deserializationSchema = - setDeserialization(context.getOptions().toConfig(), catalogTable); + setDeserialization(context.getOptions().toConfig(), catalogTable, format); return () -> (SeaTunnelSource<T, SplitT, StateT>) new AmazonSqsSource( new AmazonSqsSourceConfig(context.getOptions()), catalogTable, - deserializationSchema); + deserializationSchema, + format); } @Override @@ -100,10 +102,9 @@ public class AmazonSqsSourceFactory implements TableSourceFactory { } private DeserializationSchema<SeaTunnelRow> setDeserialization( - Config config, CatalogTable catalogTable) { + Config config, CatalogTable catalogTable, MessageFormat format) { DeserializationSchema<SeaTunnelRow> deserializationSchema; ReadonlyConfig readonlyConfig = ReadonlyConfig.fromConfig(config); - MessageFormat format = readonlyConfig.get(FORMAT); boolean ignoreParseErrors = readonlyConfig.get(IGNORE_PARSE_ERRORS); switch (format) { case JSON: @@ -124,7 +125,7 @@ public class AmazonSqsSourceFactory implements TableSourceFactory { case CANAL_JSON: deserializationSchema = CanalJsonDeserializationSchema.builder(catalogTable) - .setIgnoreParseErrors(ignoreParseErrors) + .setIgnoreParseErrors(false) .build(); break; case DEBEZIUM_JSON: @@ -133,8 +134,7 @@ public class AmazonSqsSourceFactory implements TableSourceFactory { includeSchema = config.getBoolean(DEBEZIUM_RECORD_INCLUDE_SCHEMA.key()); } deserializationSchema = - new DebeziumJsonDeserializationSchema( - catalogTable, ignoreParseErrors, includeSchema); + new DebeziumJsonDeserializationSchema(catalogTable, false, 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 5b3d361c31..ea76675dbb 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 @@ -22,6 +22,7 @@ import org.apache.seatunnel.api.source.Collector; 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.config.MessageFormat; import org.apache.seatunnel.connectors.seatunnel.amazonsqs.deserialize.AmazonSqsDeserializer; import org.apache.seatunnel.connectors.seatunnel.amazonsqs.deserialize.SeaTunnelRowDeserializer; import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader; @@ -55,11 +56,25 @@ public class AmazonSqsSourceReader extends AbstractSingleSplitReader<SeaTunnelRo AmazonSqsSourceConfig amazonSqsSourceConfig, DeserializationSchema<SeaTunnelRow> deserializationSchema, SeaTunnelRowType seaTunnelRowType) { + this( + context, + amazonSqsSourceConfig, + deserializationSchema, + seaTunnelRowType, + MessageFormat.JSON); + } + + AmazonSqsSourceReader( + SingleSplitReaderContext context, + AmazonSqsSourceConfig amazonSqsSourceConfig, + DeserializationSchema<SeaTunnelRow> deserializationSchema, + SeaTunnelRowType seaTunnelRowType, + MessageFormat format) { this.context = context; this.amazonSqsSourceConfig = amazonSqsSourceConfig; this.seaTunnelRowDeserializer = new AmazonSqsDeserializer( - deserializationSchema, amazonSqsSourceConfig.isIgnoreParseErrors()); + deserializationSchema, amazonSqsSourceConfig.isIgnoreParseErrors(), format); } @Override @@ -108,8 +123,9 @@ public class AmazonSqsSourceReader extends AbstractSingleSplitReader<SeaTunnelRo for (Message message : messages) { String messageBody = message.body(); - SeaTunnelRow seaTunnelRow = this.seaTunnelRowDeserializer.deserializeRow(messageBody); - if (seaTunnelRow != null) { + List<SeaTunnelRow> seaTunnelRows = + this.seaTunnelRowDeserializer.deserializeRows(messageBody); + for (SeaTunnelRow seaTunnelRow : seaTunnelRows) { output.collect(seaTunnelRow); } 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 index ed7cb3d1ef..a61ceaf60b 100644 --- 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 @@ -28,10 +28,14 @@ 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.RowKind; 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.common.exception.SeaTunnelRuntimeException; import org.apache.seatunnel.connectors.seatunnel.amazonsqs.config.AmazonSqsSourceConfig; +import org.apache.seatunnel.connectors.seatunnel.amazonsqs.config.MessageFormat; +import org.apache.seatunnel.connectors.seatunnel.amazonsqs.exception.AmazonSqsConnectorErrorCode; import org.apache.seatunnel.connectors.seatunnel.amazonsqs.exception.AmazonSqsConnectorException; import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext; @@ -58,6 +62,20 @@ import java.util.Map; class AmazonSqsSourceReaderTest { + private static final String CANAL_UPDATE_MESSAGE = + "{\"data\":[{\"value\":\"after\"}],\"old\":[{\"value\":\"before\"}]," + + "\"database\":\"inventory\",\"table\":\"orders\",\"type\":\"UPDATE\"}"; + private static final String CANAL_QUERY_MESSAGE = + "{\"data\":null,\"database\":\"inventory\",\"table\":\"orders\"," + + "\"type\":\"QUERY\"}"; + private static final String CANAL_PARTIALLY_INVALID_MESSAGE = + "{\"data\":[{\"value\":\"1\"},{\"value\":\"invalid\"}]," + + "\"database\":\"inventory\",\"table\":\"orders\",\"type\":\"INSERT\"}"; + private static final String DEBEZIUM_UPDATE_MESSAGE = + "{\"schema\":{},\"payload\":{\"before\":{\"value\":\"before\"}," + + "\"after\":{\"value\":\"after\"},\"op\":\"u\"," + + "\"ts_ms\":1598944202218}}"; + private static final SeaTunnelRowType ROW_TYPE = new SeaTunnelRowType( new String[] {"value"}, new SeaTunnelDataType[] {BasicType.STRING_TYPE}); @@ -254,6 +272,178 @@ class AmazonSqsSourceReaderTest { Assertions.assertEquals(1, context.noMoreElementSignals); } + @Test + void shouldDeserializeCanalUpdateAsTwoRows() throws Exception { + RecordingSqsClient sqsClient = + new RecordingSqsClient(message(CANAL_UPDATE_MESSAGE, "receipt-1")); + RecordingReaderContext context = new RecordingReaderContext(); + AmazonSqsSourceReader reader = + createReaderFromFactory(context, "canal_json", true, false, sqsClient); + RecordingCollector collector = new RecordingCollector(); + + reader.pollNext(collector); + + Assertions.assertEquals(Arrays.asList("before", "after"), collector.values()); + Assertions.assertEquals( + Arrays.asList(RowKind.UPDATE_BEFORE, RowKind.UPDATE_AFTER), collector.rowKinds()); + Assertions.assertEquals( + Collections.singletonList("receipt-1"), sqsClient.deletedReceiptHandles()); + } + + @Test + void shouldDeserializeDebeziumUpdateAsTwoRows() throws Exception { + RecordingSqsClient sqsClient = + new RecordingSqsClient(message(DEBEZIUM_UPDATE_MESSAGE, "receipt-1")); + RecordingReaderContext context = new RecordingReaderContext(); + AmazonSqsSourceReader reader = + createReaderFromFactory(context, "debezium_json", true, false, sqsClient); + RecordingCollector collector = new RecordingCollector(); + + reader.pollNext(collector); + + Assertions.assertEquals(Arrays.asList("before", "after"), collector.values()); + Assertions.assertEquals( + Arrays.asList(RowKind.UPDATE_BEFORE, RowKind.UPDATE_AFTER), collector.rowKinds()); + Assertions.assertEquals( + Collections.singletonList("receipt-1"), sqsClient.deletedReceiptHandles()); + } + + @Test + void shouldDeleteCanalMessageThatProducesNoRows() throws Exception { + RecordingSqsClient sqsClient = + new RecordingSqsClient(message(CANAL_QUERY_MESSAGE, "receipt-1")); + RecordingReaderContext context = new RecordingReaderContext(); + AmazonSqsSourceReader reader = + createReaderFromFactory(context, "canal_json", true, false, sqsClient); + RecordingCollector collector = new RecordingCollector(); + + reader.pollNext(collector); + + Assertions.assertTrue(collector.records.isEmpty()); + Assertions.assertEquals( + Collections.singletonList("receipt-1"), sqsClient.deletedReceiptHandles()); + } + + @Test + void shouldNotDeleteCanalMessageWhenSecondCollectionFails() throws Exception { + RecordingSqsClient sqsClient = + new RecordingSqsClient(message(CANAL_UPDATE_MESSAGE, "receipt-1")); + RecordingReaderContext context = new RecordingReaderContext(); + AmazonSqsSourceReader reader = + createReaderFromFactory(context, "canal_json", true, false, sqsClient); + RecordingCollector collector = new RecordingCollector("collector rejected row", 1); + + IllegalStateException exception = + Assertions.assertThrows( + IllegalStateException.class, () -> reader.pollNext(collector)); + + Assertions.assertEquals("collector rejected row", exception.getMessage()); + Assertions.assertEquals(Collections.singletonList("before"), collector.values()); + Assertions.assertTrue(sqsClient.deletedRequests.isEmpty()); + } + + @Test + void shouldDiscardBufferedRowsWhenLaterCanalRowCannotBeParsed() throws Exception { + RecordingSqsClient sqsClient = + new RecordingSqsClient(message(CANAL_PARTIALLY_INVALID_MESSAGE, "receipt-1")); + RecordingReaderContext context = new RecordingReaderContext(); + AmazonSqsSourceReader reader = + createReaderFromFactory(context, "canal_json", true, true, "int", sqsClient); + RecordingCollector collector = new RecordingCollector(); + + reader.pollNext(collector); + + Assertions.assertTrue(collector.records.isEmpty()); + Assertions.assertEquals( + Collections.singletonList("receipt-1"), sqsClient.deletedReceiptHandles()); + } + + @Test + void shouldKeepPartiallyInvalidCanalMessageWhenParseErrorsAreNotIgnored() throws Exception { + RecordingSqsClient sqsClient = + new RecordingSqsClient(message(CANAL_PARTIALLY_INVALID_MESSAGE, "receipt-1")); + RecordingReaderContext context = new RecordingReaderContext(); + AmazonSqsSourceReader reader = + createReaderFromFactory(context, "canal_json", true, false, "int", sqsClient); + RecordingCollector collector = new RecordingCollector(); + + Assertions.assertThrows(SeaTunnelRuntimeException.class, () -> reader.pollNext(collector)); + + Assertions.assertTrue(collector.records.isEmpty()); + Assertions.assertTrue(sqsClient.deletedRequests.isEmpty()); + } + + @Test + void shouldNotSwallowUnrelatedRuntimeFailureForMultiRowFormat() { + RecordingSqsClient sqsClient = + new RecordingSqsClient(message(CANAL_UPDATE_MESSAGE, "receipt-1")); + RecordingReaderContext context = new RecordingReaderContext(); + AmazonSqsConnectorException failure = + new AmazonSqsConnectorException( + AmazonSqsConnectorErrorCode.DESERIALIZE_FAILED, "unexpected failure"); + AmazonSqsSourceReader reader = + createReader( + context, + true, + true, + new RuntimeFailingDeserializationSchema(failure), + sqsClient, + MessageFormat.CANAL_JSON); + RecordingCollector collector = new RecordingCollector(); + + AmazonSqsConnectorException actual = + Assertions.assertThrows( + AmazonSqsConnectorException.class, () -> reader.pollNext(collector)); + + Assertions.assertSame(failure, actual); + Assertions.assertTrue(collector.records.isEmpty()); + Assertions.assertTrue(sqsClient.deletedRequests.isEmpty()); + } + + private static AmazonSqsSourceReader createReaderFromFactory( + RecordingReaderContext context, + String format, + boolean deleteMessage, + boolean ignoreParseErrors, + RecordingSqsClient sqsClient) + throws Exception { + return createReaderFromFactory( + context, format, deleteMessage, ignoreParseErrors, "string", sqsClient); + } + + private static AmazonSqsSourceReader createReaderFromFactory( + RecordingReaderContext context, + String format, + boolean deleteMessage, + boolean ignoreParseErrors, + String fieldType, + RecordingSqsClient sqsClient) + throws Exception { + Map<String, Object> fields = new HashMap<>(); + fields.put("value", fieldType); + 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("format", format); + options.put("delete_message", deleteMessage); + options.put("ignore_parse_errors", ignoreParseErrors); + + 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; + return reader; + } + private static AmazonSqsSourceReader createReader( RecordingReaderContext context, boolean deleteMessage, @@ -268,6 +458,22 @@ class AmazonSqsSourceReaderTest { boolean ignoreParseErrors, DeserializationSchema<SeaTunnelRow> deserializationSchema, RecordingSqsClient sqsClient) { + return createReader( + context, + deleteMessage, + ignoreParseErrors, + deserializationSchema, + sqsClient, + MessageFormat.JSON); + } + + private static AmazonSqsSourceReader createReader( + RecordingReaderContext context, + boolean deleteMessage, + boolean ignoreParseErrors, + DeserializationSchema<SeaTunnelRow> deserializationSchema, + RecordingSqsClient sqsClient, + MessageFormat format) { AmazonSqsSourceConfig config = new AmazonSqsSourceConfig( "https://sqs.us-east-1.amazonaws.com/123456789012/orders";, @@ -283,7 +489,8 @@ class AmazonSqsSourceReaderTest { new SingleSplitReaderContext(context), config, deserializationSchema, - ROW_TYPE); + ROW_TYPE, + format); reader.sqsClient = sqsClient.client; return reader; } @@ -331,21 +538,41 @@ class AmazonSqsSourceReaderTest { } } + private static final class RuntimeFailingDeserializationSchema + extends TestDeserializationSchema { + private final AmazonSqsConnectorException failure; + + private RuntimeFailingDeserializationSchema(AmazonSqsConnectorException failure) { + this.failure = failure; + } + + @Override + public void deserialize(byte[] message, Collector<SeaTunnelRow> output) { + throw failure; + } + } + private static final class RecordingCollector implements Collector<SeaTunnelRow> { private final List<SeaTunnelRow> records = new ArrayList<>(); private final String failureMessage; + private final int successfulCollectionsBeforeFailure; private RecordingCollector() { - this(null); + this(null, 0); } private RecordingCollector(String failureMessage) { + this(failureMessage, 0); + } + + private RecordingCollector(String failureMessage, int successfulCollectionsBeforeFailure) { this.failureMessage = failureMessage; + this.successfulCollectionsBeforeFailure = successfulCollectionsBeforeFailure; } @Override public void collect(SeaTunnelRow record) { - if (failureMessage != null) { + if (failureMessage != null && records.size() >= successfulCollectionsBeforeFailure) { throw new IllegalStateException(failureMessage); } records.add(record); @@ -363,6 +590,14 @@ class AmazonSqsSourceReaderTest { } return values; } + + private List<RowKind> rowKinds() { + List<RowKind> rowKinds = new ArrayList<>(); + for (SeaTunnelRow record : records) { + rowKinds.add(record.getRowKind()); + } + return rowKinds; + } } private static final class RecordingReaderContext implements SourceReader.Context {
