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-13094-526d06802ef7f00016c5fb94e9e6af1d2b0d60da in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit eab2443be5ff98850e6c4c3fe451eee9291ff6a1 Author: Goutam Adwant <[email protected]> AuthorDate: Wed Nov 4 15:45:55 2026 +0000 [Feature][Connector-V2] Add Mailchimp source (#13094) Signed-off-by: Goutam Adwant <[email protected]> --- .github/workflows/labeler/label-scope-conf.yml | 10 + config/plugin_config | 1 + .../changelog/connector-http-mailchimp.md | 20 + docs/en/connectors/source/Mailchimp.md | 140 ++++++ .../changelog/connector-http-mailchimp.md | 20 + docs/zh/connectors/source/Mailchimp.md | 132 +++++ plugin-mapping.properties | 1 + .../connector-http-mailchimp/pom.xml | 51 ++ .../seatunnel/mailchimp/MailchimpConfig.java | 104 ++++ .../seatunnel/mailchimp/MailchimpSource.java | 72 +++ .../mailchimp/MailchimpSourceFactory.java | 75 +++ .../mailchimp/MailchimpSourceOptions.java | 71 +++ .../seatunnel/mailchimp/MailchimpSourceReader.java | 321 ++++++++++++ .../seatunnel/mailchimp/MailchimpConfigTest.java | 165 +++++++ .../mailchimp/MailchimpSourceFactoryTest.java | 37 ++ .../mailchimp/MailchimpSourceReaderTest.java | 543 +++++++++++++++++++++ seatunnel-connectors-v2/connector-http/pom.xml | 1 + seatunnel-dist/pom.xml | 6 + .../connector-mailchimp-e2e/pom.xml | 53 ++ .../e2e/connector/mailchimp/MailchimpIT.java | 148 ++++++ .../test/resources/mailchimp-members-page-1.json | 23 + .../test/resources/mailchimp-members-page-2.json | 14 + .../test/resources/mailchimp_source_to_assert.conf | 59 +++ seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml | 1 + 24 files changed, 2068 insertions(+) diff --git a/.github/workflows/labeler/label-scope-conf.yml b/.github/workflows/labeler/label-scope-conf.yml index 098a37ee6b..3df6c438c6 100644 --- a/.github/workflows/labeler/label-scope-conf.yml +++ b/.github/workflows/labeler/label-scope-conf.yml @@ -241,6 +241,16 @@ hive: - changed-files: - any-glob-to-any-file: seatunnel-connectors-v2/connector-hive/** - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(hive)/**' +mailchimp: + - all: + - changed-files: + - any-glob-to-any-file: + - seatunnel-connectors-v2/connector-http/connector-http-mailchimp/** + - seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/** + - all-globs-to-all-files: + - '!seatunnel-connectors-v2/connector-http/connector-http-!(mailchimp)/**' + - '!seatunnel-e2e/seatunnel-connector-v2-e2e/connector-!(mailchimp)-e2e/**' + http: - all: - changed-files: diff --git a/config/plugin_config b/config/plugin_config index 07fd5a7a43..acf41a0c43 100644 --- a/config/plugin_config +++ b/config/plugin_config @@ -74,6 +74,7 @@ connector-mem0 connector-http-asana connector-http-intercom connector-http-typeform +connector-http-mailchimp connector-http-feishu connector-http-gitlab connector-http-github diff --git a/docs/en/connectors/changelog/connector-http-mailchimp.md b/docs/en/connectors/changelog/connector-http-mailchimp.md new file mode 100644 index 0000000000..7663d21767 --- /dev/null +++ b/docs/en/connectors/changelog/connector-http-mailchimp.md @@ -0,0 +1,20 @@ +<!-- +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. +--> + +| Change | Commit | Version | +| --- | --- | --- | +| [Feature][connector-http-mailchimp] Add bounded Mailchimp list members source | https://github.com/apache/seatunnel | dev | diff --git a/docs/en/connectors/source/Mailchimp.md b/docs/en/connectors/source/Mailchimp.md new file mode 100644 index 0000000000..9cc4e5f99b --- /dev/null +++ b/docs/en/connectors/source/Mailchimp.md @@ -0,0 +1,140 @@ +import ChangeLog from '../changelog/connector-http-mailchimp.md'; + +# Http-Mailchimp + +<!-- +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. +--> + +> Mailchimp list members source connector + +## Description + +Reads the members of one configured audience using the Mailchimp Marketing API +[`GET /3.0/lists/{list_id}/members`](https://mailchimp.com/developer/marketing/api/lists/members/list-members/). +Only bounded batch reads are supported. Campaigns, reports, OAuth and CDC are not supported. +Uses the existing HTTP and JSON dependencies; no Mailchimp SDK is required. + +## Key Features + +- [x] Batch +- [ ] Stream +- [ ] Exactly-once +- [ ] Column projection +- [ ] Parallelism + +Set source parallelism to 1. + +## Options + +| Name | Type | Required | Default | Description | +| --- | --- | --- | --- | --- | +| token | String | Yes | - | Mailchimp API key including its datacenter suffix, for example `<key>-us1`. Not an OAuth token. | +| list_id | String | Yes | - | One audience/list ID. Letters, digits, underscores and hyphens only. | +| schema | Config | Yes | - | Nonempty output schema using SeaTunnel's existing JSON format. | +| base_url | String | No | `https://<dc>.api.mailchimp.com/3.0` | Explicitly trusted API root. The default datacenter comes from `token`. | +| allow_insecure_http | Boolean | No | `false` | Allow HTTP only for isolated test fixtures with dummy credentials. Never enable with real API keys. | +| page_size | Int | No | `1000` | Members per page, from 1 to 1000. | +| retry | Int | No | `3` | Additional attempts per page on HTTP 429 or 5xx, from 0 to 10. | +| retry_backoff_ms | Int | No | `1000` | Initial retry delay, from 1 to 60000 milliseconds. Doubles up to 60000 milliseconds. | +| request_timeout_ms | Int | No | `30000` | Connection, pool acquisition and socket inactivity timeout, from 1 to 300000 milliseconds. Not a total batch deadline. | + +### Authentication and Endpoint Trust + +The `token` option holds an API key and uses SeaTunnel's existing sensitive-option masking. +It is sent only in the HTTP Basic Authorization header (username `seatunnel`, password the API key). +The connector does not log request headers, response bodies or member data, and its split state +contains no credentials or member data. Standard engine configuration and secret-management rules +still apply. Do not enable HTTP wire/header logging with real credentials. + +A custom `base_url` is an administrator-selected credential destination: use only an API endpoint +you trust. It must use HTTPS and end in `/3.0` (an optional trailing slash is accepted), with a host +and optional port, but no user information, query, fragment or other path. HTTP requires explicit +`allow_insecure_http = true` and is for isolated dummy-credential fixtures only. +Redirects, cookies and automatic HTTP-client retries are disabled; a redirect fails without forwarding +credentials. Response-provided links are never followed. + +### Schema + +Each member object is decoded independently using the existing JSON format. Typical fields are +`id`, `email_address`, `status`, `last_changed` and `merge_fields`. +Field names match the API JSON keys. `last_changed` can be read as a STRING to retain its ISO-8601 +offset. Missing optional fields become null; invalid type conversions fail without including the +member data in the error. + +`merge_fields` is not flattened. Use a nested ROW for known heterogeneous fields or a MAP when +values have a consistent type. The example uses `map<string,string>`; configure a matching schema +for your actual merge-field definitions. A field declared STRING uses the existing JSON format's +conversion rules, not a connector-specific JSON serializer. + +## Example + +```hocon +env { + parallelism = 1 + job.mode = BATCH +} +source { + Mailchimp { + token = "<MAILCHIMP_API_KEY-us1>" + list_id = "<LIST_ID>" + page_size = 1000 + schema { + fields { + id = string + email_address = string + status = string + last_changed = string + merge_fields = "map<string,string>" + } + } + } +} +sink { + Console {} +} +``` + +## Pagination, Recovery and Consistency + +Requests send `count` (at most 1000) and a zero-based `offset`. The first successful page's +`total_items` sets the batch's finite upper bound. Later growth cannot extend that bound. +The final request reduces `count` to the remaining number of members. + +Each page must contain a `members` array and a nonnegative integer `total_items` within the API's +integer offset range. A page must return the expected number of member objects. A short/empty +page before completion, shrinking total, invalid JSON, unexpected shape or schema conversion +failure fails the batch instead of silently reporting success. Responses are limited to 32 MiB +per page; lower `page_size` for large merge fields. + +This is **not a transactional or exact snapshot**. Concurrent additions, deletions and updates can +shift offsets, duplicate members or omit members even when counts match. For a stable export, avoid +audience changes during the run. No `since_last_changed` filter or durable incremental cursor is +provided. + +Recovery deliberately replays the **whole batch from offset zero**, including previously accepted +pages. This is at-least-once batch replay, not exactly-once delivery or a guarantee of capturing +every version of a changing member. Use member `id` for downstream deduplication/upserts. +Checkpoints do not persist pagination offsets or a completed-batch marker. Network waits do not +hold the collector's checkpoint lock. + +HTTP 429 and 5xx responses are retried at the same offset up to `retry` times. `Retry-After` +seconds or HTTP dates are honored with a maximum wait of 60 seconds; invalid values use exponential +backoff. Other non-200 responses (including 401 and redirects), malformed bodies and transport +failures fail immediately. Restarting the job replays the batch. Cancellation during reading aborts active I/O +and wakes backoff waits without signalling successful completion. + +<ChangeLog /> diff --git a/docs/zh/connectors/changelog/connector-http-mailchimp.md b/docs/zh/connectors/changelog/connector-http-mailchimp.md new file mode 100644 index 0000000000..7663d21767 --- /dev/null +++ b/docs/zh/connectors/changelog/connector-http-mailchimp.md @@ -0,0 +1,20 @@ +<!-- +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. +--> + +| Change | Commit | Version | +| --- | --- | --- | +| [Feature][connector-http-mailchimp] Add bounded Mailchimp list members source | https://github.com/apache/seatunnel | dev | diff --git a/docs/zh/connectors/source/Mailchimp.md b/docs/zh/connectors/source/Mailchimp.md new file mode 100644 index 0000000000..edb61f6e42 --- /dev/null +++ b/docs/zh/connectors/source/Mailchimp.md @@ -0,0 +1,132 @@ +import ChangeLog from '../changelog/connector-http-mailchimp.md'; + +# Http-Mailchimp + +<!-- +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. +--> + +> Mailchimp 列表成员源连接器 + +## 描述 + +通过 Mailchimp Marketing API 的 +[`GET /3.0/lists/{list_id}/members`](https://mailchimp.com/developer/marketing/api/lists/members/list-members/) +读取一个指定受众列表的成员。仅支持有界批处理，不支持活动、报表、OAuth 或 CDC。 +复用已有 HTTP 和 JSON 依赖，不引入 Mailchimp SDK。 + +## 主要特性 + +- [x] 批处理 +- [ ] 流处理 +- [ ] 精确一次 +- [ ] 列投影 +- [ ] 并行读取 + +源并行度必须为 1。 + +## 配置项 + +| 名称 | 类型 | 必填 | 默认值 | 描述 | +| --- | --- | --- | --- | --- | +| token | String | 是 | - | 包含数据中心后缀的 Mailchimp API 密钥，例如 `<key>-us1`。不是 OAuth 令牌。 | +| list_id | String | 是 | - | 一个受众列表 ID，仅支持字母、数字、下划线及连字符。 | +| schema | Config | 是 | - | 非空输出结构，使用 SeaTunnel 已有 JSON 格式。 | +| base_url | String | 否 | `https://<dc>.api.mailchimp.com/3.0` | 显式信任的 API 根地址。默认数据中心从 `token` 提取。 | +| allow_insecure_http | Boolean | 否 | `false` | 仅为隔离测试及虚拟凭据允许 HTTP。禁止用于真实 API 密钥。 | +| page_size | Int | 否 | `1000` | 每页成员数，范围 1 到 1000。 | +| retry | Int | 否 | `3` | 每页遇到 HTTP 429 或 5xx 时额外重试次数，范围 0 到 10。 | +| retry_backoff_ms | Int | 否 | `1000` | 初始重试间隔，范围 1 到 60000 毫秒，逐次加倍，最多 60000 毫秒。 | +| request_timeout_ms | Int | 否 | `30000` | 连接、连接池获取及套接字空闲超时，范围 1 到 300000 毫秒，并非整个批次的截止时间。 | + +### 认证与端点信任 + +`token` 保存 API 密钥，复用 SeaTunnel 已有的敏感配置脱敏规则。 +密钥仅通过 HTTP Basic Authorization 请求头发送，用户名为 `seatunnel`，密码为 API 密钥。 +连接器不记录请求头、响应正文或成员数据，分片状态不包含凭据及成员数据。 +仍需遵循引擎配置和密钥管理规则，使用真实凭据时不要开启 HTTP wire/header 调试日志。 + +自定义 `base_url` 是管理员指定的凭据接收端，必须信任该地址。地址必须使用 HTTPS， +包含主机和可选端口，路径为 `/3.0`（允许末尾斜杠），不允许用户信息、查询、片段或其他路径。 +HTTP 必须显式设置 `allow_insecure_http = true`，且仅用于虚拟凭据的隔离测试。 +连接器禁用重定向、Cookie 和 HTTP 客户端自动重试。重定向直接失败，不转发凭据。 +不访问响应中的链接。 + +### 输出结构 + +按成员对象逐条使用已有 JSON 格式解析。常用字段包括 `id`、`email_address`、`status`、 +`last_changed` 和 `merge_fields`。字段名与 API 的 JSON 键一致。 +可将 `last_changed` 设为 STRING，以保留 ISO-8601 时区偏移。 +缺失的可选字段返回 null；类型转换失败会终止读取，错误信息不包含成员数据。 + +`merge_fields` 不会自动展开。已知且类型不同的字段可使用嵌套 ROW；值类型一致时可使用 MAP。 +示例使用 `map<string,string>`，请根据实际合并字段定义配置结构。 +STRING 类型遵循已有 JSON 格式的转换规则，不提供连接器特有的 JSON 序列化。 + +## 示例 + +```hocon +env { + parallelism = 1 + job.mode = BATCH +} +source { + Mailchimp { + token = "<MAILCHIMP_API_KEY-us1>" + list_id = "<LIST_ID>" + page_size = 1000 + schema { + fields { + id = string + email_address = string + status = string + last_changed = string + merge_fields = "map<string,string>" + } + } + } +} +sink { + Console {} +} +``` + +## 分页、恢复和一致性 + +请求使用 `count`（最多 1000）及从零开始的 `offset`。 +首个成功页面的 `total_items` 决定本次批处理的有限上界，后续增长不会扩大该上界。 +最后一次请求将 `count` 调整为剩余成员数。 + +每页必须包含 `members` 数组和 API 整数偏移范围内的非负整数 `total_items`， +并返回预期数量的成员对象。完成前出现短页或空页、总数减少、JSON 无效、响应结构异常 +或字段转换失败时，批次失败，不会静默报告成功。 +单页响应上限为 32 MiB；合并字段较大时请减小 `page_size`。 + +这**不是事务快照或精确快照**。并发新增、删除和更新可能移动偏移量，即使总数相同， +也可能造成重复或遗漏。需要稳定导出时应避免读取期间修改受众列表。 +不提供 `since_last_changed` 过滤或持久化增量游标。 + +恢复时有意从偏移量零**重放整个批次**，包括已经发送的页面。这是至少一次的批次重放， +不是精确一次投递，也不保证捕获持续变化成员的所有版本。 +下游可使用成员 `id` 去重或更新写入。 +检查点不保存分页偏移量或已完成标记，网络等待不持有收集器的检查点锁。 + +HTTP 429 和 5xx 会在相同偏移量最多额外重试 `retry` 次。 +支持秒数或 HTTP 日期格式的 `Retry-After`，每次等待最多 60 秒；无效值使用指数退避。 +其他非 200 响应（包括 401 和重定向）、正文异常及传输错误直接失败。 +重启任务将重放整个批次。读取过程中的取消操作会中止进行中的 I/O 并唤醒退避等待，不报告成功完成。 + +<ChangeLog /> diff --git a/plugin-mapping.properties b/plugin-mapping.properties index 5df09428b5..854ab3b7a0 100644 --- a/plugin-mapping.properties +++ b/plugin-mapping.properties @@ -210,6 +210,7 @@ seatunnel.source.WebSocket = connector-websocket seatunnel.source.Asana = connector-http-asana seatunnel.source.Intercom = connector-http-intercom seatunnel.source.Typeform = connector-http-typeform +seatunnel.source.Mailchimp = connector-http-mailchimp # For custom transforms, make sure to use the seatunnel.transform.[PluginIdentifier]=[JarPerfix] naming convention. For example: # seatunnel.transform.Sql = seatunnel-transforms-v2 diff --git a/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/pom.xml b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/pom.xml new file mode 100644 index 0000000000..5f7b974ae7 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/pom.xml @@ -0,0 +1,51 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + + 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. + +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http</artifactId> + <version>${revision}</version> + </parent> + + <artifactId>connector-http-mailchimp</artifactId> + <name>SeaTunnel : Connectors V2 : Http : Mailchimp</name> + + <dependencies> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-base</artifactId> + <version>${project.version}</version> + </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>seatunnel-common</artifactId> + <version>${project.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>com.squareup.okhttp3</groupId> + <artifactId>mockwebserver</artifactId> + <scope>test</scope> + </dependency> + </dependencies> + +</project> diff --git a/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/main/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpConfig.java b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/main/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpConfig.java new file mode 100644 index 0000000000..5949025586 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/main/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpConfig.java @@ -0,0 +1,104 @@ +/* + * 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.mailchimp; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.configuration.util.ConfigValidator; + +import java.io.Serializable; +import java.net.URI; +import java.net.URISyntaxException; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.ALLOW_INSECURE_HTTP; +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.BASE_URL; +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.LIST_ID; +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.PAGE_SIZE; +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.REQUEST_TIMEOUT_MS; +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.RETRY; +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.RETRY_BACKOFF_MS; +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.TOKEN; + +final class MailchimpConfig implements Serializable { + private static final Pattern KEY_PATTERN = Pattern.compile("[A-Za-z0-9]+-(us[0-9]+)"); + final String apiKey; + final URI membersUri; + final int pageSize; + final int retry; + final int backoffMs; + final int timeoutMs; + + MailchimpConfig(ReadonlyConfig config) { + ConfigValidator.of(config).validate(new MailchimpSourceFactory().optionRule()); + apiKey = config.get(TOKEN); + Matcher key = KEY_PATTERN.matcher(apiKey); + if (!key.matches()) { + throw new IllegalArgumentException( + "Mailchimp token must include a valid datacenter suffix"); + } + String listId = config.get(LIST_ID); + if (!listId.matches("[A-Za-z0-9_-]+")) { + throw new IllegalArgumentException("Mailchimp list_id must be a nonempty identifier"); + } + String base = + config.getOptional(BASE_URL) + .orElse("https://"; + key.group(1) + ".api.mailchimp.com/3.0"); + try { + URI uri = new URI(base); + if ("http".equals(uri.getScheme()) && !config.get(ALLOW_INSECURE_HTTP)) { + throw new IllegalArgumentException( + "Mailchimp base_url requires HTTPS; allow_insecure_http is only for isolated fixtures with dummy credentials"); + } + if ((!"https".equals(uri.getScheme()) && !"http".equals(uri.getScheme())) + || uri.getHost() == null + || uri.getRawUserInfo() != null + || uri.getRawQuery() != null + || uri.getRawFragment() != null + || (uri.getPort() != -1 && (uri.getPort() < 1 || uri.getPort() > 65535)) + || (!"/3.0".equals(uri.getRawPath()) && !"/3.0/".equals(uri.getRawPath()))) { + throw new IllegalArgumentException( + "Mailchimp base_url must be a trusted HTTP(S) API root ending in /3.0, without credentials, query or fragment"); + } + membersUri = + new URI( + uri.getScheme(), + null, + uri.getHost(), + uri.getPort(), + "/3.0/lists/" + listId + "/members", + null, + null); + } catch (URISyntaxException e) { + // URI parser messages can contain credentials from invalid user input. + throw new IllegalArgumentException("Mailchimp base_url is not a valid API root"); + } + pageSize = range(config.get(PAGE_SIZE), 1, 1000, PAGE_SIZE.key()); + retry = range(config.get(RETRY), 0, 10, RETRY.key()); + backoffMs = range(config.get(RETRY_BACKOFF_MS), 1, 60000, RETRY_BACKOFF_MS.key()); + timeoutMs = range(config.get(REQUEST_TIMEOUT_MS), 1, 300000, REQUEST_TIMEOUT_MS.key()); + } + + private static int range(int value, int min, int max, String option) { + if (value < min || value > max) { + throw new IllegalArgumentException( + "Mailchimp " + option + " must be between " + min + " and " + max); + } + return value; + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/main/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSource.java b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/main/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSource.java new file mode 100644 index 0000000000..3858de2a36 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/main/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSource.java @@ -0,0 +1,72 @@ +/* + * 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.mailchimp; + +import org.apache.seatunnel.api.common.JobContext; +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.source.Boundedness; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.CatalogTableUtil; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.common.constants.JobMode; +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; + +import java.util.Collections; +import java.util.List; + +public class MailchimpSource extends AbstractSingleSplitSource<SeaTunnelRow> { + private final MailchimpConfig config; + private final CatalogTable table; + + public MailchimpSource(ReadonlyConfig options) { + config = new MailchimpConfig(options); + table = CatalogTableUtil.buildWithConfig(options); + if (table.getSeaTunnelRowType().getTotalFields() == 0) { + throw new IllegalArgumentException("Mailchimp schema must contain at least one field"); + } + } + + @Override + public String getPluginName() { + return "Mailchimp"; + } + + @Override + public void setJobContext(JobContext context) { + if (context.getJobMode() != JobMode.BATCH) { + throw new IllegalArgumentException("Mailchimp source only supports BATCH mode"); + } + } + + @Override + public Boundedness getBoundedness() { + return Boundedness.BOUNDED; + } + + @Override + public List<CatalogTable> getProducedCatalogTables() { + return Collections.singletonList(table); + } + + @Override + public AbstractSingleSplitReader<SeaTunnelRow> createReader(SingleSplitReaderContext context) { + return new MailchimpSourceReader(config, table, context); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/main/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSourceFactory.java b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/main/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSourceFactory.java new file mode 100644 index 0000000000..8481bb569a --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/main/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSourceFactory.java @@ -0,0 +1,75 @@ +/* + * 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.mailchimp; + +import org.apache.seatunnel.api.configuration.util.OptionRule; +import org.apache.seatunnel.api.options.ConnectorCommonOptions; +import org.apache.seatunnel.api.source.SeaTunnelSource; +import org.apache.seatunnel.api.source.SourceSplit; +import org.apache.seatunnel.api.table.connector.TableSource; +import org.apache.seatunnel.api.table.factory.Factory; +import org.apache.seatunnel.api.table.factory.TableSourceFactory; +import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext; + +import com.google.auto.service.AutoService; + +import java.io.Serializable; + +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.ALLOW_INSECURE_HTTP; +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.BASE_URL; +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.LIST_ID; +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.PAGE_SIZE; +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.REQUEST_TIMEOUT_MS; +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.RETRY; +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.RETRY_BACKOFF_MS; +import static org.apache.seatunnel.connectors.seatunnel.mailchimp.MailchimpSourceOptions.TOKEN; + +@AutoService(Factory.class) +public class MailchimpSourceFactory implements TableSourceFactory { + @Override + public String factoryIdentifier() { + return "Mailchimp"; + } + + @Override + public OptionRule optionRule() { + return OptionRule.builder() + .required(TOKEN, LIST_ID, ConnectorCommonOptions.SCHEMA) + .optional( + BASE_URL, + ALLOW_INSECURE_HTTP, + PAGE_SIZE, + RETRY, + RETRY_BACKOFF_MS, + REQUEST_TIMEOUT_MS) + .build(); + } + + @Override + @SuppressWarnings("unchecked") + public <T, SplitT extends SourceSplit, StateT extends Serializable> + TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) { + MailchimpSource source = new MailchimpSource(context.getOptions()); + return () -> (SeaTunnelSource<T, SplitT, StateT>) source; + } + + @Override + public Class<? extends SeaTunnelSource> getSourceClass() { + return MailchimpSource.class; + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/main/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSourceOptions.java b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/main/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSourceOptions.java new file mode 100644 index 0000000000..dc7d60f620 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/main/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSourceOptions.java @@ -0,0 +1,71 @@ +/* + * 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.mailchimp; + +import org.apache.seatunnel.api.configuration.Option; +import org.apache.seatunnel.api.configuration.Options; + +public final class MailchimpSourceOptions { + public static final Option<String> TOKEN = + Options.key("token") + .stringType() + .noDefaultValue() + .withDescription("Mailchimp API key, with its datacenter suffix."); + public static final Option<String> LIST_ID = + Options.key("list_id") + .stringType() + .noDefaultValue() + .withDescription("Audience/list identifier."); + public static final Option<String> BASE_URL = + Options.key("base_url") + .stringType() + .noDefaultValue() + .withDescription( + "Explicitly trusted API root including /3.0; defaults to the API key datacenter HTTPS endpoint."); + public static final Option<Boolean> ALLOW_INSECURE_HTTP = + Options.key("allow_insecure_http") + .booleanType() + .defaultValue(false) + .withDescription( + "Allow cleartext HTTP only for isolated fixtures with dummy credentials. Never use with real API keys."); + public static final Option<Integer> PAGE_SIZE = + Options.key("page_size") + .intType() + .defaultValue(1000) + .withDescription("Number of members per request, from 1 to 1000."); + public static final Option<Integer> RETRY = + Options.key("retry") + .intType() + .defaultValue(3) + .withDescription( + "Maximum retries per page for HTTP 429 and 5xx, from 0 to 10."); + public static final Option<Integer> RETRY_BACKOFF_MS = + Options.key("retry_backoff_ms") + .intType() + .defaultValue(1000) + .withDescription( + "Initial exponential retry delay in milliseconds, from 1 to 60000; capped at 60000."); + public static final Option<Integer> REQUEST_TIMEOUT_MS = + Options.key("request_timeout_ms") + .intType() + .defaultValue(30000) + .withDescription( + "Connection, pool and socket timeout in milliseconds, from 1 to 300000."); + + private MailchimpSourceOptions() {} +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/main/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSourceReader.java b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/main/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSourceReader.java new file mode 100644 index 0000000000..679cb052f8 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/main/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSourceReader.java @@ -0,0 +1,321 @@ +/* + * 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.mailchimp; + +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.DeserializationFeature; +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode; +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.ObjectMapper; + +import org.apache.seatunnel.api.source.Collector; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader; +import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext; +import org.apache.seatunnel.format.json.JsonDeserializationSchema; + +import org.apache.http.client.config.RequestConfig; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.client.utils.URIBuilder; +import org.apache.http.impl.DefaultConnectionReuseStrategy; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.time.ZonedDateTime; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeParseException; +import java.util.Base64; + +final class MailchimpSourceReader extends AbstractSingleSplitReader<SeaTunnelRow> { + private static final int MAX_PAGE_BYTES = 32 * 1024 * 1024; + private final MailchimpConfig config; + private final SingleSplitReaderContext context; + private final JsonDeserializationSchema deserializer; + private final String tableId; + private final ObjectMapper mapper = + new ObjectMapper() + .enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS) + .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS); + private final Object lifecycle = new Object(); + private volatile boolean closed; + private CloseableHttpClient client; + private HttpGet activeRequest; + + MailchimpSourceReader( + MailchimpConfig config, CatalogTable table, SingleSplitReaderContext context) { + this.config = config; + this.context = context; + deserializer = new JsonDeserializationSchema(table, false, false); + tableId = table.getTablePath().toString(); + } + + @Override + public void open() throws IOException { + synchronized (lifecycle) { + if (closed) { + throw new IOException("Mailchimp reader is closed"); + } + if (client == null) { + RequestConfig timeouts = + RequestConfig.custom() + .setConnectTimeout(config.timeoutMs) + .setSocketTimeout(config.timeoutMs) + .setConnectionRequestTimeout(config.timeoutMs) + .setRedirectsEnabled(false) + .build(); + client = + HttpClients.custom() + .setDefaultRequestConfig(timeouts) + .setConnectionReuseStrategy( + (response, context) -> + response.getStatusLine().getStatusCode() < 400 + && DefaultConnectionReuseStrategy.INSTANCE + .keepAlive(response, context)) + .disableRedirectHandling() + .disableAutomaticRetries() + .disableCookieManagement() + .build(); + } + } + } + + /** + * One bounded batch. Checkpoints deliberately retain an empty split, so a recovered reader + * replays from offset zero instead of treating offsets in a mutable audience as durable + * cursors. + */ + @Override + public void pollNext(Collector<SeaTunnelRow> output) throws Exception { + if (closed || noMoreSplits) { + return; + } + long offset = 0; + long target = -1; + while (!closed) { + int count = + target < 0 ? config.pageSize : (int) Math.min(config.pageSize, target - offset); + JsonNode page = requestPage(offset, count); + JsonNode members = page.get("members"); + JsonNode total = page.get("total_items"); + if (!page.isObject() + || members == null + || !members.isArray() + || total == null + || !total.isIntegralNumber() + || !total.canConvertToInt() + || total.intValue() < 0) { + throw new IOException( + "Mailchimp page must contain members array and nonnegative integer total_items"); + } + if (target < 0) { + target = total.intValue(); + } + if (total.intValue() < target || members.size() != Math.min(count, target - offset)) { + throw new IOException( + "Mailchimp pagination is incomplete or audience changed; retry the batch"); + } + // Validate and convert the entire page before accepting any of its rows. + SeaTunnelRow[] rows = new SeaTunnelRow[members.size()]; + for (int i = 0; i < rows.length; i++) { + if (!members.get(i).isObject()) { + throw new IOException("Mailchimp members must be JSON objects"); + } + try { + rows[i] = deserializer.convertToRowData(members.get(i)); + rows[i].setTableId(tableId); + } catch (Exception e) { + // JSON conversion errors may include member data. Do not retain their causes. + throw new IOException("Mailchimp member does not match the configured schema"); + } + } + for (SeaTunnelRow row : rows) { + synchronized (output.getCheckpointLock()) { + if (closed) { + return; + } + output.collect(row); + } + } + offset += rows.length; + if (offset >= target) { + boolean finished; + synchronized (lifecycle) { + finished = !closed; + if (finished) { + noMoreSplits = true; + } + } + // Completion may wait for a remote acknowledgement; never block cancel on it. + if (finished) { + context.signalNoMoreElement(); + } + return; + } + } + } + + private JsonNode requestPage(long offset, int count) throws Exception { + for (int attempt = 0; ; attempt++) { + HttpGet request = + new HttpGet( + new URIBuilder(config.membersUri) + .addParameter("count", Integer.toString(count)) + .addParameter("offset", Long.toString(offset)) + .build()); + String basic = + Base64.getEncoder() + .encodeToString( + ("seatunnel:" + config.apiKey) + .getBytes(StandardCharsets.UTF_8)); + request.setHeader("Authorization", "Basic " + basic); + request.setHeader("Accept", "application/json"); + CloseableHttpClient httpClient; + synchronized (lifecycle) { + ensureOpen(); + activeRequest = request; + httpClient = client; + } + int status; + long delay = Math.min(60000L, (long) config.backoffMs << attempt); + byte[] body = null; + try (CloseableHttpResponse response = httpClient.execute(request)) { + status = response.getStatusLine().getStatusCode(); + if (status == 200) { + if (response.getEntity() == null) { + throw new IOException("Missing response"); + } + try (InputStream in = response.getEntity().getContent()) { + boolean consumed = false; + try { + body = readPage(in); + consumed = true; + } finally { + if (!consumed) { + // Entity close can drain an unfinished tail. Abort before close. + request.abort(); + } + } + } + } else if (response.getFirstHeader("Retry-After") != null) { + delay = retryDelay(response.getFirstHeader("Retry-After").getValue(), delay); + } + } catch (IOException | RuntimeException e) { + // Transport errors can expose URI values. No response/error content is surfaced. + throw new IOException("Mailchimp request failed or reader was closed"); + } finally { + synchronized (lifecycle) { + activeRequest = null; + } + request.releaseConnection(); + } + if (status == 200) { + try { + JsonNode page = mapper.readTree(body); + if (page == null) { + throw new IOException("Empty response"); + } + return page; + } catch (IOException e) { + throw new IOException("Mailchimp returned invalid JSON"); + } + } + if ((status != 429 && (status < 500 || status > 599)) || attempt >= config.retry) { + throw new IOException("Mailchimp request failed with HTTP " + status); + } + synchronized (lifecycle) { + ensureOpen(); + long deadline = System.nanoTime() + delay * 1_000_000L; + long remaining; + while ((remaining = deadline - System.nanoTime()) > 0) { + try { + lifecycle.wait(Math.max(1, remaining / 1_000_000L)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Mailchimp retry interrupted"); + } + ensureOpen(); + } + } + } + } + + private byte[] readPage(InputStream in) throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + byte[] buffer = new byte[8192]; + int n; + while ((n = in.read(buffer)) != -1) { + if (out.size() > MAX_PAGE_BYTES - n) { + throw new IOException("Mailchimp page exceeds size limit"); + } + out.write(buffer, 0, n); + } + return out.toByteArray(); + } + + private static long retryDelay(String value, long fallback) { + try { + long seconds = Long.parseLong(value); + return seconds < 0 ? fallback : Math.min(seconds, 60) * 1000L; + } catch (NumberFormatException e) { + try { + long millis = + ZonedDateTime.parse(value, DateTimeFormatter.RFC_1123_DATE_TIME) + .toInstant() + .toEpochMilli() + - System.currentTimeMillis(); + return Math.max(1, Math.min(60000, millis)); + } catch (DateTimeParseException ignored) { + return fallback; + } + } + } + + private void ensureOpen() throws IOException { + if (closed || client == null) { + throw new IOException("Mailchimp reader is not open"); + } + } + + @Override + public void cancel() { + try { + close(); + } catch (IOException e) { + throw new IllegalStateException("Mailchimp reader could not close its HTTP client"); + } + } + + /** Abort in-flight I/O and wake retry waits without waiting for the checkpoint lock. */ + @Override + public void close() throws IOException { + synchronized (lifecycle) { + closed = true; + if (activeRequest != null) { + activeRequest.abort(); + } + lifecycle.notifyAll(); + if (client != null) { + client.close(); + } + } + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/test/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpConfigTest.java b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/test/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpConfigTest.java new file mode 100644 index 0000000000..172826c618 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/test/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpConfigTest.java @@ -0,0 +1,165 @@ +/* + * 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.mailchimp; + +import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory; + +import org.apache.seatunnel.api.common.JobContext; +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.source.Boundedness; +import org.apache.seatunnel.api.source.SourceReader; +import org.apache.seatunnel.common.constants.JobMode; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.HashMap; +import java.util.Map; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class MailchimpConfigTest { + static Map<String, Object> options() { + return new HashMap<>( + ConfigFactory.parseString( + "token=testkey-us1\nlist_id=audience1\nschema { fields { " + + "id=string, email_address=string, status=string, last_changed=string, " + + "merge_fields=\"map<string,string>\" } }") + .root() + .unwrapped()); + } + + static ReadonlyConfig config(Map<String, Object> options) { + return ReadonlyConfig.fromMap(options); + } + + @Test + void defaultsAndBoundedMode() throws Exception { + MailchimpConfig cfg = new MailchimpConfig(config(options())); + Assertions.assertEquals( + "https://us1.api.mailchimp.com/3.0/lists/audience1/members";, + cfg.membersUri.toString()); + Assertions.assertEquals(1000, cfg.pageSize); + MailchimpSource source = new MailchimpSource(config(options())); + Assertions.assertEquals(Boundedness.BOUNDED, source.getBoundedness()); + Assertions.assertEquals( + 5, source.getProducedCatalogTables().get(0).getSeaTunnelRowType().getTotalFields()); + JobContext streaming = new JobContext(); + streaming.setJobMode(JobMode.STREAMING); + Assertions.assertThrows( + IllegalArgumentException.class, () -> source.setJobContext(streaming)); + SourceReader.Context context = mock(SourceReader.Context.class); + when(context.getIndexOfSubtask()).thenReturn(1); + Assertions.assertThrows(IllegalArgumentException.class, () -> source.createReader(context)); + } + + @ParameterizedTest + @ValueSource(strings = {"token", "list_id", "schema"}) + void missingRequired(String key) { + Map<String, Object> opts = options(); + opts.remove(key); + Assertions.assertThrows(Exception.class, () -> new MailchimpSource(config(opts))); + } + + @ParameterizedTest + @ValueSource(strings = {"", " ", "secret", "secret-us1\r\nInjected", "secret-us1.evil"}) + void invalidKeysAreNotEchoed(String value) { + Map<String, Object> opts = options(); + opts.put("token", value); + Exception error = + Assertions.assertThrows( + IllegalArgumentException.class, () -> new MailchimpConfig(config(opts))); + Assertions.assertNull(error.getCause()); + Assertions.assertFalse(error.getMessage().contains("secret")); + } + + @ParameterizedTest + @ValueSource(strings = {"", "../other", "x?fields=email_address", "x/y", "x%2fy", "x#fragment"}) + void listCannotChangeRequestTarget(String value) { + Map<String, Object> opts = options(); + opts.put("list_id", value); + Assertions.assertThrows( + IllegalArgumentException.class, () -> new MailchimpConfig(config(opts))); + } + + @ParameterizedTest + @ValueSource( + strings = { + "https://secret@host/3.0";, + "https://host/3.0?secret=1";, + "https://host/3.0#secret";, + "file:///3.0", + "https://host/3.0/evil";, + "https://host:0/3.0";, + "https://host:65536/3.0";, + "https://host/%33.0";, + "https://host/3.0\\evil";, + "not a secret uri" + }) + void rejectsUnsafeUriShapesWithoutEcho(String url) { + Map<String, Object> opts = options(); + opts.put("base_url", url); + Exception error = + Assertions.assertThrows( + IllegalArgumentException.class, () -> new MailchimpConfig(config(opts))); + Assertions.assertFalse(error.getMessage().contains("secret")); + Assertions.assertNull(error.getCause()); + } + + @Test + void explicitEndpointAndNumericBounds() { + Map<String, Object> opts = options(); + opts.put("base_url", "https://trusted.example:8443/3.0/";); + opts.put("page_size", 1); + opts.put("retry", 0); + Assertions.assertEquals( + "https://trusted.example:8443/3.0/lists/audience1/members";, + new MailchimpConfig(config(opts)).membersUri.toString()); + Map<String, int[]> bad = new HashMap<>(); + bad.put("page_size", new int[] {0, 1001}); + bad.put("retry", new int[] {-1, 11}); + bad.put("retry_backoff_ms", new int[] {0, 60001}); + bad.put("request_timeout_ms", new int[] {0, 300001}); + bad.forEach( + (key, values) -> { + for (int value : values) { + Map<String, Object> invalid = options(); + invalid.put(key, value); + Assertions.assertThrows( + IllegalArgumentException.class, + () -> new MailchimpConfig(config(invalid))); + } + }); + opts.put("schema", new HashMap<>()); + Assertions.assertThrows( + IllegalArgumentException.class, () -> new MailchimpSource(config(opts))); + } + + @Test + void httpRequiresExplicitFixtureOptIn() { + Map<String, Object> opts = options(); + opts.put("base_url", "http://localhost:8080/3.0";); + Assertions.assertThrows( + IllegalArgumentException.class, () -> new MailchimpConfig(config(opts))); + opts.put("allow_insecure_http", true); + Assertions.assertEquals("http", new MailchimpConfig(config(opts)).membersUri.getScheme()); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/test/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSourceFactoryTest.java b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/test/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSourceFactoryTest.java new file mode 100644 index 0000000000..3aadfe1ddb --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/test/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSourceFactoryTest.java @@ -0,0 +1,37 @@ +/* + * 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.mailchimp; + +import org.apache.seatunnel.api.table.factory.FactoryUtil; +import org.apache.seatunnel.api.table.factory.TableSourceFactory; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class MailchimpSourceFactoryTest { + @Test + void discoversMailchimpByIdentifier() { + ClassLoader loader = getClass().getClassLoader(); + Assertions.assertNotNull( + FactoryUtil.discoverFactory(loader, TableSourceFactory.class, "Http")); + Assertions.assertEquals( + "Mailchimp", + FactoryUtil.discoverFactory(loader, TableSourceFactory.class, "Mailchimp") + .factoryIdentifier()); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/test/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSourceReaderTest.java b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/test/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSourceReaderTest.java new file mode 100644 index 0000000000..0c842d0f2e --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-mailchimp/src/test/java/org/apache/seatunnel/connectors/seatunnel/mailchimp/MailchimpSourceReaderTest.java @@ -0,0 +1,543 @@ +/* + * 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.mailchimp; + +import org.apache.seatunnel.api.source.Collector; +import org.apache.seatunnel.api.source.SourceReader; +import org.apache.seatunnel.api.table.catalog.CatalogTableUtil; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplit; +import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext; + +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.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import com.sun.net.httpserver.HttpServer; +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import okhttp3.mockwebserver.RecordedRequest; +import okhttp3.mockwebserver.SocketPolicy; + +import java.io.IOException; +import java.math.BigDecimal; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Base64; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class MailchimpSourceReaderTest { + private MockWebServer server; + private MailchimpSourceReader reader; + private SourceReader.Context context; + private Collector<SeaTunnelRow> output; + private final List<SeaTunnelRow> rows = new ArrayList<>(); + + static String member(String id) { + return "{\"id\":\"" + + id + + "\",\"email_address\":\"" + + id + + "@example.invalid\",\"status\":\"subscribed\",\"last_changed\":\"2026-01-01T00:00:00Z\"," + + "\"merge_fields\":{\"FNAME\":\"Example\"}}"; + } + + static String page(int total, String... members) { + return "{\"total_items\":" + total + ",\"members\":[" + String.join(",", members) + "]}"; + } + + @BeforeEach + @SuppressWarnings("unchecked") + void setUp() throws Exception { + server = new MockWebServer(); + server.start(); + context = mock(SourceReader.Context.class); + output = mock(Collector.class); + when(output.getCheckpointLock()).thenReturn(new Object()); + doAnswer( + call -> { + rows.add(call.getArgument(0)); + return null; + }) + .when(output) + .collect(any(SeaTunnelRow.class)); + reader = newReader(2, 1); + reader.open(); + } + + private MailchimpSourceReader newReader(int retry, int backoff) { + return newReader(retry, backoff, "map<string,string>"); + } + + @SuppressWarnings("unchecked") + private MailchimpSourceReader newReader(int retry, int backoff, String mergeType) { + Map<String, Object> opts = MailchimpConfigTest.options(); + Map<String, Object> schema = new LinkedHashMap<>((Map<String, Object>) opts.get("schema")); + Map<String, Object> fields = + new LinkedHashMap<>((Map<String, Object>) schema.get("fields")); + fields.put("merge_fields", mergeType); + schema.put("fields", fields); + opts.put("schema", schema); + opts.put("base_url", server.url("/3.0").toString()); + opts.put("allow_insecure_http", true); + opts.put("page_size", 2); + opts.put("retry", retry); + opts.put("retry_backoff_ms", backoff); + opts.put("request_timeout_ms", 30000); + return new MailchimpSourceReader( + new MailchimpConfig(MailchimpConfigTest.config(opts)), + CatalogTableUtil.buildWithConfig(MailchimpConfigTest.config(opts)), + new SingleSplitReaderContext(context)); + } + + @AfterEach + void tearDown() throws Exception { + reader.close(); + server.shutdown(); + } + + @Test + void readsMultiplePagesAndSchemaWithoutExtraRequest() throws Exception { + server.enqueue(new MockResponse().setBody(page(3, member("one"), member("two")))); + server.enqueue(new MockResponse().setBody(page(3, member("three")))); + reader.pollNext(output); + reader.pollNext(output); + Assertions.assertEquals(3, rows.size()); + Assertions.assertEquals("one", rows.get(0).getField(0)); + Assertions.assertEquals("[email protected]", rows.get(2).getField(1)); + Assertions.assertEquals( + Collections.singletonMap("FNAME", "Example"), rows.get(0).getField(4)); + Assertions.assertEquals("default.default.default", rows.get(0).getTableId()); + Assertions.assertEquals(2, server.getRequestCount()); + RecordedRequest first = server.takeRequest(); + Assertions.assertEquals("GET", first.getMethod()); + Assertions.assertEquals("/3.0/lists/audience1/members?count=2&offset=0", first.getPath()); + Assertions.assertEquals( + "Basic " + + Base64.getEncoder() + .encodeToString( + "seatunnel:testkey-us1".getBytes(StandardCharsets.UTF_8)), + first.getHeader("Authorization")); + Assertions.assertEquals( + "/3.0/lists/audience1/members?count=1&offset=2", server.takeRequest().getPath()); + verify(context).signalNoMoreElement(); + } + + @Test + void emptyAudienceCompletes() throws Exception { + server.enqueue(new MockResponse().setBody(page(0))); + reader.pollNext(output); + Assertions.assertTrue(rows.isEmpty()); + verify(context).signalNoMoreElement(); + } + + @ParameterizedTest + @ValueSource(strings = {"map<string,decimal(24,18)>", "{VALUE=\"decimal(24,18)\"}"}) + void preservesNestedDecimalPrecision(String mergeType) throws Exception { + reader.close(); + reader = newReader(0, 1, mergeType); + reader.open(); + server.enqueue( + new MockResponse() + .setBody( + page( + 1, + "{\"id\":\"one\",\"merge_fields\":{\"VALUE\":0.123456789012345678}}"))); + reader.pollNext(output); + Object merge = rows.get(0).getField(4); + Object value = + merge instanceof Map + ? ((Map<?, ?>) merge).get("VALUE") + : ((SeaTunnelRow) merge).getField(0); + Assertions.assertEquals(new BigDecimal("0.123456789012345678"), value); + verify(context).signalNoMoreElement(); + } + + @Test + void exactFullPageCompletesAndMissingOptionalFieldsAreNull() throws Exception { + server.enqueue(new MockResponse().setBody(page(2, "{\"id\":\"one\"}", "{\"id\":\"two\"}"))); + reader.pollNext(output); + Assertions.assertEquals(2, rows.size()); + Assertions.assertNull(rows.get(0).getField(1)); + Assertions.assertEquals(1, server.getRequestCount()); + verify(context).signalNoMoreElement(); + } + + @Test + void growthCannotExtendInitialBound() throws Exception { + server.enqueue(new MockResponse().setBody(page(3, member("one"), member("two")))); + server.enqueue(new MockResponse().setBody(page(100, member("three")))); + reader.pollNext(output); + Assertions.assertEquals(3, rows.size()); + Assertions.assertEquals(2, server.getRequestCount()); + } + + @ParameterizedTest + @ValueSource( + strings = { + "{}", + "[]", + "null", + "{\"members\":{},\"total_items\":0}", + "{\"members\":[],\"total_items\":\"0\"}", + "{\"members\":[],\"total_items\":-1}", + "{\"members\":[],\"total_items\":2147483648}", + "{\"members\":[],\"total_items\":1}", + "{\"members\":[{}],\"total_items\":2}", + "{\"members\":[null],\"total_items\":1}", + "{\"members\":[1],\"total_items\":1}", + "{\"members\":[],\"total_items\":0} {}", + "{\"members\":[{\"merge_fields\":42}],\"total_items\":1}", + "[email protected]" + }) + void rejectsMalformedOrTruncatedPages(String body) { + server.enqueue(new MockResponse().setBody(body)); + IOException error = + Assertions.assertThrows(IOException.class, () -> reader.pollNext(output)); + Assertions.assertNull(error.getCause()); + Assertions.assertFalse(error.getMessage().contains("PII")); + Assertions.assertTrue(rows.isEmpty()); + verify(context, never()).signalNoMoreElement(); + } + + @Test + void retries429And5xxAtSameOffset() throws Exception { + server.enqueue( + new MockResponse() + .setResponseCode(429) + .setHeader("Retry-After", "0") + .setBody("private")); + server.enqueue( + new MockResponse() + .setResponseCode(503) + .setHeader("Retry-After", "invalid") + .setBody("private")); + server.enqueue(new MockResponse().setBody(page(1, member("one")))); + reader.pollNext(output); + Assertions.assertEquals(1, rows.size()); + for (int i = 0; i < 3; i++) { + Assertions.assertEquals( + "/3.0/lists/audience1/members?count=2&offset=0", + server.takeRequest().getPath()); + } + } + + @Test + void retriesBodylessErrorsAndReusesSuccessfulPageConnection() throws Exception { + HttpServer fixture = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + AtomicInteger requests = new AtomicInteger(); + List<String> paths = Collections.synchronizedList(new ArrayList<>()); + List<Integer> ports = Collections.synchronizedList(new ArrayList<>()); + fixture.createContext( + "/3.0/lists/audience1/members", + exchange -> { + int request = requests.incrementAndGet(); + paths.add(exchange.getRequestURI().toString()); + ports.add(exchange.getRemoteAddress().getPort()); + try { + if (request <= 2) { + exchange.getResponseHeaders().set("Retry-After", "0"); + exchange.sendResponseHeaders(request == 1 ? 429 : 503, -1); + } else { + byte[] body = + (request == 3 + ? page(3, member("one"), member("two")) + : page(3, member("three"))) + .getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, body.length); + exchange.getResponseBody().write(body); + } + } finally { + exchange.close(); + } + }); + fixture.start(); + try { + reader.close(); + Map<String, Object> opts = MailchimpConfigTest.options(); + opts.put("base_url", "http://127.0.0.1:"; + fixture.getAddress().getPort() + "/3.0"); + opts.put("allow_insecure_http", true); + opts.put("page_size", 2); + opts.put("retry", 2); + opts.put("retry_backoff_ms", 1); + opts.put("request_timeout_ms", 3000); + reader = + new MailchimpSourceReader( + new MailchimpConfig(MailchimpConfigTest.config(opts)), + CatalogTableUtil.buildWithConfig(MailchimpConfigTest.config(opts)), + new SingleSplitReaderContext(context)); + reader.open(); + Assertions.assertDoesNotThrow( + () -> reader.pollNext(output), + () -> "Bodyless retry sequence failed after " + requests.get() + " requests"); + Assertions.assertEquals(3, rows.size()); + Assertions.assertEquals(4, requests.get()); + for (int i = 0; i < 3; i++) { + Assertions.assertEquals( + "/3.0/lists/audience1/members?count=2&offset=0", paths.get(i)); + } + Assertions.assertEquals("/3.0/lists/audience1/members?count=1&offset=2", paths.get(3)); + Assertions.assertEquals( + ports.get(2), ports.get(3), "Successful pages reuse a connection"); + verify(context).signalNoMoreElement(); + } finally { + fixture.stop(0); + } + } + + @ParameterizedTest + @ValueSource(ints = {400, 401, 403, 404, 204, 301, 302, 307, 308}) + void permanentErrorsAndRedirectsNeverRetryOrExposeBodies(int status) throws Exception { + server.enqueue( + new MockResponse() + .setResponseCode(status) + .setHeader("Location", server.url("/stolen")) + .setBody("testkey-us1 [email protected]")); + IOException error = + Assertions.assertThrows(IOException.class, () -> reader.pollNext(output)); + Assertions.assertTrue(error.getMessage().contains("HTTP " + status)); + Assertions.assertNull(error.getCause()); + Assertions.assertFalse(error.getMessage().contains("testkey")); + Assertions.assertFalse(error.getMessage().contains("private")); + Assertions.assertEquals(1, server.getRequestCount()); + verify(context, never()).signalNoMoreElement(); + } + + @ParameterizedTest + @ValueSource(ints = {429, 500, 503}) + void retryBudgetIsFinite(int status) { + for (int i = 0; i < 3; i++) { + server.enqueue(new MockResponse().setResponseCode(status).setBody("private")); + } + Assertions.assertThrows(IOException.class, () -> reader.pollNext(output)); + Assertions.assertEquals(3, server.getRequestCount()); + verify(context, never()).signalNoMoreElement(); + } + + @Test + void redirectCannotForwardCredentialsToAnotherOrigin() throws Exception { + try (MockWebServer destination = new MockWebServer()) { + destination.start(); + server.enqueue( + new MockResponse() + .setResponseCode(302) + .setHeader("Location", destination.url("/credential-target"))); + Assertions.assertThrows(IOException.class, () -> reader.pollNext(output)); + Assertions.assertNull(destination.takeRequest(100, TimeUnit.MILLISECONDS)); + Assertions.assertEquals(1, server.getRequestCount()); + verify(context, never()).signalNoMoreElement(); + } + } + + @Test + void httpDateRetryAfterIsAccepted() throws Exception { + server.enqueue( + new MockResponse() + .setResponseCode(429) + .setHeader("Retry-After", "Thu, 01 Jan 1970 00:00:00 GMT")); + server.enqueue(new MockResponse().setBody(page(0))); + reader.pollNext(output); + Assertions.assertEquals(2, server.getRequestCount()); + verify(context).signalNoMoreElement(); + } + + @Test + void brokenTransportDoesNotComplete() { + server.enqueue( + new MockResponse() + .setBody(page(1, member("one"))) + .setSocketPolicy(SocketPolicy.DISCONNECT_DURING_RESPONSE_BODY)); + Assertions.assertThrows(IOException.class, () -> reader.pollNext(output)); + Assertions.assertTrue(rows.isEmpty()); + verify(context, never()).signalNoMoreElement(); + } + + @Test + void oversizedUnfinishedBodyFailsPromptlyWithoutCancellation() throws Exception { + server.enqueue( + new MockResponse() + .setBody("x".repeat(32 * 1024 * 1024 + 16384)) + .setHeader("Content-Length", 64 * 1024 * 1024)); + CompletableFuture<Exception> task = pollInBackground(); + Assertions.assertInstanceOf(IOException.class, task.get(5, TimeUnit.SECONDS)); + Assertions.assertTrue(rows.isEmpty()); + verify(context, never()).signalNoMoreElement(); + } + + @Test + void failureAfterAcceptedPageReplaysWholeBatchOnRestore() throws Exception { + server.enqueue(new MockResponse().setBody(page(3, member("one"), member("two")))); + server.enqueue(new MockResponse().setResponseCode(401).setBody("private")); + Assertions.assertThrows(IOException.class, () -> reader.pollNext(output)); + Assertions.assertEquals(2, rows.size()); + List<SingleSplit> state = reader.snapshotState(1); + Assertions.assertNull( + state.get(0).getState(), "No offset, credentials or PII in split state"); + verify(context, never()).signalNoMoreElement(); + reader.close(); + reader = newReader(0, 1); + reader.addSplits(state); + reader.open(); + server.enqueue(new MockResponse().setBody(page(3, member("one"), member("two")))); + server.enqueue(new MockResponse().setBody(page(3, member("three")))); + reader.pollNext(output); + Assertions.assertEquals(5, rows.size()); + server.takeRequest(); + server.takeRequest(); + Assertions.assertTrue(server.takeRequest().getPath().endsWith("offset=0")); + verify(context).signalNoMoreElement(); + } + + @Test + void audienceShrinkIsFailureNotSuccess() { + server.enqueue(new MockResponse().setBody(page(3, member("one"), member("two")))); + server.enqueue(new MockResponse().setBody(page(2))); + Assertions.assertThrows(IOException.class, () -> reader.pollNext(output)); + Assertions.assertEquals(2, rows.size()); + verify(context, never()).signalNoMoreElement(); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void closeOrCancelWakesRetryAndIsIdempotent(boolean cancel) throws Exception { + reader.close(); + reader = newReader(10, 60000); + reader.open(); + server.enqueue( + new MockResponse().setResponseCode(429).setHeader("Retry-After", "9999999999999")); + CompletableFuture<Exception> task = pollInBackground(); + Assertions.assertNotNull(server.takeRequest(2, TimeUnit.SECONDS)); + if (cancel) { + reader.cancel(); + } else { + reader.close(); + } + Assertions.assertInstanceOf(IOException.class, task.get(2, TimeUnit.SECONDS)); + reader.close(); + Assertions.assertThrows(IOException.class, () -> reader.open()); + Assertions.assertEquals(1, server.getRequestCount()); + verify(context, never()).signalNoMoreElement(); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void closeOrCancelAbortsActiveIo(boolean cancel) throws Exception { + server.enqueue(new MockResponse().setSocketPolicy(SocketPolicy.NO_RESPONSE)); + CompletableFuture<Exception> task = pollInBackground(); + Assertions.assertNotNull(server.takeRequest(2, TimeUnit.SECONDS)); + Assertions.assertTrue( + CompletableFuture.supplyAsync( + () -> { + synchronized (output.getCheckpointLock()) { + return true; + } + }) + .get(1, TimeUnit.SECONDS), + "Network I/O must not hold the checkpoint lock"); + if (cancel) { + reader.cancel(); + } else { + reader.close(); + } + Assertions.assertInstanceOf(IOException.class, task.get(2, TimeUnit.SECONDS)); + verify(context, never()).signalNoMoreElement(); + } + + @Test + void cancellationBeforeOpenIsSticky() throws Exception { + reader.close(); + reader = newReader(0, 1); + reader.cancel(); + reader.cancel(); + Assertions.assertThrows(IOException.class, () -> reader.open()); + reader.pollNext(output); + Assertions.assertEquals(0, server.getRequestCount()); + verify(context, never()).signalNoMoreElement(); + } + + @Test + void cancelDoesNotWaitForCompletionAcknowledgement() throws Exception { + server.enqueue(new MockResponse().setBody(page(0))); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + doAnswer( + call -> { + entered.countDown(); + release.await(); + return null; + }) + .when(context) + .signalNoMoreElement(); + CompletableFuture<Exception> poll = pollInBackground(); + Assertions.assertTrue(entered.await(2, TimeUnit.SECONDS)); + CompletableFuture<Void> cancel = CompletableFuture.runAsync(reader::cancel); + try { + cancel.get(1, TimeUnit.SECONDS); + } finally { + release.countDown(); + poll.get(2, TimeUnit.SECONDS); + cancel.get(2, TimeUnit.SECONDS); + } + } + + @Test + void closeDuringCollectionDoesNotSignalCompletion() throws Exception { + server.enqueue(new MockResponse().setBody(page(2, member("one"), member("two")))); + doAnswer( + call -> { + rows.add(call.getArgument(0)); + reader.close(); + return null; + }) + .when(output) + .collect(any(SeaTunnelRow.class)); + reader.pollNext(output); + Assertions.assertEquals(1, rows.size()); + verify(context, never()).signalNoMoreElement(); + } + + private CompletableFuture<Exception> pollInBackground() { + return CompletableFuture.supplyAsync( + () -> { + try { + reader.pollNext(output); + return null; + } catch (Exception e) { + return e; + } + }); + } +} diff --git a/seatunnel-connectors-v2/connector-http/pom.xml b/seatunnel-connectors-v2/connector-http/pom.xml index 264bd42087..87d301a9fc 100644 --- a/seatunnel-connectors-v2/connector-http/pom.xml +++ b/seatunnel-connectors-v2/connector-http/pom.xml @@ -53,6 +53,7 @@ <module>connector-http-asana</module> <module>connector-http-intercom</module> <module>connector-http-typeform</module> + <module>connector-http-mailchimp</module> <module>connector-mem0</module> </modules> diff --git a/seatunnel-dist/pom.xml b/seatunnel-dist/pom.xml index f93cd8346e..9697e98641 100644 --- a/seatunnel-dist/pom.xml +++ b/seatunnel-dist/pom.xml @@ -275,6 +275,12 @@ <version>${project.version}</version> <scope>provided</scope> </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-mailchimp</artifactId> + <version>${project.version}</version> + <scope>provided</scope> + </dependency> <dependency> <groupId>org.apache.seatunnel</groupId> <artifactId>connector-kafka</artifactId> diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/pom.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/pom.xml new file mode 100644 index 0000000000..e461dd8d69 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/pom.xml @@ -0,0 +1,53 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + 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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.seatunnel</groupId> + <artifactId>seatunnel-connector-v2-e2e</artifactId> + <version>${revision}</version> + </parent> + <artifactId>connector-mailchimp-e2e</artifactId> + <name>SeaTunnel : E2E : Connector V2 : Mailchimp</name> + + <dependencies> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-mailchimp</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-base</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-assert</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-common</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + </dependencies> +</project> diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/mailchimp/MailchimpIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/mailchimp/MailchimpIT.java new file mode 100644 index 0000000000..2bdd79103e --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/mailchimp/MailchimpIT.java @@ -0,0 +1,148 @@ +/* + * 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.e2e.connector.mailchimp; + +import org.apache.seatunnel.e2e.common.TestResource; +import org.apache.seatunnel.e2e.common.TestSuiteBase; +import org.apache.seatunnel.e2e.common.container.EngineType; +import org.apache.seatunnel.e2e.common.container.TestContainer; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.TestTemplate; +import org.testcontainers.Testcontainers; +import org.testcontainers.containers.Container; + +import com.sun.net.httpserver.HttpServer; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import java.util.Collections; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.regex.Pattern; + +class MailchimpIT extends TestSuiteBase implements TestResource { + private HttpServer server; + private final AtomicInteger requests = new AtomicInteger(); + private final AtomicReference<String> invalidRequest = new AtomicReference<>(); + + @BeforeAll + @Override + public void startUp() throws IOException { + server = HttpServer.create(new InetSocketAddress(0), 0); + Testcontainers.exposeHostPorts(server.getAddress().getPort()); + server.createContext( + "/3.0/lists/audience1/members", + exchange -> { + int request = requests.incrementAndGet(); + String auth = + "Basic " + + Base64.getEncoder() + .encodeToString( + "seatunnel:fixturekey-us1" + .getBytes(StandardCharsets.UTF_8)); + String expected = request <= 3 ? "count=2&offset=0" : "count=1&offset=2"; + int code = 200; + String body; + if (!"GET".equals(exchange.getRequestMethod()) + || !auth.equals(exchange.getRequestHeaders().getFirst("Authorization")) + || !expected.equals(exchange.getRequestURI().getRawQuery()) + || !"/3.0/lists/audience1/members" + .equals(exchange.getRequestURI().getPath())) { + invalidRequest.set( + "Unexpected request method, authentication or pagination"); + code = 400; + body = "{}"; + } else if (request == 1 || request == 2) { + code = request == 1 ? 429 : 503; + exchange.getResponseHeaders().set("Retry-After", "0"); + body = "{}"; + } else { + String resource = + request == 3 + ? "/mailchimp-members-page-1.json" + : "/mailchimp-members-page-2.json"; + try (InputStream input = getClass().getResourceAsStream(resource)) { + if (input == null) { + throw new IOException("Missing Mailchimp fixture"); + } + body = new String(input.readAllBytes(), StandardCharsets.UTF_8); + } + } + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "application/json"); + if (code == 429 || code == 503) { + exchange.sendResponseHeaders(code, -1); + exchange.close(); + return; + } + exchange.sendResponseHeaders(code, bytes.length); + try (OutputStream out = exchange.getResponseBody()) { + out.write(bytes); + } + }); + server.start(); + } + + @BeforeEach + void reset() { + requests.set(0); + invalidRequest.set(null); + } + + @AfterAll + @Override + public void tearDown() { + if (server != null) { + server.stop(0); + } + } + + @TestTemplate + void listMembersToAssert(TestContainer container) throws Exception { + Container.ExecResult result = + container.executeJob( + "/mailchimp_source_to_assert.conf", + Collections.singletonList( + "mailchimp.test.port=" + server.getAddress().getPort())); + Assertions.assertEquals(0, result.getExitCode(), result.getStderr()); + if (container.identifier().getEngineType() == EngineType.SEATUNNEL) { + String clientLogs = result.getStdout() + "\n" + result.getStderr(); + Assertions.assertTrue( + Pattern.compile("Job \\([0-9]+\\) end with state FINISHED") + .matcher(clientLogs) + .find(), + clientLogs); + Assertions.assertFalse( + Pattern.compile("Job \\([0-9]+\\) end with state FAILED|jobStatus=FAILED") + .matcher(clientLogs) + .find(), + clientLogs); + } + Assertions.assertNull(invalidRequest.get()); + Assertions.assertEquals( + 4, requests.get(), "Retries stay at offset zero, followed by the final page"); + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/src/test/resources/mailchimp-members-page-1.json b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/src/test/resources/mailchimp-members-page-1.json new file mode 100644 index 0000000000..20287920ea --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/src/test/resources/mailchimp-members-page-1.json @@ -0,0 +1,23 @@ +{ + "members": [ + { + "id": "member1", + "email_address": "[email protected]", + "status": "subscribed", + "last_changed": "2026-01-01T00:00:00Z", + "merge_fields": { + "FNAME": "Example" + } + }, + { + "id": "member2", + "email_address": "[email protected]", + "status": "subscribed", + "last_changed": "2026-01-01T00:00:00Z", + "merge_fields": { + "FNAME": "Example" + } + } + ], + "total_items": 3 +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/src/test/resources/mailchimp-members-page-2.json b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/src/test/resources/mailchimp-members-page-2.json new file mode 100644 index 0000000000..536999c960 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/src/test/resources/mailchimp-members-page-2.json @@ -0,0 +1,14 @@ +{ + "members": [ + { + "id": "member3", + "email_address": "[email protected]", + "status": "subscribed", + "last_changed": "2026-01-01T00:00:00Z", + "merge_fields": { + "FNAME": "Example" + } + } + ], + "total_items": 3 +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/src/test/resources/mailchimp_source_to_assert.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/src/test/resources/mailchimp_source_to_assert.conf new file mode 100644 index 0000000000..f448360eb1 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mailchimp-e2e/src/test/resources/mailchimp_source_to_assert.conf @@ -0,0 +1,59 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "BATCH" + job.retry.times = 0 +} + +source { + Mailchimp { + base_url = "http://host.testcontainers.internal:${mailchimp.test.port}/3.0"; + allow_insecure_http = true + token = "fixturekey-us1" + list_id = "audience1" + page_size = 2 + retry_backoff_ms = 10 + schema { + fields { + id = string + email_address = string + status = string + last_changed = string + merge_fields = "map<string,string>" + } + } + } +} + +sink { + Assert { + rules { + row_rules = [ + { rule_type = "MAX_ROW", rule_value = 3 }, + { rule_type = "MIN_ROW", rule_value = 3 } + ] + field_rules = [ + { field_name = "id", field_type = "string", field_value = [{ rule_type = "NOT_NULL" }] }, + { field_name = "email_address", field_type = "string", field_value = [{ rule_type = "NOT_NULL" }] }, + { field_name = "status", field_type = "string", field_value = [{ rule_type = "NOT_NULL" }] }, + { field_name = "last_changed", field_type = "string", field_value = [{ rule_type = "NOT_NULL" }] } + ] + } + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml index 7368584279..ec1e0adba3 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml @@ -54,6 +54,7 @@ <module>connector-asana-e2e</module> <module>connector-intercom-e2e</module> <module>connector-typeform-e2e</module> + <module>connector-mailchimp-e2e</module> <module>connector-http-e2e</module> <module>connector-paypal-e2e</module> <module>connector-tiktok-ads-e2e</module>
