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-12282-a51712781b17b9982b8f288304508a31e3e27feb in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit e86c03e53757b63869cf5a60de1e21cf36806ace Author: Goutam Adwant <[email protected]> AuthorDate: Tue Sep 15 12:55:15 2026 +0000 [Feature][Connector-V2] Add PayPal Transaction Search source (#12282) Signed-off-by: Goutam Adwant <[email protected]> --- .github/workflows/labeler/label-scope-conf.yml | 4 +- config/plugin_config | 1 + .../connectors/changelog/connector-http-paypal.md | 5 + docs/en/connectors/source/PayPal.md | 107 +++++ .../connectors/changelog/connector-http-paypal.md | 5 + docs/zh/connectors/source/PayPal.md | 107 +++++ plugin-mapping.properties | 1 + .../{ => connector-http-paypal}/pom.xml | 36 +- .../seatunnel/paypal/source/PayPalClient.java | 296 +++++++++++++ .../seatunnel/paypal/source/PayPalConfig.java | 155 +++++++ .../seatunnel/paypal/source/PayPalResponse.java | 168 +++++++ .../seatunnel/paypal/source/PayPalSource.java | 100 +++++ .../paypal/source/PayPalSourceFactory.java | 70 +++ .../paypal/source/PayPalSourceOptions.java | 85 ++++ .../paypal/source/PayPalSourceReader.java | 95 ++++ .../seatunnel/paypal/source/PayPalClientTest.java | 487 +++++++++++++++++++++ .../paypal/source/PayPalResponseTest.java | 402 +++++++++++++++++ .../paypal/source/TestPayPalPackagingIT.java | 141 ++++++ seatunnel-connectors-v2/connector-http/pom.xml | 1 + seatunnel-dist/pom.xml | 6 + .../connector-paypal-e2e/pom.xml | 54 +++ .../seatunnel/e2e/connector/paypal/PayPalIT.java | 155 +++++++ .../e2e/connector/paypal/PayPalMaskingTest.java | 63 +++ .../src/test/resources/paypal_to_assert.conf | 52 +++ seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml | 1 + 25 files changed, 2571 insertions(+), 26 deletions(-) diff --git a/.github/workflows/labeler/label-scope-conf.yml b/.github/workflows/labeler/label-scope-conf.yml index f4fe6c9946..8ccad991d9 100644 --- a/.github/workflows/labeler/label-scope-conf.yml +++ b/.github/workflows/labeler/label-scope-conf.yml @@ -239,7 +239,9 @@ hive: http: - all: - changed-files: - - any-glob-to-any-file: seatunnel-connectors-v2/connector-http/** + - any-glob-to-any-file: + - seatunnel-connectors-v2/connector-http/** + - seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/** - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(http)/**' prometheus: - all: diff --git a/config/plugin_config b/config/plugin_config index d4efe4cdee..a278e647f0 100644 --- a/config/plugin_config +++ b/config/plugin_config @@ -74,6 +74,7 @@ connector-http-github connector-http-jira connector-http-klaviyo connector-http-stripe +connector-http-paypal connector-http-lemlist connector-http-linear connector-http-myhours diff --git a/docs/en/connectors/changelog/connector-http-paypal.md b/docs/en/connectors/changelog/connector-http-paypal.md new file mode 100644 index 0000000000..6dd26bec05 --- /dev/null +++ b/docs/en/connectors/changelog/connector-http-paypal.md @@ -0,0 +1,5 @@ +# Changelog + +## Next version + +- Add bounded PayPal Transaction Search source with OAuth authentication. diff --git a/docs/en/connectors/source/PayPal.md b/docs/en/connectors/source/PayPal.md new file mode 100644 index 0000000000..71655b07a2 --- /dev/null +++ b/docs/en/connectors/source/PayPal.md @@ -0,0 +1,107 @@ +import ChangeLog from '../changelog/connector-http-paypal.md'; + +# PayPal + +> PayPal Transaction Search source connector. + +## Description + +Reads one account's transactions through `GET /v1/reporting/transactions` using first-party OAuth client credentials. This is a bounded BATCH report, not CDC, a payment API, or a balance reconciliation service. + +## Key Features + +- [x] Batch +- [ ] Streaming +- [ ] Exactly-once + +Parallelism must be 1. Recovery replays the entire configured window; no page offset is checkpointed. A failed job may already have emitted rows to a nontransactional sink. + +## Prerequisites + +Use a PayPal REST application authorized for Transaction Search for its own account. Enable Transaction Search permissions for the app and verify the issued credentials have reporting access; successful token issuance alone does not prove this. Permission changes may require a fresh token. Third-party account access and partner authorization are out of scope. + +Production origin is `https://api-m.paypal.com`; sandbox origin is `https://api-m.sandbox.paypal.com`. Sandbox credentials and reporting availability are separate from production. A mock test does not prove sandbox or production permissions, data availability, or completeness. This connector has no live-account verification claim. + +See [the official Transaction Search schema](https://github.com/paypal/paypal-rest-api-specifications/blob/main/openapi/reporting_transactions_v1.json), [List transactions](https://developer.paypal.com/api/transaction-search/v1/search-get), and [REST authentication](https://developer.paypal.com/api/rest/authentication/). + +## Reporting Contract + +Supply explicit absolute RFC3339 `start_date` and `end_date` with seconds and an offset, with start before end and at most 31 days between them. Dates are normalized to UTC, never the host time zone. PayPal provides up to three years of history and transactions can take up to three hours to appear. + +The connector always sends `fields=all` and `balance_affecting_records_only=N`, retaining both balance-affecting and non-affecting reporting records. Transaction IDs are NOT unique; no primary key or ID deduplication is applied. Preserve this distinction in downstream storage. + +Every response must include account, dates, page, totals, and a transaction array. Returned dates must cover exactly the requested interval after UTC normalization. In particular, PayPal's returned `end_date` may instead be the last date available; a shortened response fails. Wait for reporting availability or explicitly narrow the request, then rerun. The connector does not silently move the requested end. + +Malformed rows, API error envelopes (even with HTTP 200), inconsistent page sizes, changed account/totals, and detected truncation fail the job. `RESULTSET_TOO_LARGE` fails with advice to narrow the date window. The connector safety cap is 10,000 records; exactly 10,000 is accepted when every page and count validates. Larger results fail. The connector never truncates or automatically splits the interval. Reconcile boundaries and replay effects explicitly when changing windows. + +Stable totals and matching coverage detect only some problems. PayPal does not provide a remote immutable snapshot here; same-count changes, delayed updates, or omissions may remain undetectable. No snapshot-consistency, lossless interval-splitting, or complete-history guarantee is made. + +## Output Schema + +The schema is fixed; no user-defined `schema` option is supported. + +| Field | Type | Nullable | Meaning | +| --- | --- | --- | --- | +| account_number | STRING | no | Reporting account from the response | +| transaction_id | STRING | yes | Non-unique reporting transaction ID | +| transaction_event_code | STRING | yes | PayPal event code | +| transaction_status | STRING | yes | PayPal status | +| transaction_initiation_date | STRING | yes | RFC3339 timestamp normalized to UTC | +| transaction_updated_date | STRING | yes | RFC3339 timestamp normalized to UTC | +| transaction_amount | DECIMAL(38,9) | yes | Gross amount in native currency units | +| transaction_currency | STRING | yes | Gross amount currency | +| fee_amount | DECIMAL(38,9) | yes | Fee in native currency units | +| fee_currency | STRING | yes | Fee currency, independent of gross currency | +| content | STRING | no | Entire transaction detail object as JSON, including references and additional fields | + +Optional fields may be absent or null. Present money objects must contain a string amount and three-letter currency. Zero- and three-decimal currencies are supported. Amounts must fit DECIMAL(38,9) exactly; excess scale or precision fails rather than rounding. Additional monetary fields remain in `content` in their original JSON representation. No floating-point conversion is used. Raw records can contain personal or financial data: use access-controlled sinks and do not log them. + +## Options + +| Name | Type | Required | Default | Description | +| --- | --- | --- | --- | --- | +| client_id | STRING | yes | - | First-party REST app client ID | +| client_secret | STRING | yes | - | Client secret; native parsed-config log masking applies | +| start_date | STRING | yes | - | Absolute start with seconds and offset | +| end_date | STRING | yes | - | Absolute end, no more than 31 days after start | +| api_base_url | STRING | no | https://api-m.paypal.com | Exact production or sandbox origin, no trailing slash | +| page_size | INT | no | 100 | 1 to 500 | +| max_retries | INT | no | 3 | Additional transient retries, 0 to 5 | +| retry_delay_ms | INT | no | 1000 | 1 to 60000 | +| request_timeout_ms | INT | no | 30000 | Connect/read timeout and request abort deadline, 1 to 120000 | +| max_response_bytes | INT | no | 8388608 | Decompressed bytes per response, 1024 to 16777216 | +| mock_mode | BOOLEAN | no | false | Custom origin for tests; requires exactly mock-client / mock-secret | + +OAuth tokens are acquired via the real token endpoint and refreshed on expiry or once after HTTP 401 per page. HTTP 429/500/502/503/504 and transport failures use bounded retries. Other errors fail. Numeric Retry-After up to 60 seconds is honored; longer or unsupported values fail so the job can be retried later. Redirects are not followed. Close aborts active HTTP requests and wakes retry waits. + +Request deadlines cannot interrupt JVM DNS resolution or a blocked downstream collector; they are not an unconditional wall-clock job deadline. The page count, response size, attempt count and wait duration are bounded. Keep HTTP wire/header logging disabled. Do not place secrets in endpoint URLs, dotted option names, or diagnostics. Use environment substitution or an approved secret provider. `mock_mode` is not an official PayPal emulator and must never use real credentials. + +## Example + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} +source { + PayPal { + plugin_output = "transactions" + client_id = ${PAYPAL_CLIENT_ID} + client_secret = ${PAYPAL_CLIENT_SECRET} + start_date = "2026-01-01T00:00:00Z" + end_date = "2026-01-02T00:00:00Z" + page_size = 100 + } +} +sink { + LocalFile { + plugin_input = "transactions" + path = "/data/paypal" + file_format_type = "json" + } +} +``` + +## Changelog + +<ChangeLog /> diff --git a/docs/zh/connectors/changelog/connector-http-paypal.md b/docs/zh/connectors/changelog/connector-http-paypal.md new file mode 100644 index 0000000000..6dd26bec05 --- /dev/null +++ b/docs/zh/connectors/changelog/connector-http-paypal.md @@ -0,0 +1,5 @@ +# Changelog + +## Next version + +- Add bounded PayPal Transaction Search source with OAuth authentication. diff --git a/docs/zh/connectors/source/PayPal.md b/docs/zh/connectors/source/PayPal.md new file mode 100644 index 0000000000..c54ab83e37 --- /dev/null +++ b/docs/zh/connectors/source/PayPal.md @@ -0,0 +1,107 @@ +import ChangeLog from '../changelog/connector-http-paypal.md'; + +# PayPal + +> PayPal Transaction Search 数据源连接器。 + +## 描述 + +使用第一方 OAuth 客户端凭据，通过 `GET /v1/reporting/transactions` 读取一个账户的有界交易报表。仅支持 BATCH，不是 CDC、支付接口或余额对账服务。 + +## 主要功能 + +- [x] 批处理 +- [ ] 流处理 +- [ ] Exactly-once + +并行度必须为 1。恢复时重新读取整个配置窗口，不保存分页偏移量。任务失败前可能已向非事务型下游写入部分记录。 + +## 前提条件 + +REST 应用必须获得其所属账户的 Transaction Search 权限。请启用应用的交易查询权限并确认报表访问；获取 OAuth token 成功不代表具有报表权限。权限变更可能需要重新获取 token。不支持第三方账户或合作伙伴授权流程。 + +生产地址为 `https://api-m.paypal.com`，沙箱地址为 `https://api-m.sandbox.paypal.com`。沙箱凭据和报表可用性独立于生产环境。模拟测试不证明沙箱或生产权限、数据可用性或完整性。本连接器不声称已通过真实账户验证。 + +参考 [官方 OpenAPI](https://github.com/paypal/paypal-rest-api-specifications/blob/main/openapi/reporting_transactions_v1.json)、[交易查询](https://developer.paypal.com/api/transaction-search/v1/search-get) 和 [REST 认证](https://developer.paypal.com/api/rest/authentication/)。 + +## 报表约定 + +必须指定带秒和时区偏移的绝对 RFC3339 `start_date`、`end_date`。开始早于结束，跨度不超过 31 天。时间统一转换为 UTC，不使用主机时区。PayPal 提供最近三年的历史数据，交易最多可能延迟三小时才出现在报表中。 + +固定请求 `fields=all` 和 `balance_affecting_records_only=N`，保留影响余额及不影响余额的所有返回记录。交易 ID 在报表系统中不唯一，不建立主键、不按 ID 去重，下游也应保留此区别。 + +每页必须包含账户、时间范围、页码、总数及交易数组。返回时间范围经 UTC 归一化后必须与请求完全一致。PayPal 的 `end_date` 也可能表示当前可提供数据的最后时间：若返回范围缩短，任务失败，不会静默修改请求。请等待报表可用，或显式缩小窗口后重跑。 + +遇到格式错误、HTTP 200 中的业务错误、分页数量不一致、账户或总数变化及可检测的截断时失败。`RESULTSET_TOO_LARGE` 提示缩小日期窗口。连接器限制为最多 10,000 条；恰好 10,000 条且所有分页及总数校验通过时可读取，超过限制则失败。不截断、不自动拆分窗口。调整窗口时须明确核对边界和重放影响。 + +总数稳定、时间范围匹配只能检测部分问题，不代表远端不可变快照。同数量的记录变化、延迟更新或遗漏可能无法检测。不保证快照一致性、无损时间切分或完整历史。 + +## 输出结构 + +固定结构，不支持自定义 `schema`。 + +| 字段 | 类型 | 可空 | 含义 | +| --- | --- | --- | --- | +| account_number | STRING | 否 | 响应中的报表账户 | +| transaction_id | STRING | 是 | 非唯一交易 ID | +| transaction_event_code | STRING | 是 | 事件代码 | +| transaction_status | STRING | 是 | 状态 | +| transaction_initiation_date | STRING | 是 | UTC RFC3339 创建时间 | +| transaction_updated_date | STRING | 是 | UTC RFC3339 更新时间 | +| transaction_amount | DECIMAL(38,9) | 是 | 原币单位总金额 | +| transaction_currency | STRING | 是 | 总金额币种 | +| fee_amount | DECIMAL(38,9) | 是 | 原币单位手续费 | +| fee_currency | STRING | 是 | 手续费币种，与总金额币种独立 | +| content | STRING | 否 | 完整交易对象 JSON，包含引用及额外字段 | + +可选字段缺失或为 null 均保留为空。存在的金额对象必须含字符串金额和三字母币种；支持零位及三位小数币种。金额须能精确表示为 DECIMAL(38,9)，超出精度或小数位时失败而非舍入。其他金额字段保留在原始 JSON 中，不进行浮点转换。原始记录可能包含个人或财务数据，请使用受控下游并避免记录到日志。 + +## 配置项 + +| 名称 | 类型 | 必填 | 默认值 | 说明 | +| --- | --- | --- | --- | --- | +| client_id | STRING | 是 | - | 第一方 REST 应用 ID | +| client_secret | STRING | 是 | - | 客户端密钥，使用原生配置日志脱敏 | +| start_date | STRING | 是 | - | 带秒和偏移的绝对开始时间 | +| end_date | STRING | 是 | - | 绝对结束时间，窗口不超过 31 天 | +| api_base_url | STRING | 否 | https://api-m.paypal.com | 精确生产或沙箱源地址，不带末尾斜线 | +| page_size | INT | 否 | 100 | 1 到 500 | +| max_retries | INT | 否 | 3 | 额外瞬时故障重试，0 到 5 | +| retry_delay_ms | INT | 否 | 1000 | 1 到 60000 毫秒 | +| request_timeout_ms | INT | 否 | 30000 | 连接、读取及请求中止期限，1 到 120000 毫秒 | +| max_response_bytes | INT | 否 | 8388608 | 每次响应解压后字节上限，1024 到 16777216 | +| mock_mode | BOOLEAN | 否 | false | 测试自定义源地址，凭据必须严格为 mock-client / mock-secret | + +通过真实 token 接口获取 OAuth token，到期或每页首次 HTTP 401 后刷新。仅对 HTTP 429/500/502/503/504 及传输错误进行有界重试。遵守最多 60 秒的数字 Retry-After；更长或不支持的值直接失败，请稍后重跑。不跟随重定向。关闭时中止活动请求并唤醒重试等待。 + +请求期限不能中断 JVM DNS 解析或阻塞的下游 collect，因此不是无条件的任务总耗时保证。分页数、响应大小、请求次数和等待时间均受限。请关闭 HTTP wire/header 日志；密钥不得写入 URL、点分配置名或诊断信息。使用环境变量或认可的密钥提供方式。模拟模式不是官方 PayPal 模拟器，不得使用真实凭据。 + +## 示例 + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} +source { + PayPal { + plugin_output = "transactions" + client_id = ${PAYPAL_CLIENT_ID} + client_secret = ${PAYPAL_CLIENT_SECRET} + start_date = "2026-01-01T00:00:00Z" + end_date = "2026-01-02T00:00:00Z" + page_size = 100 + } +} +sink { + LocalFile { + plugin_input = "transactions" + path = "/data/paypal" + file_format_type = "json" + } +} +``` + +## 变更日志 + +<ChangeLog /> diff --git a/plugin-mapping.properties b/plugin-mapping.properties index f3bf789fbe..265a2c5398 100644 --- a/plugin-mapping.properties +++ b/plugin-mapping.properties @@ -118,6 +118,7 @@ seatunnel.source.Shopify = connector-http-shopify seatunnel.source.Zendesk = connector-http-zendesk seatunnel.sink.Zendesk = connector-http-zendesk seatunnel.source.Stripe = connector-http-stripe +seatunnel.source.PayPal = connector-http-paypal seatunnel.sink.Slack = connector-slack seatunnel.source.OneSignal = connector-http-onesignal seatunnel.source.Jira = connector-http-jira diff --git a/seatunnel-connectors-v2/connector-http/pom.xml b/seatunnel-connectors-v2/connector-http/connector-http-paypal/pom.xml similarity index 52% copy from seatunnel-connectors-v2/connector-http/pom.xml copy to seatunnel-connectors-v2/connector-http/connector-http-paypal/pom.xml index 05f73a000c..13877b6323 100644 --- a/seatunnel-connectors-v2/connector-http/pom.xml +++ b/seatunnel-connectors-v2/connector-http/connector-http-paypal/pom.xml @@ -22,33 +22,19 @@ <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.apache.seatunnel</groupId> - <artifactId>seatunnel-connectors-v2</artifactId> + <artifactId>connector-http</artifactId> <version>${revision}</version> </parent> - <artifactId>connector-http</artifactId> - <packaging>pom</packaging> - <name>SeaTunnel : Connectors V2 : Http :</name> - <modules> - <module>connector-http-base</module> - <module>connector-http-feishu</module> - <module>connector-http-wechat</module> - <module>connector-http-myhours</module> - <module>connector-http-lemlist</module> - <module>connector-http-klaviyo</module> - <module>connector-http-onesignal</module> - <module>connector-http-jira</module> - <module>connector-http-gitlab</module> - <module>connector-http-github</module> - <module>connector-http-notion</module> - <module>connector-http-persistiq</module> - <module>connector-http-airtable</module> - <module>connector-http-posthog</module> - <module>connector-http-shopify</module> - <module>connector-http-zendesk</module> - <module>connector-http-stripe</module> - <module>connector-http-linear</module> - <module>connector-mem0</module> - </modules> + <artifactId>connector-http-paypal</artifactId> + <name>SeaTunnel : Connectors V2 : Http : PayPal</name> + + <dependencies> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-base</artifactId> + <version>${project.version}</version> + </dependency> + </dependencies> </project> diff --git a/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalClient.java b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalClient.java new file mode 100644 index 0000000000..8553a29b92 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalClient.java @@ -0,0 +1,296 @@ +/* + * 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.paypal.source; + +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode; + +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.methods.HttpPost; +import org.apache.http.client.methods.HttpRequestBase; +import org.apache.http.client.utils.URIBuilder; +import org.apache.http.entity.ContentType; +import org.apache.http.entity.StringEntity; +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.util.Base64; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +/** + * Owns OAuth renewal, streaming response limits and cancellation for Transaction Search. These + * request-lifecycle requirements are not provided by the shared HTTP source framework; response + * decoding remains separate so transaction amounts are validated without rounding. + */ +final class PayPalClient implements AutoCloseable { + private final PayPalConfig config; + private final CloseableHttpClient client; + private final ScheduledThreadPoolExecutor timer; + private volatile HttpRequestBase active; + private volatile boolean closed; + private String token; + private long expiresAt; + + PayPalClient(PayPalConfig config) { + this.config = config; + client = + HttpClients.custom() + .disableAutomaticRetries() + .disableRedirectHandling() + .disableCookieManagement() + .setDefaultRequestConfig( + RequestConfig.custom() + .setConnectTimeout(config.timeout) + .setSocketTimeout(config.timeout) + .setConnectionRequestTimeout(config.timeout) + .build()) + .build(); + timer = + new ScheduledThreadPoolExecutor( + 1, + r -> { + Thread thread = new Thread(r, "paypal-request-deadline"); + thread.setDaemon(true); + return thread; + }); + timer.setRemoveOnCancelPolicy(true); + } + + JsonNode page(int page) throws Exception { + checkOpen(); + boolean refreshed = false; + for (int attempt = 0; ; attempt++) { + if (token == null || System.nanoTime() >= expiresAt) { + authenticate(); + } + HttpGet request = + new HttpGet( + new URIBuilder(config.origin + "/v1/reporting/transactions") + .addParameter("start_date", config.start.toString()) + .addParameter("end_date", config.end.toString()) + .addParameter("fields", "all") + .addParameter("balance_affecting_records_only", "N") + .addParameter("page_size", Integer.toString(config.pageSize)) + .addParameter("page", Integer.toString(page)) + .build()); + request.setHeader("Authorization", "Bearer " + token); + Reply reply; + try { + reply = execute(request); + } catch (IOException e) { + retry(attempt, config.retryDelay); + continue; + } + if (reply.status == 401 && !refreshed) { + token = null; + refreshed = true; + continue; + } + if (transientStatus(reply.status)) { + retry(attempt, reply.delay); + continue; + } + return success(reply); + } + } + + private void authenticate() throws Exception { + for (int attempt = 0; ; attempt++) { + HttpPost request = new HttpPost(config.origin + "/v1/oauth2/token"); + String basic = + Base64.getEncoder() + .encodeToString( + (config.clientId + ":" + config.clientSecret) + .getBytes(StandardCharsets.UTF_8)); + request.setHeader("Authorization", "Basic " + basic); + request.setEntity( + new StringEntity( + "grant_type=client_credentials", + ContentType.APPLICATION_FORM_URLENCODED)); + long issuedAt = System.nanoTime(); + Reply reply; + try { + reply = execute(request); + } catch (IOException e) { + retry(attempt, config.retryDelay); + continue; + } + if (transientStatus(reply.status)) { + retry(attempt, reply.delay); + continue; + } + JsonNode root = success(reply); + String accessToken = PayPalResponse.text(root, "access_token", true); + int lifetime = PayPalResponse.integer(root, "expires_in"); + if (!"Bearer".equalsIgnoreCase(PayPalResponse.text(root, "token_type", true)) + || lifetime < 1 + || lifetime > 86400 + || !accessToken.matches("[A-Za-z0-9._~+/-]+=*") + || accessToken.length() > 8192) { + throw PayPalResponse.failure("Invalid OAuth token response"); + } + // Count lifetime from before token exchange, never from after a slow response. + expiresAt = issuedAt + TimeUnit.SECONDS.toNanos(lifetime); + token = accessToken; + return; + } + } + + private Reply execute(HttpRequestBase request) throws IOException { + ScheduledFuture<?> deadline; + synchronized (this) { + checkOpen(); + active = request; + deadline = timer.schedule(request::abort, config.timeout, TimeUnit.MILLISECONDS); + } + request.setHeader("Accept", "application/json"); + request.setHeader("PayPal-Enforce-ISO8601-Format", "true"); + CloseableHttpResponse response = null; + try { + response = client.execute(request); + int status = response.getStatusLine().getStatusCode(); + int delay = config.retryDelay; + if (transientStatus(status) && response.getFirstHeader("Retry-After") != null) { + try { + long seconds = + Long.parseLong(response.getFirstHeader("Retry-After").getValue()); + if (seconds < 0 || seconds > 60) { + throw new NumberFormatException(); + } + delay = Math.max(delay, (int) seconds * 1000); + } catch (NumberFormatException e) { + request.abort(); + throw PayPalResponse.failure( + "Retry-After exceeds bounded retry policy or is unsupported; retry the job later"); + } + } + if (transientStatus(status) || status == 401 || status == 403) { + return new Reply(status, new byte[0], delay); + } + if (response.getEntity() == null) { + throw PayPalResponse.failure("Empty HTTP response"); + } + if (response.getEntity().getContentLength() > config.maxBytes) { + request.abort(); + throw PayPalResponse.failure("HTTP response exceeds max_response_bytes"); + } + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + { + InputStream input = response.getEntity().getContent(); + byte[] buffer = new byte[8192]; + int count; + while ((count = input.read(buffer)) != -1) { + checkOpen(); + if (bytes.size() > config.maxBytes - count) { + request.abort(); + throw PayPalResponse.failure("HTTP response exceeds max_response_bytes"); + } + bytes.write(buffer, 0, count); + } + } + return new Reply(status, bytes.toByteArray(), delay); + } finally { + // Abort before closing prevents draining oversized or malformed responses. + request.abort(); + deadline.cancel(false); + if (response != null) { + try { + response.close(); + } catch (IOException ignored) { + // The request is already aborted; never attach transport details to errors. + } + } + synchronized (this) { + active = null; + } + } + } + + private JsonNode success(Reply reply) { + if (reply.status == 401 || reply.status == 403) { + throw PayPalResponse.failure( + "HTTP " + + reply.status + + "; check app credentials and Transaction Search permission (body withheld)"); + } + JsonNode root = PayPalResponse.parse(reply.body); + PayPalResponse.rejectError(root); + if (reply.status != 200) { + throw PayPalResponse.failure( + "HTTP " + + reply.status + + "; check app credentials and Transaction Search permission (body withheld)"); + } + return root; + } + + private static boolean transientStatus(int status) { + return status == 429 || status == 500 || status == 502 || status == 503 || status == 504; + } + + private synchronized void retry(int attempt, int delay) throws InterruptedException { + checkOpen(); + if (attempt >= config.retries) { + throw PayPalResponse.failure( + "HTTP retry budget exhausted (transport details withheld)"); + } + long end = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(delay); + long remaining; + while (!closed && (remaining = end - System.nanoTime()) > 0) { + TimeUnit.NANOSECONDS.timedWait(this, remaining); + } + checkOpen(); + } + + private void checkOpen() { + if (closed || Thread.currentThread().isInterrupted()) { + throw PayPalResponse.failure("Request cancelled"); + } + } + + @Override + public synchronized void close() throws IOException { + closed = true; + token = null; + if (active != null) { + active.abort(); + } + notifyAll(); + timer.shutdownNow(); + client.close(); + } + + private static final class Reply { + private final int status; + private final byte[] body; + private final int delay; + + private Reply(int status, byte[] body, int delay) { + this.status = status; + this.body = body; + this.delay = delay; + } + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalConfig.java b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalConfig.java new file mode 100644 index 0000000000..d4dba9525f --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalConfig.java @@ -0,0 +1,155 @@ +/* + * 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.paypal.source; + +import org.apache.seatunnel.api.configuration.Option; +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.configuration.util.ConfigValidator; +import org.apache.seatunnel.api.options.EnvCommonOptions; + +import java.io.Serializable; +import java.net.URI; +import java.time.Duration; +import java.time.Instant; +import java.time.OffsetDateTime; +import java.time.format.DateTimeFormatter; + +final class PayPalConfig implements Serializable { + final String clientId; + final String clientSecret; + final String origin; + final Instant start; + final Instant end; + final int pageSize; + final int retries; + final int retryDelay; + final int timeout; + final int maxBytes; + + PayPalConfig(ReadonlyConfig config) { + ConfigValidator.validateUnknownKeys( + config, new PayPalSourceFactory().optionRule(), "PayPal"); + if (config.getSourceMap().containsKey("schema")) { + throw new IllegalArgumentException( + "PayPal uses a fixed schema and does not support a schema option"); + } + if (get(config, EnvCommonOptions.PARALLELISM) != 1) { + throw new IllegalArgumentException("PayPal source requires parallelism=1"); + } + clientId = credential(get(config, PayPalSourceOptions.CLIENT_ID), "client_id"); + clientSecret = credential(get(config, PayPalSourceOptions.CLIENT_SECRET), "client_secret"); + start = date(get(config, PayPalSourceOptions.START_DATE)); + end = date(get(config, PayPalSourceOptions.END_DATE)); + if (!start.isBefore(end) + || Duration.between(start, end).compareTo(Duration.ofDays(31)) > 0) { + throw new IllegalArgumentException( + "PayPal requires start_date < end_date and a window <=31 days"); + } + if (get(config, PayPalSourceOptions.MOCK_MODE) + && !(clientId.equals("mock-client") && clientSecret.equals("mock-secret"))) { + throw new IllegalArgumentException( + "PayPal mock_mode requires client_id=mock-client and client_secret=mock-secret"); + } + origin = + origin( + get(config, PayPalSourceOptions.API_BASE_URL), + get(config, PayPalSourceOptions.MOCK_MODE)); + pageSize = range(get(config, PayPalSourceOptions.PAGE_SIZE), 1, 500, "page_size"); + retries = range(get(config, PayPalSourceOptions.MAX_RETRIES), 0, 5, "max_retries"); + retryDelay = + range(get(config, PayPalSourceOptions.RETRY_DELAY_MS), 1, 60000, "retry_delay_ms"); + timeout = + range( + get(config, PayPalSourceOptions.REQUEST_TIMEOUT_MS), + 1, + 120000, + "request_timeout_ms"); + maxBytes = + range( + get(config, PayPalSourceOptions.MAX_RESPONSE_BYTES), + 1024, + 16777216, + "max_response_bytes"); + } + + private static <T> T get(ReadonlyConfig config, Option<T> option) { + try { + return config.get(option); + } catch (RuntimeException e) { + throw new IllegalArgumentException( + "Invalid PayPal option " + option.key() + " (value withheld)"); + } + } + + static Instant date(String value) { + try { + if (value == null + || !value.matches( + "[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(\\.[0-9]{1,9})?(Z|[+-][0-9]{2}:[0-9]{2})")) { + throw new IllegalArgumentException(); + } + return OffsetDateTime.parse(value, DateTimeFormatter.ISO_OFFSET_DATE_TIME).toInstant(); + } catch (RuntimeException e) { + // Parser causes can contain configuration values or response data. + throw new IllegalArgumentException( + "PayPal requires valid absolute RFC3339 dates with seconds and offset"); + } + } + + private static String credential(String value, String key) { + if (value == null + || value.trim().isEmpty() + || value.length() > 4096 + || value.indexOf(':') >= 0 + || value.chars().anyMatch(c -> c < 33 || c > 126)) { + throw new IllegalArgumentException("Invalid PayPal " + key); + } + return value; + } + + private static String origin(String value, boolean mock) { + try { + URI uri = URI.create(value); + boolean official = + value.equals("https://api-m.paypal.com";) + || value.equals("https://api-m.sandbox.paypal.com";); + if ((!official && !mock) + || uri.getHost() == null + || uri.getRawUserInfo() != null + || uri.getRawQuery() != null + || uri.getRawFragment() != null + || !uri.getRawPath().isEmpty() + || !(uri.getScheme().equals("https") + || (mock && uri.getScheme().equals("http")))) { + throw new IllegalArgumentException(); + } + return value; + } catch (RuntimeException e) { + throw new IllegalArgumentException( + "PayPal api_base_url must be an official origin; custom origins require mock_mode and dummy credentials"); + } + } + + private static int range(int value, int min, int max, String key) { + if (value < min || value > max) { + throw new IllegalArgumentException( + "PayPal " + key + " must be between " + min + " and " + max); + } + return value; + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalResponse.java b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalResponse.java new file mode 100644 index 0000000000..b6c4817c07 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalResponse.java @@ -0,0 +1,168 @@ +/* + * 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.paypal.source; + +import org.apache.seatunnel.shade.com.fasterxml.jackson.core.JsonParser; +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.table.type.SeaTunnelRow; + +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.util.ArrayList; +import java.util.List; + +final class PayPalResponse { + private static final ObjectMapper MAPPER = + new ObjectMapper() + .enable(JsonParser.Feature.STRICT_DUPLICATE_DETECTION) + .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS); + final int totalItems; + final int totalPages; + final String account; + final List<SeaTunnelRow> rows; + + PayPalResponse(JsonNode root, PayPalConfig config, int page) { + rejectError(root); + totalItems = integer(root, "total_items"); + totalPages = integer(root, "total_pages"); + if (totalItems > 10000) { + throw failure( + "Result exceeds the 10000-record safety cap; narrow start_date/end_date and reconcile boundaries explicitly"); + } + int expectedPages = (totalItems + config.pageSize - 1) / config.pageSize; + if (integer(root, "page") != page + || (totalPages != expectedPages && !(totalItems == 0 && totalPages == 1))) { + throw failure("Inconsistent page metadata"); + } + account = text(root, "account_number", true); + if (!PayPalConfig.date(text(root, "start_date", true)).equals(config.start) + || !PayPalConfig.date(text(root, "end_date", true)).equals(config.end)) { + throw failure( + "Response coverage differs from requested dates; wait for reporting availability or explicitly narrow the window"); + } + JsonNode details = root.path("transaction_details"); + int expectedRows = Math.min(config.pageSize, totalItems - (page - 1) * config.pageSize); + if (!details.isArray() || details.size() != expectedRows) { + throw failure("Missing, truncated or inconsistent transaction_details"); + } + rows = new ArrayList<>(details.size()); + for (JsonNode detail : details) { + if (!detail.isObject() || !detail.path("transaction_info").isObject()) { + throw failure("Invalid transaction_info object"); + } + JsonNode info = detail.get("transaction_info"); + Object[] amount = money(info.get("transaction_amount")); + Object[] fee = money(info.get("fee_amount")); + rows.add( + new SeaTunnelRow( + new Object[] { + account, + text(info, "transaction_id", false), + text(info, "transaction_event_code", false), + text(info, "transaction_status", false), + timestamp(info, "transaction_initiation_date"), + timestamp(info, "transaction_updated_date"), + amount[0], + amount[1], + fee[0], + fee[1], + detail.toString() + })); + } + } + + static JsonNode parse(byte[] bytes) { + try { + JsonNode root = MAPPER.readTree(bytes); + if (root == null || !root.isObject()) { + throw new IllegalArgumentException(); + } + return root; + } catch (Exception e) { + throw failure("Invalid JSON response (body and parser cause withheld)"); + } + } + + static void rejectError(JsonNode node) { + boolean tooLarge = "RESULTSET_TOO_LARGE".equals(node.path("name").asText()); + for (JsonNode detail : node.path("details")) { + tooLarge |= "RESULTSET_TOO_LARGE".equals(detail.path("issue").asText()); + } + if (tooLarge) { + throw failure( + "RESULTSET_TOO_LARGE: narrow start_date/end_date; automatic lossless splitting is not supported"); + } + if (node.has("name") || node.has("error") || node.has("details")) { + throw failure("API returned an error envelope (body withheld)"); + } + } + + static int integer(JsonNode node, String field) { + JsonNode value = node.path(field); + if (!value.isIntegralNumber() || !value.canConvertToInt() || value.intValue() < 0) { + throw failure("Missing or invalid " + field); + } + return value.intValue(); + } + + static String text(JsonNode node, String field, boolean required) { + JsonNode value = node.get(field); + if (!required && (value == null || value.isNull())) { + return null; + } + if (value == null || !value.isTextual() || (required && value.textValue().isEmpty())) { + throw failure("Missing or invalid " + field); + } + return value.textValue(); + } + + private static String timestamp(JsonNode node, String field) { + String value = text(node, field, false); + return value == null ? null : PayPalConfig.date(value).toString(); + } + + private static Object[] money(JsonNode node) { + if (node == null || node.isNull()) { + return new Object[] {null, null}; + } + String value = text(node, "value", true); + String currency = text(node, "currency_code", true); + if (!node.isObject() + || value.length() > 32 + || !value.matches("-?([0-9]+|[0-9]*\\.[0-9]+)") + || !currency.matches("[A-Z]{3}")) { + throw failure("Malformed currency amount"); + } + try { + BigDecimal decimal = new BigDecimal(value).setScale(9, RoundingMode.UNNECESSARY); + if (decimal.precision() > 38) { + throw new ArithmeticException(); + } + return new Object[] {decimal, currency}; + } catch (ArithmeticException e) { + throw failure("Currency amount cannot be represented exactly as DECIMAL(38,9)"); + } + } + + static IllegalStateException failure(String message) { + return new IllegalStateException("PayPal: " + message); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalSource.java b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalSource.java new file mode 100644 index 0000000000..07d9ede187 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalSource.java @@ -0,0 +1,100 @@ +/* + * 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.paypal.source; + +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.PhysicalColumn; +import org.apache.seatunnel.api.table.catalog.TableIdentifier; +import org.apache.seatunnel.api.table.catalog.TablePath; +import org.apache.seatunnel.api.table.catalog.TableSchema; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.DecimalType; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +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; + +/** A single-account bounded report, preserving reporting records without snapshot guarantees. */ +public class PayPalSource extends AbstractSingleSplitSource<SeaTunnelRow> { + public static final String PLUGIN_NAME = "PayPal"; + private final PayPalConfig config; + + public PayPalSource(ReadonlyConfig options) { + config = new PayPalConfig(options); + } + + @Override + public String getPluginName() { + return PLUGIN_NAME; + } + + @Override + public void setJobContext(JobContext context) { + if (context.getJobMode() != JobMode.BATCH) { + throw new IllegalArgumentException("PayPal source supports BATCH only"); + } + } + + @Override + public Boundedness getBoundedness() { + return Boundedness.BOUNDED; + } + + @Override + public List<CatalogTable> getProducedCatalogTables() { + TableSchema.Builder schema = TableSchema.builder(); + String[] names = { + "account_number", + "transaction_id", + "transaction_event_code", + "transaction_status", + "transaction_initiation_date", + "transaction_updated_date", + "transaction_amount", + "transaction_currency", + "fee_amount", + "fee_currency", + "content" + }; + for (int i = 0; i < names.length; i++) { + SeaTunnelDataType<?> type = + i == 6 || i == 8 ? new DecimalType(38, 9) : BasicType.STRING_TYPE; + schema.column(PhysicalColumn.of(names[i], type, 0, i != 0 && i != 10, null, null)); + } + return Collections.singletonList( + CatalogTable.of( + TableIdentifier.of(PLUGIN_NAME, TablePath.DEFAULT), + schema.build(), + Collections.emptyMap(), + Collections.emptyList(), + null)); + } + + @Override + public AbstractSingleSplitReader<SeaTunnelRow> createReader(SingleSplitReaderContext context) { + return new PayPalSourceReader(config, context); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalSourceFactory.java b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalSourceFactory.java new file mode 100644 index 0000000000..dc3db51323 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalSourceFactory.java @@ -0,0 +1,70 @@ +/* + * 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.paypal.source; + +import org.apache.seatunnel.api.configuration.util.OptionRule; +import org.apache.seatunnel.api.options.MultiTableCommonOptions; +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; + +@AutoService(Factory.class) +public class PayPalSourceFactory implements TableSourceFactory { + @Override + public String factoryIdentifier() { + return PayPalSource.PLUGIN_NAME; + } + + @Override + public OptionRule optionRule() { + return OptionRule.builder() + .required( + PayPalSourceOptions.CLIENT_ID, + PayPalSourceOptions.CLIENT_SECRET, + PayPalSourceOptions.START_DATE, + PayPalSourceOptions.END_DATE) + .optional( + MultiTableCommonOptions.MULTI_TABLE_FAILURE_POLICY, + PayPalSourceOptions.API_BASE_URL, + PayPalSourceOptions.MOCK_MODE, + PayPalSourceOptions.PAGE_SIZE, + PayPalSourceOptions.MAX_RETRIES, + PayPalSourceOptions.RETRY_DELAY_MS, + PayPalSourceOptions.REQUEST_TIMEOUT_MS, + PayPalSourceOptions.MAX_RESPONSE_BYTES) + .build(); + } + + @Override + public <T, SplitT extends SourceSplit, StateT extends Serializable> + TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) { + return () -> (SeaTunnelSource<T, SplitT, StateT>) new PayPalSource(context.getOptions()); + } + + @Override + public Class<? extends SeaTunnelSource> getSourceClass() { + return PayPalSource.class; + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalSourceOptions.java b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalSourceOptions.java new file mode 100644 index 0000000000..fd4beb38ca --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalSourceOptions.java @@ -0,0 +1,85 @@ +/* + * 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.paypal.source; + +import org.apache.seatunnel.api.configuration.Option; +import org.apache.seatunnel.api.configuration.Options; + +public final class PayPalSourceOptions { + public static final Option<String> CLIENT_ID = + Options.key("client_id") + .stringType() + .noDefaultValue() + .withDescription("First-party PayPal REST app client ID."); + public static final Option<String> CLIENT_SECRET = + Options.key("client_secret") + .stringType() + .noDefaultValue() + .withDescription("PayPal client secret; masked in parsed job configuration."); + public static final Option<String> START_DATE = + Options.key("start_date") + .stringType() + .noDefaultValue() + .withDescription("Absolute RFC3339 start with seconds and offset."); + public static final Option<String> END_DATE = + Options.key("end_date") + .stringType() + .noDefaultValue() + .withDescription( + "Absolute RFC3339 end, at most 31 days after start; shortened response coverage fails."); + public static final Option<String> API_BASE_URL = + Options.key("api_base_url") + .stringType() + .defaultValue("https://api-m.paypal.com";) + .withDescription("PayPal production or sandbox origin."); + public static final Option<Boolean> MOCK_MODE = + Options.key("mock_mode") + .booleanType() + .defaultValue(false) + .withDescription( + "Allow an HTTP mock origin with dummy credentials only; never enable for PayPal."); + public static final Option<Integer> PAGE_SIZE = + Options.key("page_size") + .intType() + .defaultValue(100) + .withDescription("Records per page, 1 to 500."); + public static final Option<Integer> MAX_RETRIES = + Options.key("max_retries") + .intType() + .defaultValue(3) + .withDescription( + "Additional attempts for HTTP 429/5xx and transport failures, 0 to 5."); + public static final Option<Integer> RETRY_DELAY_MS = + Options.key("retry_delay_ms") + .intType() + .defaultValue(1000) + .withDescription( + "Retry delay, 1 to 60000 ms. Longer Retry-After fails instead of retrying early."); + public static final Option<Integer> REQUEST_TIMEOUT_MS = + Options.key("request_timeout_ms") + .intType() + .defaultValue(30000) + .withDescription("Connect, socket and request abort deadline, 1 to 120000 ms."); + public static final Option<Integer> MAX_RESPONSE_BYTES = + Options.key("max_response_bytes") + .intType() + .defaultValue(8388608) + .withDescription("Maximum uncompressed response bytes, 1024 to 16777216."); + + private PayPalSourceOptions() {} +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalSourceReader.java b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalSourceReader.java new file mode 100644 index 0000000000..6b52e22254 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/main/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalSourceReader.java @@ -0,0 +1,95 @@ +/* + * 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.paypal.source; + +import org.apache.seatunnel.api.source.Collector; +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 java.io.IOException; + +final class PayPalSourceReader extends AbstractSingleSplitReader<SeaTunnelRow> { + private final PayPalConfig config; + private final SingleSplitReaderContext context; + private volatile boolean closed; + private PayPalClient client; + + PayPalSourceReader(PayPalConfig config, SingleSplitReaderContext context) { + this.config = config; + this.context = context; + } + + /** Create worker-local HTTP resources; credentials and clients never enter split state. */ + @Override + public synchronized void open() { + if (client != null) { + throw PayPalResponse.failure("Reader already opened"); + } + if (closed) { + throw PayPalResponse.failure("Reader closed"); + } + client = new PayPalClient(config); + } + + /** Replay the bounded report on recovery; page offsets are not durable snapshots. */ + @Override + public void internalPollNext(Collector<SeaTunnelRow> output) throws Exception { + int total = -1; + int pages = -1; + String account = null; + for (int page = 1; ; page++) { + if (closed) { + throw PayPalResponse.failure("Reader closed"); + } + PayPalResponse response = new PayPalResponse(client.page(page), config, page); + if (total < 0) { + total = response.totalItems; + pages = response.totalPages; + account = response.account; + } else if (total != response.totalItems + || pages != response.totalPages + || !account.equals(response.account)) { + throw PayPalResponse.failure( + "Report totals or account changed between pages; no stable remote snapshot is guaranteed"); + } + for (SeaTunnelRow row : response.rows) { + if (closed) { + throw PayPalResponse.failure("Reader closed"); + } + output.collect(row); + } + if (page >= pages) { + break; + } + } + if (closed) { + throw PayPalResponse.failure("Reader closed"); + } + context.signalNoMoreElement(); + } + + /** Abort in-flight HTTP and wake retry waits before releasing worker-local resources. */ + @Override + public synchronized void close() throws IOException { + closed = true; + if (client != null) { + client.close(); + } + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/test/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalClientTest.java b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/test/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalClientTest.java new file mode 100644 index 0000000000..b72899d907 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/test/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalClientTest.java @@ -0,0 +1,487 @@ +/* + * 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.paypal.source; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.source.Collector; +import org.apache.seatunnel.api.source.SourceReader; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext; + +import org.junit.jupiter.api.AfterEach; +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.HttpExchange; +import com.sun.net.httpserver.HttpServer; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Base64; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.zip.GZIPOutputStream; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class PayPalClientTest { + @Test + void statusPolicyWorksWithBodyless503429And401() throws Exception { + Reply unavailable = new Reply(503, ""); + unavailable.bodyless = true; + replies.add(unavailable); + token(3600, "first-token"); + Reply unauthorized = new Reply(401, ""); + unauthorized.bodyless = true; + replies.add(unauthorized); + token(3600, "next-token"); + page(1, 0, ""); + assertNotNull(client().page(1)); + assertEquals(5, requests.size()); + } + + @Test + void bodylessRateLimitUsesBoundedRetry() throws Exception { + token(3600, "token"); + Reply limit = new Reply(429, ""); + limit.bodyless = true; + replies.add(limit); + page(1, 0, ""); + assertNotNull(client().page(1)); + assertEquals(3, requests.size()); + } + + @Test + void repeatedOpenIsRejectedAndCloseIsIdempotent() throws Exception { + PayPalSourceReader reader = + new PayPalSourceReader( + new PayPalConfig(ReadonlyConfig.fromMap(options)), + new SingleSplitReaderContext(mock(SourceReader.Context.class))); + reader.open(); + try { + assertThrows(IllegalStateException.class, reader::open); + } finally { + reader.close(); + reader.close(); + } + assertThrows(IllegalStateException.class, reader::open); + } + + private HttpServer server; + private ExecutorService executor; + private final Queue<Reply> replies = new ConcurrentLinkedQueue<>(); + private final List<String> requests = Collections.synchronizedList(new ArrayList<>()); + private final CountDownLatch arrived = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + private Map<String, Object> options; + private PayPalClient client; + + @BeforeEach + void start() throws Exception { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + executor = Executors.newCachedThreadPool(); + server.setExecutor(executor); + server.createContext("/", this::serve); + server.start(); + options = PayPalResponseTest.options(); + options.put("mock_mode", true); + options.put("api_base_url", "http://127.0.0.1:"; + server.getAddress().getPort()); + options.put("retry_delay_ms", 10); + options.put("max_retries", 1); + } + + private void serve(HttpExchange exchange) throws IOException { + Reply reply = replies.poll(); + String body = new String(read(exchange.getRequestBody()), StandardCharsets.UTF_8); + requests.add( + exchange.getRequestMethod() + + " " + + exchange.getRequestURI() + + " " + + exchange.getRequestHeaders().getFirst("Authorization") + + " " + + body + + " " + + exchange.getRequestHeaders().getFirst("PayPal-Enforce-ISO8601-Format")); + if (reply == null) { + exchange.sendResponseHeaders(500, -1); + exchange.close(); + return; + } + try { + if (reply.delay > 0) { + Thread.sleep(reply.delay); + } + if (reply.gzip) { + exchange.getResponseHeaders().set("Content-Encoding", "gzip"); + } + if (reply.retryAfter != null) { + exchange.getResponseHeaders().set("Retry-After", reply.retryAfter); + } + if (reply.status == 302) { + exchange.getResponseHeaders() + .set("Location", options.get("api_base_url") + "/credential-theft"); + } + exchange.sendResponseHeaders( + reply.status, reply.bodyless ? -1 : reply.chunked ? 0 : reply.body.length); + if (reply.bodyless) { + arrived.countDown(); + return; + } + if (reply.block) { + exchange.getResponseBody().write(' '); + exchange.getResponseBody().flush(); + arrived.countDown(); + release.await(5, TimeUnit.SECONDS); + } else { + exchange.getResponseBody().write(reply.body); + arrived.countDown(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (IOException ignored) { + /* Cancellation intentionally closes the peer socket. */ + } finally { + exchange.close(); + } + } + + private static byte[] read(InputStream input) throws IOException { + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + byte[] buffer = new byte[1024]; + int count; + while ((count = input.read(buffer)) != -1) { + bytes.write(buffer, 0, count); + } + return bytes.toByteArray(); + } + + private PayPalClient client() { + client = new PayPalClient(new PayPalConfig(ReadonlyConfig.fromMap(options))); + return client; + } + + private void token(int lifetime, String value) { + replies.add( + new Reply( + 200, + "{\"access_token\":\"" + + value + + "\",\"token_type\":\"Bearer\",\"expires_in\":" + + lifetime + + "}")); + } + + private Reply page(int number, int total, String records) { + Reply reply = new Reply(200, PayPalResponseTest.page(number, total, records)); + replies.add(reply); + return reply; + } + + @AfterEach + void stop() throws Exception { + release.countDown(); + if (client != null) { + client.close(); + } + server.stop(0); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + + @Test + void exchangesOAuthAndRefreshesExpiredTokenAcrossPages() throws Exception { + token(1, "first-token"); + page( + 1, + 3, + PayPalResponseTest.record("123", "JPY") + + "," + + PayPalResponseTest.record("0.123", "TND")) + .delay = + 1200; + token(3600, "next-token"); + page(2, 3, PayPalResponseTest.record("-1.23", "USD")); + PayPalClient transport = client(); + assertEquals(1, transport.page(1).path("page").intValue()); + assertEquals(2, transport.page(2).path("page").intValue()); + assertEquals(4, requests.size()); + String basic = + Base64.getEncoder() + .encodeToString("mock-client:mock-secret".getBytes(StandardCharsets.UTF_8)); + assertTrue( + requests.get(0) + .startsWith( + "POST /v1/oauth2/token Basic " + + basic + + " grant_type=client_credentials")); + assertTrue(requests.get(1).contains("balance_affecting_records_only=N")); + assertTrue(requests.get(1).contains("fields=all")); + assertTrue(requests.get(1).contains("page_size=2")); + assertTrue(requests.get(1).contains("Bearer first-token")); + assertTrue(requests.get(3).contains("Bearer next-token")); + assertTrue(requests.get(3).endsWith("true")); + } + + @Test + void unauthorizedRefreshIsBoundedAndSanitized() throws Exception { + token(3600, "first-token"); + replies.add(new Reply(401, "secret-error-body")); + token(3600, "second-token"); + replies.add(new Reply(401, "secret-error-body")); + Exception error = assertThrows(Exception.class, () -> client().page(1)); + assertTrue(error.getMessage().contains("401")); + assertFalse(error.toString().contains("secret-error-body")); + assertNull(error.getCause()); + assertEquals(4, requests.size()); + } + + @ParameterizedTest + @ValueSource(ints = {400, 401, 403}) + void invalidOAuthCredentialsFailWithoutRetries(int status) { + replies.add( + new Reply( + status, + "{\"error\":\"invalid_client\",\"error_description\":\"secret-body\"}")); + Exception error = assertThrows(Exception.class, () -> client().page(1)); + assertFalse(error.toString().contains("secret-body")); + assertNull(error.getCause()); + assertEquals(1, requests.size()); + } + + @ParameterizedTest + @ValueSource( + strings = { + "{\"access_token\":\"secret-value\",\"expires_in\":-1,\"token_type\":\"Bearer\"}", + "{\"access_token\":\"secret-value\",\"expires_in\":1.5,\"token_type\":\"Bearer\"}", + "{\"access_token\":\"secret-value\",\"expires_in\":3600,\"token_type\":\"Basic\"}", + "{\"access_token\":\"secret-value\",\"expires_in\":3600}" + }) + void rejectsMalformedTokensWithoutCauses(String body) { + replies.add(new Reply(200, body)); + Exception error = assertThrows(Exception.class, () -> client().page(1)); + assertFalse(error.toString().contains("secret-value")); + assertNull(error.getCause()); + } + + @Test + void retriesOAuthAndReportTransientResponses() throws Exception { + replies.add(new Reply(503, "unavailable")); + token(3600, "token"); + replies.add(new Reply(429, "rate-limited")); + page(1, 0, ""); + assertEquals(0, client().page(1).path("total_items").intValue()); + assertEquals(4, requests.size()); + } + + @Test + void rejectsLongRetryAfterAndRedirect() { + token(3600, "token"); + Reply busy = new Reply(429, "secret-body"); + busy.retryAfter = "61"; + replies.add(busy); + assertThrows(Exception.class, () -> client().page(1)); + assertEquals(2, requests.size()); + } + + @Test + void neverFollowsRedirectWithCredentials() { + replies.add(new Reply(302, "{}")); + assertThrows(Exception.class, () -> client().page(1)); + assertEquals(1, requests.size()); + assertFalse(requests.get(0).contains("credential-theft")); + } + + @Test + void failsOnResultsetTooLargeAndPermissionErrors() throws Exception { + token(3600, "token"); + replies.add( + new Reply(400, "{\"name\":\"RESULTSET_TOO_LARGE\",\"message\":\"secret-body\"}")); + Exception error = assertThrows(Exception.class, () -> client().page(1)); + assertTrue(error.getMessage().contains("narrow")); + assertFalse(error.toString().contains("secret-body")); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void boundsPlainAndGzipResponses(boolean gzip) throws Exception { + options.put("max_response_bytes", 1024); + Reply huge = new Reply(200, String.join("", Collections.nCopies(2048, "x"))); + huge.chunked = true; + if (gzip) { + ByteArrayOutputStream compressed = new ByteArrayOutputStream(); + try (GZIPOutputStream output = new GZIPOutputStream(compressed)) { + output.write(huge.body); + } + huge.body = compressed.toByteArray(); + huge.gzip = true; + } + replies.add(huge); + assertThrows(Exception.class, () -> client().page(1)); + assertEquals(1, requests.size()); + } + + @Test + void deadlineAbortsStalledBodyAndRetryBudgetTerminates() { + options.put("request_timeout_ms", 150); + options.put("max_retries", 0); + Reply stalled = new Reply(200, "{}"); + stalled.block = true; + stalled.chunked = true; + replies.add(stalled); + long before = System.nanoTime(); + assertThrows(Exception.class, () -> client().page(1)); + assertTrue(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - before) < 3000); + } + + @Test + void closeAbortsActiveBody() throws Exception { + Reply stalled = new Reply(200, "{}"); + stalled.block = true; + stalled.chunked = true; + replies.add(stalled); + PayPalClient transport = client(); + Future<?> result = + executor.submit(() -> assertThrows(Exception.class, () -> transport.page(1))); + assertTrue(arrived.await(3, TimeUnit.SECONDS)); + transport.close(); + result.get(3, TimeUnit.SECONDS); + } + + @Test + void closeWakesRetryWait() throws Exception { + options.put("retry_delay_ms", 60000); + replies.add(new Reply(503, "{}")); + PayPalClient transport = client(); + CountDownLatch complete = new CountDownLatch(1); + Thread worker = + new Thread( + () -> { + try { + assertThrows(Exception.class, () -> transport.page(1)); + } finally { + complete.countDown(); + } + }); + worker.start(); + assertTrue(arrived.await(3, TimeUnit.SECONDS)); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(3); + while (worker.getState() != Thread.State.TIMED_WAITING && System.nanoTime() < deadline) { + Thread.sleep(10); + } + assertEquals(Thread.State.TIMED_WAITING, worker.getState()); + transport.close(); + assertTrue(complete.await(3, TimeUnit.SECONDS)); + worker.join(); + assertEquals(1, requests.size()); + } + + @Test + void readerPreservesRecordsAndSignalsCompletionOnlyAfterValidation() throws Exception { + token(3600, "token"); + page( + 1, + 3, + PayPalResponseTest.record("1", "USD") + + "," + + PayPalResponseTest.record("2", "USD")); + page(2, 3, "{\"transaction_info\":{}}"); + SourceReader.Context context = mock(SourceReader.Context.class); + Collector<SeaTunnelRow> collector = mock(Collector.class); + when(collector.getCheckpointLock()).thenReturn(new Object()); + try (PayPalSourceReader reader = + new PayPalSourceReader( + new PayPalConfig(ReadonlyConfig.fromMap(options)), + new SingleSplitReaderContext(context))) { + reader.open(); + reader.pollNext(collector); + reader.pollNext(collector); + verify(collector, times(3)).collect(any(SeaTunnelRow.class)); + verify(context).signalNoMoreElement(); + } + } + + @Test + void changedTotalsFailWithoutCompletionSignal() throws Exception { + token(3600, "token"); + page( + 1, + 3, + PayPalResponseTest.record("1", "USD") + + "," + + PayPalResponseTest.record("2", "USD")); + page( + 2, + 4, + PayPalResponseTest.record("3", "USD") + + "," + + PayPalResponseTest.record("4", "USD")); + SourceReader.Context context = mock(SourceReader.Context.class); + Collector<SeaTunnelRow> collector = mock(Collector.class); + try (PayPalSourceReader reader = + new PayPalSourceReader( + new PayPalConfig(ReadonlyConfig.fromMap(options)), + new SingleSplitReaderContext(context))) { + reader.open(); + assertThrows(Exception.class, () -> reader.internalPollNext(collector)); + verify(context, never()).signalNoMoreElement(); + verify(collector, times(2)).collect(any(SeaTunnelRow.class)); + } + } + + private static class Reply { + private final int status; + private byte[] body; + private int delay; + private boolean gzip; + private boolean chunked; + private boolean block; + private boolean bodyless; + private String retryAfter; + + private Reply(int status, String body) { + this.status = status; + this.body = body.getBytes(StandardCharsets.UTF_8); + } + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/test/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalResponseTest.java b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/test/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalResponseTest.java new file mode 100644 index 0000000000..c58faf1aa7 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/test/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/PayPalResponseTest.java @@ -0,0 +1,402 @@ +/* + * 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.paypal.source; + +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode; +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode; + +import org.apache.seatunnel.api.common.JobContext; +import org.apache.seatunnel.api.common.multitable.MultiTableFailureHelper; +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.source.Boundedness; +import org.apache.seatunnel.api.table.factory.Factory; +import org.apache.seatunnel.common.constants.JobMode; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.ByteArrayOutputStream; +import java.io.ObjectOutputStream; +import java.math.BigDecimal; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; +import java.util.ServiceLoader; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class PayPalResponseTest { + @Test + void permitsExactlyTenThousandWithConsistentFinalPage() { + PayPalResponse response = + new PayPalResponse( + json(page(5000, 10000, record("1", "USD") + "," + record("2", "USD"))), + config(), + 5000); + assertEquals(10000, response.totalItems); + assertEquals(2, response.rows.size()); + } + + @Test + void acceptsNativeMetadataAndNestedDagOptions() { + Map<String, Object> values = options(); + values.put("metadata_datasource_id", "data-source"); + values.put("dag-parsing.mode", "SINGLENESS"); + assertNotNull(new PayPalConfig(ReadonlyConfig.fromMap(values))); + values.remove("dag-parsing.mode"); + values.put("dag-parsing", java.util.Collections.singletonMap("mode", "SINGLENESS")); + assertNotNull(new PayPalConfig(ReadonlyConfig.fromMap(values))); + } + + @Test + void acceptsEngineInjectedFailurePolicyWithoutMultiTableCapability() { + ReadonlyConfig injected = + MultiTableFailureHelper.withMultiTableFailurePolicy( + ReadonlyConfig.fromMap(options()), + ReadonlyConfig.fromMap(java.util.Collections.emptyMap())); + assertNotNull(new PayPalSource(injected)); + } + + @Test + void invalidOptionTypesNeverExposeValuesOrCauses() { + for (String key : + new String[] { + "page_size", + "max_retries", + "request_timeout_ms", + "retry_delay_ms", + "max_response_bytes", + "parallelism", + "mock_mode" + }) { + Map<String, Object> values = options(); + values.put(key, "DO-NOT-LOG-VALUE"); + RuntimeException error = + assertThrows( + RuntimeException.class, + () -> new PayPalSource(ReadonlyConfig.fromMap(values))); + assertFalse(error.toString().contains("DO-NOT-LOG-VALUE")); + assertNull(error.getCause()); + } + } + + @Test + void rejectsUnsupportedOptionsAndExecutionModes() { + for (String key : + new String[] { + "schema", + "fields", + "access_token", + "balance_affecting_records_only", + "unknown_option" + }) { + Map<String, Object> options = options(); + options.put(key, "value-not-to-log"); + assertThrows( + RuntimeException.class, + () -> new PayPalSource(ReadonlyConfig.fromMap(options))); + } + Map<String, Object> options = options(); + options.put("parallelism", 2); + assertThrows( + RuntimeException.class, () -> new PayPalSource(ReadonlyConfig.fromMap(options))); + PayPalSource source = new PayPalSource(ReadonlyConfig.fromMap(options())); + assertThrows( + IllegalArgumentException.class, + () -> source.setJobContext(new JobContext().setJobMode(JobMode.STREAMING))); + source.setJobContext(new JobContext().setJobMode(JobMode.BATCH)); + } + + static Map<String, Object> options() { + Map<String, Object> options = new HashMap<>(); + options.put("client_id", "mock-client"); + options.put("client_secret", "mock-secret"); + options.put("start_date", "2026-01-01T00:00:00Z"); + options.put("end_date", "2026-01-02T00:00:00Z"); + options.put("page_size", 2); + return options; + } + + static PayPalConfig config() { + return new PayPalConfig(ReadonlyConfig.fromMap(options())); + } + + static String record(String amount, String currency) { + return "{\"transaction_info\":{\"transaction_id\":\"SAME-ID\",\"transaction_amount\":{\"value\":\"" + + amount + + "\",\"currency_code\":\"" + + currency + + "\"}}}"; + } + + static String page(int number, int total, String records) { + return "{\"account_number\":\"ACCOUNT\",\"start_date\":\"2026-01-01T00:00:00Z\",\"end_date\":\"2026-01-02T00:00:00Z\",\"page\":" + + number + + ",\"total_items\":" + + total + + ",\"total_pages\":" + + ((total + 1) / 2) + + ",\"transaction_details\":[" + + records + + "]}"; + } + + static JsonNode json(String value) { + return PayPalResponse.parse(value.getBytes(StandardCharsets.UTF_8)); + } + + @Test + void preservesDuplicateIdsAndExactCurrencyAmounts() { + PayPalResponse response = + new PayPalResponse( + json(page(1, 2, record("123", "JPY") + "," + record("-0.123", "TND"))), + config(), + 1); + assertEquals(2, response.rows.size()); + assertEquals(response.rows.get(0).getField(1), response.rows.get(1).getField(1)); + assertEquals(new BigDecimal("123.000000000"), response.rows.get(0).getField(6)); + assertEquals(new BigDecimal("-0.123000000"), response.rows.get(1).getField(6)); + assertNull(response.rows.get(0).getField(8)); + } + + @Test + void optionalFieldsRemainNullAndRawRecordIsPreserved() { + String record = + "{\"transaction_info\":{\"transaction_id\":null,\"fee_amount\":null,\"paypal_reference_id\":null},\"payer_info\":{\"unknown\":\"kept\"}}"; + PayPalResponse response = new PayPalResponse(json(page(1, 1, record)), config(), 1); + for (int index = 1; index < 10; index++) { + assertNull(response.rows.get(0).getField(index)); + } + assertEquals(json(record), json((String) response.rows.get(0).getField(10))); + } + + @Test + void acceptsEmptyResultsAndEquivalentOffsets() { + String content = + page(1, 0, "").replace("2026-01-01T00:00:00Z", "2026-01-01T01:00:00+01:00"); + assertTrue(new PayPalResponse(json(content), config(), 1).rows.isEmpty()); + assertTrue( + new PayPalResponse( + json(content.replace("\"total_pages\":0", "\"total_pages\":1")), + config(), + 1) + .rows.isEmpty()); + } + + @ParameterizedTest + @ValueSource( + strings = { + "page", + "total_items", + "total_pages", + "start_date", + "end_date", + "account_number", + "transaction_details" + }) + void rejectsMissingEnvelopeFields(String field) { + ObjectNode root = (ObjectNode) json(page(1, 1, record("1", "USD"))); + root.remove(field); + assertThrows(RuntimeException.class, () -> new PayPalResponse(root, config(), 1)); + } + + @ParameterizedTest + @ValueSource( + strings = { + "\"transaction_info\":null", + "\"transaction_info\":[]", + "\"transaction_info\":{\"transaction_amount\":{}}", + "\"transaction_info\":{\"transaction_status\":123}", + "\"transaction_info\":{\"fee_amount\":{\"value\":1,\"currency_code\":\"USD\"}}", + "\"transaction_info\":{\"transaction_initiation_date\":\"secret-malformed-date\"}" + }) + void rejectsMalformedRecordsWithoutLeakingValues(String content) { + RuntimeException error = + assertThrows( + RuntimeException.class, + () -> + new PayPalResponse( + json(page(1, 1, "{" + content + "}")), config(), 1)); + assertFalse(error.toString().contains("secret-malformed-date")); + assertNull(error.getCause()); + } + + @ParameterizedTest + @ValueSource( + strings = {"1e3", "NaN", "0.0000000001", "123456789012345678901234567890", "1.0secret"}) + void refusesLossyOrInvalidMoney(String amount) { + assertThrows( + RuntimeException.class, + () -> new PayPalResponse(json(page(1, 1, record(amount, "USD"))), config(), 1)); + } + + @Test + void rejectsShortCoverageAndTruncationAndCap() { + assertThrows( + RuntimeException.class, + () -> + new PayPalResponse( + json( + page(1, 0, "") + .replace( + "2026-01-02T00:00:00Z", + "2026-01-01T23:59:59Z")), + config(), + 1)); + assertThrows( + RuntimeException.class, + () -> new PayPalResponse(json(page(1, 2, record("1", "USD"))), config(), 1)); + assertThrows( + RuntimeException.class, + () -> new PayPalResponse(json(page(1, 10000, "")), config(), 1)); + assertThrows( + RuntimeException.class, + () -> + new PayPalResponse( + json( + page(1, 0, "") + .replace( + "\"total_items\":0", + "\"total_items\":0.0")), + config(), + 1)); + } + + @Test + void detectsApiErrorsAndMalformedJson() { + for (String error : + new String[] { + "{\"name\":\"RESULTSET_TOO_LARGE\",\"message\":\"secret\"}", + "{\"details\":[{\"issue\":\"RESULTSET_TOO_LARGE\"}]}" + }) { + IllegalStateException exception = + assertThrows( + IllegalStateException.class, + () -> PayPalResponse.rejectError(json(error))); + assertTrue(exception.getMessage().contains("narrow")); + assertFalse(exception.toString().contains("secret")); + } + for (String malformed : + new String[] {"{\"secret\":", "{}", "{} {}", "{\"a\":1,\"a\":2}", "[]", "null"}) { + assertThrows( + RuntimeException.class, () -> new PayPalResponse(json(malformed), config(), 1)); + } + } + + @ParameterizedTest + @ValueSource( + strings = { + "2026-01-01", + "today", + "2026-02-30T00:00:00Z", + "2026-01-01T00:00:00", + "2026-01-01T00:00Z" + }) + void requiresAbsoluteDatesWithSeconds(String value) { + assertThrows(IllegalArgumentException.class, () -> PayPalConfig.date(value)); + } + + @Test + void validatesConfigAndMockCredentialsBeforeHttp() { + for (String key : + new String[] { + "page_size", + "max_retries", + "retry_delay_ms", + "request_timeout_ms", + "max_response_bytes" + }) { + Map<String, Object> values = options(); + values.put(key, -1); + assertThrows( + IllegalArgumentException.class, + () -> new PayPalConfig(ReadonlyConfig.fromMap(values))); + } + for (String endpoint : + new String[] { + "http://localhost:8080";, + "https://api-m.paypal.com/";, + "https://[email protected]";, + "https://evil.example";, + "https://api-m.paypal.com?secret=yes"; + }) { + Map<String, Object> values = options(); + values.put("api_base_url", endpoint); + assertThrows( + IllegalArgumentException.class, + () -> new PayPalConfig(ReadonlyConfig.fromMap(values))); + } + Map<String, Object> values = options(); + values.put("mock_mode", true); + values.put("api_base_url", "http://localhost:8080";); + assertNotNull(new PayPalConfig(ReadonlyConfig.fromMap(values))); + values.put("client_secret", "real-secret"); + assertThrows( + IllegalArgumentException.class, + () -> new PayPalConfig(ReadonlyConfig.fromMap(values))); + values.put("client_secret", "mock-secret"); + values.put("client_id", "real-id"); + assertThrows( + IllegalArgumentException.class, + () -> new PayPalConfig(ReadonlyConfig.fromMap(values))); + } + + @Test + void validatesIntervalAndHasNoClockDependentAgeRestriction() { + Map<String, Object> values = options(); + values.put("end_date", "2026-02-02T00:00:00Z"); + assertThrows( + IllegalArgumentException.class, + () -> new PayPalConfig(ReadonlyConfig.fromMap(values))); + values.put("end_date", values.get("start_date")); + assertThrows( + IllegalArgumentException.class, + () -> new PayPalConfig(ReadonlyConfig.fromMap(values))); + values.put("start_date", "2099-01-01T00:00:00Z"); + values.put("end_date", "2099-02-01T00:00:00Z"); + assertNotNull(new PayPalConfig(ReadonlyConfig.fromMap(values))); + } + + @Test + void factorySpiAndSourceSerialization() throws Exception { + boolean found = false; + for (Factory factory : ServiceLoader.load(Factory.class)) { + if (factory.factoryIdentifier().equals("PayPal")) { + found = true; + assertTrue(factory instanceof PayPalSourceFactory); + } + } + assertTrue(found); + PayPalSource source = new PayPalSource(ReadonlyConfig.fromMap(options())); + assertEquals(Boundedness.BOUNDED, source.getBoundedness()); + assertEquals( + 11, + source.getProducedCatalogTables().get(0).getSeaTunnelRowType().getTotalFields()); + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream output = new ObjectOutputStream(bytes)) { + output.writeObject(source); + } + assertTrue(bytes.size() > 0); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/test/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/TestPayPalPackagingIT.java b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/test/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/TestPayPalPackagingIT.java new file mode 100644 index 0000000000..e26a1379f1 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-paypal/src/test/java/org/apache/seatunnel/connectors/seatunnel/paypal/source/TestPayPalPackagingIT.java @@ -0,0 +1,141 @@ +/* + * 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.paypal.source; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.table.factory.Factory; + +import org.junit.jupiter.api.Test; + +import com.sun.net.httpserver.HttpServer; + +import java.io.File; +import java.lang.reflect.Constructor; +import java.lang.reflect.Method; +import java.net.InetSocketAddress; +import java.net.URL; +import java.net.URLClassLoader; +import java.nio.charset.StandardCharsets; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.jar.JarFile; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class TestPayPalPackagingIT { + @Test + void packagedFactoryAndOAuthTransportLoadWithoutConnectorParentFallback() throws Exception { + File[] files = + new File("target") + .listFiles( + (dir, name) -> + name.startsWith("connector-http-paypal-") + && name.endsWith(".jar") + && !name.contains("sources") + && !name.contains("tests")); + assertNotNull(files); + assertEquals(1, files.length); + try (JarFile jar = new JarFile(files[0])) { + assertNotNull( + jar.getJarEntry( + "META-INF/services/org.apache.seatunnel.api.table.factory.Factory")); + assertNotNull(jar.getJarEntry("org/apache/http/impl/client/HttpClients.class")); + } + AtomicInteger requests = new AtomicInteger(); + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext( + "/", + exchange -> { + requests.incrementAndGet(); + String response; + if (exchange.getRequestURI().getPath().equals("/v1/oauth2/token")) { + assertTrue( + exchange.getRequestHeaders() + .getFirst("Authorization") + .startsWith("Basic ")); + response = + "{\"access_token\":\"packaged-token\",\"token_type\":\"Bearer\",\"expires_in\":3600}"; + } else { + assertEquals( + "Bearer packaged-token", + exchange.getRequestHeaders().getFirst("Authorization")); + response = PayPalResponseTest.page(1, 0, ""); + } + byte[] bytes = response.getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, bytes.length); + exchange.getResponseBody().write(bytes); + exchange.close(); + }); + server.start(); + try (URLClassLoader loader = + new URLClassLoader( + new URL[] {files[0].toURI().toURL()}, getClass().getClassLoader()) { + @Override + protected synchronized Class<?> loadClass(String name, boolean resolve) + throws ClassNotFoundException { + if (name.startsWith("org.apache.seatunnel.connectors.seatunnel.paypal.") + || name.startsWith("org.apache.http.")) { + Class<?> result = findLoadedClass(name); + if (result == null) { + result = findClass(name); + } + if (resolve) { + resolveClass(result); + } + return result; + } + return super.loadClass(name, resolve); + } + }) { + String prefix = "org.apache.seatunnel.connectors.seatunnel.paypal.source."; + Factory factory = + (Factory) + loader.loadClass(prefix + "PayPalSourceFactory") + .getConstructor() + .newInstance(); + assertEquals("PayPal", factory.factoryIdentifier()); + assertEquals(loader, factory.getClass().getClassLoader()); + Map<String, Object> options = PayPalResponseTest.options(); + options.put("mock_mode", true); + options.put("api_base_url", "http://127.0.0.1:"; + server.getAddress().getPort()); + ReadonlyConfig readonly = ReadonlyConfig.fromMap(options); + assertNotNull( + loader.loadClass(prefix + "PayPalSource") + .getConstructor(ReadonlyConfig.class) + .newInstance(readonly)); + Class<?> configClass = loader.loadClass(prefix + "PayPalConfig"); + Constructor<?> configConstructor = + configClass.getDeclaredConstructor(ReadonlyConfig.class); + configConstructor.setAccessible(true); + Object config = configConstructor.newInstance(readonly); + Class<?> clientClass = loader.loadClass(prefix + "PayPalClient"); + Constructor<?> clientConstructor = clientClass.getDeclaredConstructor(configClass); + clientConstructor.setAccessible(true); + try (AutoCloseable client = (AutoCloseable) clientConstructor.newInstance(config)) { + Method page = clientClass.getDeclaredMethod("page", int.class); + page.setAccessible(true); + assertNotNull(page.invoke(client, 1)); + assertEquals(2, requests.get()); + } + } finally { + server.stop(0); + } + } +} diff --git a/seatunnel-connectors-v2/connector-http/pom.xml b/seatunnel-connectors-v2/connector-http/pom.xml index 05f73a000c..4d33447da9 100644 --- a/seatunnel-connectors-v2/connector-http/pom.xml +++ b/seatunnel-connectors-v2/connector-http/pom.xml @@ -47,6 +47,7 @@ <module>connector-http-shopify</module> <module>connector-http-zendesk</module> <module>connector-http-stripe</module> + <module>connector-http-paypal</module> <module>connector-http-linear</module> <module>connector-mem0</module> </modules> diff --git a/seatunnel-dist/pom.xml b/seatunnel-dist/pom.xml index 13787f09cc..089cf82cc9 100644 --- a/seatunnel-dist/pom.xml +++ b/seatunnel-dist/pom.xml @@ -326,6 +326,12 @@ <version>${project.version}</version> <scope>provided</scope> </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-paypal</artifactId> + <version>${project.version}</version> + <scope>provided</scope> + </dependency> <dependency> <groupId>org.apache.seatunnel</groupId> <artifactId>connector-http-onesignal</artifactId> diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/pom.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/pom.xml new file mode 100644 index 0000000000..99d1b7fc4b --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/pom.xml @@ -0,0 +1,54 @@ +<?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 https://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-paypal-e2e</artifactId> + <name>SeaTunnel : E2E : Connector V2 : PayPal</name> + <dependencies> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>seatunnel-core-starter</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-paypal</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.mock-server</groupId> + <artifactId>mockserver-netty-no-dependencies</artifactId> + <version>5.14.0</version> + <scope>test</scope> + </dependency> + </dependencies> +</project> diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paypal/PayPalIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paypal/PayPalIT.java new file mode 100644 index 0000000000..95b27bb4ce --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paypal/PayPalIT.java @@ -0,0 +1,155 @@ +/* + * 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.paypal; + +import org.apache.seatunnel.e2e.common.TestResource; +import org.apache.seatunnel.e2e.common.TestSuiteBase; +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.TestTemplate; +import org.mockserver.client.MockServerClient; +import org.mockserver.matchers.Times; +import org.mockserver.model.HttpRequest; +import org.mockserver.model.HttpResponse; +import org.mockserver.verify.VerificationTimes; +import org.testcontainers.containers.Container; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.utility.DockerImageName; + +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import java.util.concurrent.TimeUnit; + +public class PayPalIT extends TestSuiteBase implements TestResource { + private GenericContainer<?> fixture; + private MockServerClient client; + + @BeforeAll + @Override + public void startUp() { + fixture = + new GenericContainer<>(DockerImageName.parse("mockserver/mockserver:5.14.0")) + .withNetwork(NETWORK) + .withNetworkAliases("paypal-fixture") + .withExposedPorts(1080) + .withEnv("MOCKSERVER_LOG_LEVEL", "WARN") + .waitingFor(Wait.forHttp("/").forStatusCode(404)); + fixture.start(); + client = new MockServerClient(fixture.getHost(), fixture.getMappedPort(1080)); + } + + @AfterAll + @Override + public void tearDown() { + try { + if (client != null) { + client.close(); + } + } finally { + try { + if (fixture != null) { + fixture.stop(); + } + } finally { + NETWORK.close(); + } + } + } + + @TestTemplate + public void readsAllRecordsThroughOAuthExpiryRefresh(TestContainer container) throws Exception { + client.reset(); + HttpRequest oauth = + HttpRequest.request() + .withMethod("POST") + .withPath("/v1/oauth2/token") + .withHeader( + "Authorization", + "Basic " + + Base64.getEncoder() + .encodeToString( + "mock-client:mock-secret" + .getBytes(StandardCharsets.UTF_8))) + .withBody("grant_type=client_credentials"); + client.when(oauth, Times.exactly(1)) + .respond( + HttpResponse.response() + .withHeader("Content-Type", "application/json") + .withBody( + "{\"access_token\":\"first-token\",\"token_type\":\"Bearer\",\"expires_in\":1}")); + client.when(oauth) + .respond( + HttpResponse.response() + .withHeader("Content-Type", "application/json") + .withBody( + "{\"access_token\":\"next-token\",\"token_type\":\"Bearer\",\"expires_in\":3600}")); + HttpRequest first = request(1, "first-token"); + HttpRequest last = request(2, "next-token"); + client.when(first) + .respond( + HttpResponse.response() + .withHeader("Content-Type", "application/json") + .withDelay(TimeUnit.MILLISECONDS, 1500) + .withBody(report(1, row("T0006") + "," + row("T1107")))); + client.when(last) + .respond( + HttpResponse.response() + .withHeader("Content-Type", "application/json") + .withBody(report(2, row("T0006")))); + Container.ExecResult result = container.executeJob("/paypal_to_assert.conf"); + Assertions.assertEquals(0, result.getExitCode(), result.getStderr()); + client.verify(oauth, VerificationTimes.exactly(2)); + client.verify(first, VerificationTimes.exactly(1)); + client.verify(last, VerificationTimes.exactly(1)); + client.verify( + HttpRequest.request().withPath("/v1/reporting/transactions"), + VerificationTimes.exactly(2)); + } + + private HttpRequest request(int page, String token) { + return HttpRequest.request() + .withMethod("GET") + .withPath("/v1/reporting/transactions") + .withHeader("Authorization", "Bearer " + token) + .withHeader("PayPal-Enforce-ISO8601-Format", "true") + .withQueryStringParameter("start_date", "2026-01-01T00:00:00Z") + .withQueryStringParameter("end_date", "2026-01-02T00:00:00Z") + .withQueryStringParameter("fields", "all") + .withQueryStringParameter("balance_affecting_records_only", "N") + .withQueryStringParameter("page_size", "2") + .withQueryStringParameter("page", Integer.toString(page)); + } + + private String report(int page, String records) { + return "{\"account_number\":\"ACCOUNT\",\"start_date\":\"2026-01-01T00:00:00Z\",\"end_date\":\"2026-01-02T00:00:00Z\",\"page\":" + + page + + ",\"total_pages\":2,\"total_items\":3,\"transaction_details\":[" + + records + + "]}"; + } + + private String row(String event) { + return "{\"transaction_info\":{\"transaction_id\":\"SAME-ID\",\"transaction_event_code\":\"" + + event + + "\",\"transaction_amount\":{\"value\":\"0.123\",\"currency_code\":\"TND\"},\"fee_amount\":null}}"; + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paypal/PayPalMaskingTest.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paypal/PayPalMaskingTest.java new file mode 100644 index 0000000000..bd750b2b52 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/src/test/java/org/apache/seatunnel/e2e/connector/paypal/PayPalMaskingTest.java @@ -0,0 +1,63 @@ +/* + * 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.paypal; + +import org.apache.seatunnel.shade.com.typesafe.config.Config; +import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory; + +import org.apache.seatunnel.core.starter.utils.ConfigBuilder; +import org.apache.seatunnel.core.starter.utils.ConfigShadeUtils; + +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class PayPalMaskingTest { + @Test + void nativeParsedConfigMasksClientSecretWithoutChangingSourceConfig() { + Config config = + ConfigFactory.parseString( + "source { PayPal { client_id = \"example-id\", client_secret = \"DO-NOT-LOG-PAYPAL\" } }"); + Object sources = config.root().unwrapped().get("source"); + assertTrue(sources instanceof List); + assertEquals(1, ((List<?>) sources).size()); + assertTrue(((List<?>) sources).get(0) instanceof Map); + assertTrue( + ConfigShadeUtils.getLogDesensitizationOptions(config).contains("client_secret"), + "Native mask rules missing from " + + ConfigShadeUtils.class + .getProtectionDomain() + .getCodeSource() + .getLocation()); + Map<String, Object> masked = + ConfigBuilder.configDesensitization( + config.root().unwrapped(), + ConfigShadeUtils.getLogDesensitizationOptions(config)); + assertFalse(masked.toString().contains("DO-NOT-LOG-PAYPAL")); + assertTrue( + config.getConfigList("source") + .get(0) + .getString("client_secret") + .equals("DO-NOT-LOG-PAYPAL")); + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/src/test/resources/paypal_to_assert.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/src/test/resources/paypal_to_assert.conf new file mode 100644 index 0000000000..fe040f0e2e --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paypal-e2e/src/test/resources/paypal_to_assert.conf @@ -0,0 +1,52 @@ +# +# 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" +} +source { + PayPal { + plugin_output = "transactions" + client_id = "mock-client" + client_secret = "mock-secret" + mock_mode = true + api_base_url = "http://paypal-fixture:1080"; + start_date = "2026-01-01T00:00:00Z" + end_date = "2026-01-02T00:00:00Z" + page_size = 2 + } +} +sink { + Assert { + plugin_input = "transactions" + rules { + row_rules = [ + {rule_type = MIN_ROW, rule_value = 3}, + {rule_type = MAX_ROW, rule_value = 3} + ] + field_rules = [ + {field_name = account_number, field_type = string, field_value = [{rule_type = NOT_NULL, equals_to = "ACCOUNT"}]}, + {field_name = transaction_id, field_type = string, field_value = [{rule_type = NOT_NULL, equals_to = "SAME-ID"}]}, + {field_name = transaction_amount, field_type = "decimal(38,9)", field_value = [{rule_type = NOT_NULL, equals_to = "0.123000000"}]}, + {field_name = transaction_currency, field_type = string, field_value = [{rule_type = NOT_NULL, equals_to = "TND"}]}, + {field_name = fee_amount, field_type = "decimal(38,9)", field_value = [{rule_type = NULL}]}, + {field_name = content, 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 2341cbe852..4dc825191b 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml @@ -51,6 +51,7 @@ <module>connector-cassandra-e2e</module> <module>connector-neo4j-e2e</module> <module>connector-http-e2e</module> + <module>connector-paypal-e2e</module> <module>connector-tiktok-ads-e2e</module> <module>connector-rabbitmq-e2e</module> <module>connector-kafka-e2e</module>
