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-12451-0f8913a8c3f30dc5bb8cab07bda91365a8811cb5 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 719c3e2b311fa6a050749984a43e6a837b9f802f Author: Goutam Adwant <[email protected]> AuthorDate: Sun Oct 11 09:46:08 2026 +0000 [Feature][Connector-V2] Add Sentry project error-events source (#12451) Signed-off-by: Goutam Adwant <[email protected]> --- .github/workflows/labeler/label-scope-conf.yml | 4 +- docs/en/connectors/source/Sentry.md | 129 ++++ docs/zh/connectors/source/Sentry.md | 124 ++++ plugin-mapping.properties | 1 + seatunnel-connectors-v2/connector-sentry/pom.xml | 15 + .../seatunnel/sentry/source/SentryClient.java | 270 ++++++++ .../seatunnel/sentry/source/SentryPage.java | 171 +++++ .../seatunnel/sentry/source/SentrySource.java | 99 +++ .../sentry/source/SentrySourceConfig.java | 177 ++++++ .../sentry/source/SentrySourceFactory.java | 70 +++ .../sentry/source/SentrySourceOptions.java | 96 +++ .../sentry/source/SentrySourceReader.java | 97 +++ .../seatunnel/sentry/SentryFactoryTest.java | 15 + .../seatunnel/sentry/source/SentrySourceTest.java | 685 +++++++++++++++++++++ seatunnel-dist/release-docs/LICENSE | 4 + seatunnel-dist/release-docs/NOTICE | 16 + .../connector-sentry-e2e/pom.xml | 52 ++ .../e2e/connector/sentry/SentryFixtureTest.java | 37 ++ .../seatunnel/e2e/connector/sentry/SentryIT.java | 130 ++++ .../e2e/connector/sentry/SentryMaskingTest.java | 101 +++ .../src/test/resources/logback-test.xml | 27 + .../src/test/resources/sentry_to_assert.conf | 53 ++ seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml | 1 + 23 files changed, 2373 insertions(+), 1 deletion(-) diff --git a/.github/workflows/labeler/label-scope-conf.yml b/.github/workflows/labeler/label-scope-conf.yml index 8a86ab89fd..4687fe725f 100644 --- a/.github/workflows/labeler/label-scope-conf.yml +++ b/.github/workflows/labeler/label-scope-conf.yml @@ -363,7 +363,9 @@ selectdb-cloud: sentry: - all: - changed-files: - - any-glob-to-any-file: seatunnel-connectors-v2/connector-sentry/** + - any-glob-to-any-file: + - seatunnel-connectors-v2/connector-sentry/** + - seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/** - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(sentry)/**' socket: - all: diff --git a/docs/en/connectors/source/Sentry.md b/docs/en/connectors/source/Sentry.md new file mode 100644 index 0000000000..c4fd461d9c --- /dev/null +++ b/docs/en/connectors/source/Sentry.md @@ -0,0 +1,129 @@ +import ChangeLog from '../changelog/connector-sentry.md'; + +# Sentry + +> Sentry project error-events source. + +## Description + +Reads one project's error events using `GET /api/0/projects/{organization}/{project}/events/`. +Use this source to analyze error events in an external warehouse. The existing Sentry sink and its DSN authentication are unchanged. + +## Key Features + +- [x] Batch +- [ ] Streaming +- [ ] Exactly-once + +Only BATCH mode and parallelism 1 are supported. The source reads an explicit time window. +The engine-injected failure policy must be `FAIL_FAST`; continuing other tables is not supported by this single-project source. +It does not discover projects, read traces or attachments, or create an immutable snapshot. + +## Prerequisites + +Create a bearer token with `project:read` access to the chosen project. A sink DSN cannot read events. +For a regional or self-hosted deployment, set its HTTPS origin in `api_base_url`. Redirects are rejected; configure the final API origin directly. +TLS certificate verification remains enabled. HTTP is supported only for fixtures with `mock_mode=true` and the literal dummy `token="mock-token"`. + +See the [events API](https://docs.sentry.io/api/events/list-a-projects-error-events/) and +[pagination contract](https://docs.sentry.io/api/pagination/). + +## Reading and Recovery Contract + +Provide absolute RFC3339 `start_time` and `end_time` with an offset, and start strictly before end. +Both are normalized to UTC and sent on every page. The source sends `full=false` and `sample=false`; +`content` contains the returned summary event, not a full stack trace. `page_size` maps to Sentry's `per_page` parameter. + +Pagination follows only the `rel="next"` link whose `results="true"`; a next link with `results="false"` completes the read. +Only its cursor is copied into the next request. The configured origin, endpoint and time bounds are never replaced by the server's link. +Missing/malformed pagination, repeated cursors, invalid rows, duplicate JSON object keys, oversized responses and exceeding `max_pages` fail the job instead of silently truncating it. + +HTTP 429, 500, 502, 503, 504 and transport failures are retried within `max_retries`. +`Retry-After` seconds and HTTP dates are respected. Delays exceeding 60 seconds fail so the job can be retried later; the connector never retries earlier than the server requests. +Authentication failures and redirects are not retried. Closing the reader aborts in-flight requests and wakes retry waits. +The timeout bounds HTTP I/O and schedules a per-attempt abort. Synchronous operating-system DNS resolution, JSON parsing and retry waits are not covered by a strict total job deadline. + +Recovery replays the entire configured window; cursors are not durable snapshot offsets and are not checkpointed. +The reader holds the checkpoint lock during the bounded read, so use manageable windows and suitable checkpoint timeouts. +Rows already written to a nontransactional sink can be duplicated after a failed attempt. Configure downstream deduplication if needed. +Sentry retention, delayed ingestion, event removal and changes while paginating can affect results. No complete-history, gap-free adjacent-window or snapshot-consistency guarantee is made. + +## Output Schema + +The schema is fixed; custom `schema` is not supported. All fields use STRING to preserve Sentry identifiers and timestamp representations. + +| Field | Nullable | Meaning | +| --- | --- | --- | +| event_id | no | `eventID` from the response | +| group_id | yes | `groupID` | +| project_id | yes | `projectID` | +| date_created | yes | `dateCreated`, unchanged | +| title | yes | Event title | +| message | yes | Event message | +| platform | yes | Event platform | +| content | no | Complete returned event object serialized as JSON | + +Missing optional fields become null. Unexpected types fail. JSON numeric values retain decimal precision, but `content` is not byte-identical to the HTTP response. +Event content can contain personal data: use access-controlled sinks and avoid logging rows. +The `token` option is masked by SeaTunnel's default parsed-configuration logging rules. + +## Options + +| Name | Type | Required | Default | Description | +| --- | --- | --- | --- | --- | +| token | STRING | yes | - | Bearer token with project read access | +| organization | STRING | yes | - | Organization ID or slug | +| project | STRING | yes | - | Project ID or slug | +| start_time | STRING | yes | - | Absolute RFC3339 start | +| end_time | STRING | yes | - | Absolute RFC3339 end | +| api_base_url | STRING | no | https://sentry.io | HTTPS origin without a path, query, fragment or trailing slash | +| page_size | INT | no | 100 | Events requested per page, 1–100 | +| max_pages | INT | no | 10000 | Page safety limit, 1–100000; fails when exceeded | +| max_retries | INT | no | 3 | Additional transient attempts, 0–5 | +| retry_delay_ms | INT | no | 1000 | Minimum retry delay, 1–60000 ms | +| request_timeout_ms | INT | no | 30000 | Per-attempt HTTP connection/socket timeout and abort deadline, 1–120000 ms | +| max_response_bytes | INT | no | 8388608 | Uncompressed body limit, 1024–16777216 bytes | +| mock_mode | BOOLEAN | no | false | Permit an HTTP fixture with dummy token only | + +Source [common options](../common-options/source-common-options.md) such as `plugin_output` remain available. +Project names accept letters, digits, underscores and hyphens. `query`, `full`, `sample`, `dsn` and a custom schema are not source options. + +## Example + +Set the `SENTRY_TOKEN` and `WAREHOUSE_PASSWORD` environment variables before starting the job. The unquoted HOCON substitutions below resolve their values when the configuration is loaded. + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} +source { + Sentry { + plugin_output = "events" + token = ${SENTRY_TOKEN} + organization = "example" + project = "backend" + start_time = "2026-01-01T00:00:00Z" + end_time = "2026-01-02T00:00:00Z" + } +} +sink { + Jdbc { + plugin_input = "events" + url = "jdbc:postgresql://warehouse:5432/analytics" + driver = "org.postgresql.Driver" + user = "writer" + password = ${WAREHOUSE_PASSWORD} + database = "analytics" + table = "sentry_events" + generate_sink_sql = true + } +} +``` + +Create the destination table with the eight STRING-compatible columns above and install its JDBC driver. +The integration test runs an engine job against a deterministic Sentry API fixture; it does not establish compatibility with every Sentry cloud or self-hosted release. + +## Changelog + +<ChangeLog /> diff --git a/docs/zh/connectors/source/Sentry.md b/docs/zh/connectors/source/Sentry.md new file mode 100644 index 0000000000..6d98355c04 --- /dev/null +++ b/docs/zh/connectors/source/Sentry.md @@ -0,0 +1,124 @@ +import ChangeLog from '../changelog/connector-sentry.md'; + +# Sentry + +> Sentry 项目错误事件 Source 连接器。 + +## 描述 + +通过 `GET /api/0/projects/{organization}/{project}/events/` 读取单个项目的错误事件，用于外部数据仓库分析。现有 Sentry Sink 和 DSN 认证保持不变。 + +## 主要特性 + +- [x] 批处理 +- [ ] 流处理 +- [ ] 精确一次 + +仅支持 BATCH 模式和并行度 1，读取显式指定的时间范围。不支持项目发现、追踪、附件或一致性快照。 +引擎注入的失败策略必须为 `FAIL_FAST`，单项目 Source 不支持继续处理其他表的策略。 + +## 前置条件 + +使用具有目标项目 `project:read` 权限的 Bearer Token，Sink DSN 不能用于读取事件。 +区域服务或自托管部署通过 `api_base_url` 指定 HTTPS 源站，不能包含路径、查询参数、片段或末尾斜杠。 +不跟随重定向，请直接配置最终 API 地址。默认验证 TLS 证书。 +仅在 `mock_mode=true` 且 `token="mock-token"` 时允许 HTTP 测试服务，不得使用真实凭据。 + +参考 [事件 API](https://docs.sentry.io/api/events/list-a-projects-error-events/) 和 [分页协议](https://docs.sentry.io/api/pagination/)。 + +## 读取与恢复语义 + +`start_time` 和 `end_time` 必须是含时区偏移的绝对 RFC3339 时间，且开始早于结束。每页请求均携带规范化为 UTC 的时间范围。 +固定发送 `full=false` 和 `sample=false`；`content` 是 API 返回的摘要事件，不包含完整堆栈。 `page_size` 对应 `per_page`。 + +仅在 Link 响应头中 `rel="next"` 且 `results="true"` 时读取下一页，`results="false"` 表示结束。 +只提取游标，下一页始终使用已配置的源站、路径和时间范围。 +缺失或非法分页头、循环游标、非法记录、重复 JSON 键、响应超限或超过 `max_pages` 均使作业失败，不会静默截断。 + +HTTP 429、500、502、503、504 和传输失败会进行有界重试。支持以秒数或 HTTP 日期表示的 `Retry-After`。 +超过 60 秒的服务端等待要求会使作业失败，留待稍后重试，不会提前请求。认证失败和重定向不重试。 +关闭 Reader 会中止正在执行的请求并唤醒重试等待。 +超时参数控制 HTTP I/O 并为每次尝试安排中止。操作系统同步 DNS 解析、JSON 解析和重试等待不受严格的作业总时限约束。 + +恢复时重新读取整个时间范围，不对远端游标做检查点恢复。读取期间持有检查点锁，因此应选择合理时间范围和检查点超时。 +失败前已写入非事务 Sink 的记录可能在重试后重复，必要时在下游去重。 +数据保留策略、延迟写入、事件删除及分页期间数据变更可能影响结果，不保证完整历史、相邻范围无遗漏或快照一致性。 + +## 输出结构 + +固定结构，不支持自定义 `schema`。所有列均为 STRING，以保留标识符和时间戳表达。 + +| 列 | 可为空 | 含义 | +| --- | --- | --- | +| event_id | 否 | API 的 `eventID` | +| group_id | 是 | `groupID` | +| project_id | 是 | `projectID` | +| date_created | 是 | 原始 `dateCreated` | +| title | 是 | 事件标题 | +| message | 是 | 事件消息 | +| platform | 是 | 事件平台 | +| content | 否 | 完整返回事件对象序列化后的 JSON | + +缺少可选字段时输出 null，类型不符则失败。JSON 数值保留十进制精度，但 `content` 不保证与响应逐字节一致。 +事件可能包含个人数据，请限制目标存储访问并避免记录行内容。默认配置日志脱敏规则会隐藏 `token`。 + +## 参数 + +| 参数 | 类型 | 必填 | 默认值 | 描述 | +| --- | --- | --- | --- | --- | +| token | STRING | 是 | - | 具有项目读取权限的 Bearer Token | +| organization | STRING | 是 | - | 组织 ID 或名称 | +| project | STRING | 是 | - | 项目 ID 或名称 | +| start_time | STRING | 是 | - | 绝对 RFC3339 开始时间 | +| end_time | STRING | 是 | - | 绝对 RFC3339 结束时间 | +| api_base_url | STRING | 否 | https://sentry.io | HTTPS 源站，不含路径及末尾斜杠 | +| page_size | INT | 否 | 100 | 每页请求记录数，1–100 | +| max_pages | INT | 否 | 10000 | 页数上限，1–100000；超限失败 | +| max_retries | INT | 否 | 3 | 额外重试次数，0–5 | +| retry_delay_ms | INT | 否 | 1000 | 最小重试间隔，1–60000 毫秒 | +| request_timeout_ms | INT | 否 | 30000 | 每次 HTTP 连接/读取超时及中止期限，1–120000 毫秒 | +| max_response_bytes | INT | 否 | 8388608 | 解压后响应字节上限，1024–16777216 | +| mock_mode | BOOLEAN | 否 | false | 仅允许使用假 Token 的 HTTP 测试服务 | + +支持 `plugin_output` 等 [Source 通用参数](../common-options/source-common-options.md)。 +组织和项目名称接受字母、数字、下划线和连字符。Source 不支持 `query`、`full`、`sample`、`dsn` 或自定义 schema。 + +## 示例 + +运行前请设置 `SENTRY_TOKEN` 和 `WAREHOUSE_PASSWORD` 环境变量。以下未加引号的 HOCON 替换表达式将在加载配置时解析为对应值。 + +```hocon +env { + parallelism = 1 + job.mode = "BATCH" +} +source { + Sentry { + plugin_output = "events" + token = ${SENTRY_TOKEN} + organization = "example" + project = "backend" + start_time = "2026-01-01T00:00:00Z" + end_time = "2026-01-02T00:00:00Z" + } +} +sink { + Jdbc { + plugin_input = "events" + url = "jdbc:postgresql://warehouse:5432/analytics" + driver = "org.postgresql.Driver" + user = "writer" + password = ${WAREHOUSE_PASSWORD} + database = "analytics" + table = "sentry_events" + generate_sink_sql = true + } +} +``` + +请按上述八个字段创建兼容 STRING 类型的目标表并安装 JDBC 驱动。 +集成测试使用确定性的 Sentry API 模拟服务运行引擎作业，不代表已验证所有 Sentry 云服务或自托管版本。 + +## 更新日志 + +<ChangeLog /> diff --git a/plugin-mapping.properties b/plugin-mapping.properties index a3fa2cc514..65c23b908c 100644 --- a/plugin-mapping.properties +++ b/plugin-mapping.properties @@ -80,6 +80,7 @@ seatunnel.sink.Databend = connector-databend seatunnel.source.Databend = connector-databend seatunnel.sink.DataHub = connector-datahub seatunnel.sink.Sentry = connector-sentry +seatunnel.source.Sentry = connector-sentry seatunnel.source.MongoDB = connector-mongodb seatunnel.sink.MongoDB = connector-mongodb seatunnel.sink.Couchbase = connector-couchbase diff --git a/seatunnel-connectors-v2/connector-sentry/pom.xml b/seatunnel-connectors-v2/connector-sentry/pom.xml index eb691200f4..500f17832d 100644 --- a/seatunnel-connectors-v2/connector-sentry/pom.xml +++ b/seatunnel-connectors-v2/connector-sentry/pom.xml @@ -34,6 +34,21 @@ </properties> <dependencies> + <dependency> + <groupId>org.awaitility</groupId> + <artifactId>awaitility</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.httpcomponents</groupId> + <artifactId>httpclient</artifactId> + <version>4.5.13</version> + </dependency> + <dependency> + <groupId>org.apache.httpcomponents</groupId> + <artifactId>httpcore</artifactId> + <version>4.4.16</version> + </dependency> <dependency> <groupId>org.apache.seatunnel</groupId> <artifactId>seatunnel-api</artifactId> diff --git a/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentryClient.java b/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentryClient.java new file mode 100644 index 0000000000..49f6395a96 --- /dev/null +++ b/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentryClient.java @@ -0,0 +1,270 @@ +/* + * 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.sentry.source; + +import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated; +import org.apache.seatunnel.connectors.seatunnel.sentry.exception.SentryConnectorException; + +import org.apache.http.Header; +import org.apache.http.client.config.RequestConfig; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.client.utils.URIBuilder; +import org.apache.http.config.ConnectionConfig; +import org.apache.http.config.MessageConstraints; +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.time.Duration; +import java.time.Instant; +import java.time.ZonedDateTime; +import java.time.format.DateTimeFormatter; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +/** A cancellable, size-bounded client for Sentry's header-based pagination contract. */ +final class SentryClient implements AutoCloseable { + static SentryConnectorException failure(String message) { + return new SentryConnectorException( + CommonErrorCodeDeprecated.HTTP_OPERATION_FAILED, message); + } + + private final SentrySourceConfig config; + private final CloseableHttpClient client; + private final ScheduledThreadPoolExecutor timer; + private volatile boolean closed; + private HttpGet active; + + SentryClient(SentrySourceConfig config) { + this.config = config; + client = + HttpClients.custom() + .setDefaultConnectionConfig( + ConnectionConfig.custom() + .setMessageConstraints( + MessageConstraints.custom() + .setMaxLineLength(16384) + .setMaxHeaderCount(64) + .build()) + .build()) + .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, "sentry-source-request-deadline"); + thread.setDaemon(true); + return thread; + }); + timer.setRemoveOnCancelPolicy(true); + } + + SentryPage page(String cursor) throws Exception { + for (int attempt = 0; ; attempt++) { + checkOpen(); + URIBuilder uri = + new URIBuilder(config.endpoint) + .addParameter("start", config.start.toString()) + .addParameter("end", config.end.toString()) + .addParameter("per_page", Integer.toString(config.pageSize)) + .addParameter("full", "false") + .addParameter("sample", "false"); + if (cursor != null) { + uri.addParameter("cursor", cursor); + } + HttpGet request = new HttpGet(uri.build()); + request.setHeader("Authorization", "Bearer " + config.token); + request.setHeader("Accept", "application/json"); + Reply reply; + try { + reply = execute(request); + } catch (IOException e) { + retry(attempt, config.retryDelay); + continue; + } + if (transientStatus(reply.status)) { + retry(attempt, reply.delay); + continue; + } + if (reply.status != 200) { + throw failure( + "Sentry HTTP " + + reply.status + + "; check endpoint, project and project:read permission (response withheld)"); + } + return new SentryPage(reply.body, reply.link, config.endpoint); + } + } + + private Reply execute(HttpGet request) throws IOException { + ScheduledFuture<?> deadline; + synchronized (this) { + checkOpen(); + active = request; + deadline = timer.schedule(request::abort, config.timeout, TimeUnit.MILLISECONDS); + } + CloseableHttpResponse response = null; + try { + response = client.execute(request); + int status = response.getStatusLine().getStatusCode(); + if (status != 200) { + int delay = config.retryDelay; + if (transientStatus(status) && response.getFirstHeader("Retry-After") != null) { + delay = + retryDelay( + response.getFirstHeader("Retry-After").getValue(), + delay, + Instant.now()); + } + return new Reply(status, null, null, delay); + } + Header[] headers = response.getHeaders("Link"); + StringBuilder links = new StringBuilder(); + for (Header header : headers) { + if (links.length() + header.getValue().length() + 2 > 16384) { + throw failure("Oversized Sentry Link header"); + } + if (links.length() > 0) { + links.append(", "); + } + links.append(header.getValue()); + } + if (response.getEntity() == null) { + throw failure("Missing Sentry response body"); + } + if (response.getEntity().getContentLength() > config.maxBytes) { + throw failure("Sentry 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) { + throw failure("Sentry response exceeds max_response_bytes"); + } + bytes.write(buffer, 0, count); + } + return new Reply( + status, bytes.toByteArray(), headers.length == 0 ? null : links.toString(), 0); + } finally { + request.abort(); + deadline.cancel(false); + if (response != null) { + try { + response.close(); + } catch (IOException ignored) { + // Request already aborted; never attach transport details containing + // credentials. + } + } + synchronized (this) { + active = null; + } + } + } + + static int retryDelay(String value, int minimum, Instant now) { + long millis; + try { + if (value.matches("[0-9]+")) { + millis = Math.multiplyExact(Long.parseLong(value), 1000L); + } else { + millis = + Math.max( + 0, + Duration.between( + now, + ZonedDateTime.parse( + value, + DateTimeFormatter + .RFC_1123_DATE_TIME) + .toInstant()) + .toMillis()); + } + if (millis > 60000) { + throw new IllegalArgumentException(); + } + } catch (RuntimeException e) { + throw failure("Retry-After exceeds 60 seconds or is invalid; retry the job later"); + } + return Math.max(minimum, (int) millis); + } + + 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 failure("Sentry 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 failure("Sentry request cancelled"); + } + } + + @Override + public synchronized void close() throws IOException { + closed = true; + if (active != null) { + active.abort(); + } + notifyAll(); + timer.shutdownNow(); + client.close(); + } + + private static final class Reply { + final int status; + final byte[] body; + final String link; + final int delay; + + Reply(int status, byte[] body, String link, int delay) { + this.status = status; + this.body = body; + this.link = link; + this.delay = delay; + } + } +} diff --git a/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentryPage.java b/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentryPage.java new file mode 100644 index 0000000000..86ba363eab --- /dev/null +++ b/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentryPage.java @@ -0,0 +1,171 @@ +/* + * 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.sentry.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 org.apache.http.NameValuePair; +import org.apache.http.client.utils.URLEncodedUtils; + +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +final class SentryPage { + private static final ObjectMapper MAPPER = + new ObjectMapper() + .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS) + .enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS) + .enable(JsonParser.Feature.STRICT_DUPLICATE_DETECTION); + private static final Pattern LINK = Pattern.compile("^\\s*<([^>]+)>\\s*(.*)$"); + private static final Pattern PARAMETER = + Pattern.compile("\\s*;\\s*([A-Za-z][A-Za-z0-9_-]*)\\s*=\\s*\\\"([^\\\"]*)\\\"\\s*"); + final List<SeaTunnelRow> rows; + final String nextCursor; + + SentryPage(byte[] body, String link, String endpoint) { + nextCursor = nextCursor(link, endpoint); + JsonNode root; + try { + root = MAPPER.readTree(body); + } catch (Exception e) { + throw SentryClient.failure("Invalid JSON event response (body withheld)"); + } + if (root == null || !root.isArray()) { + throw SentryClient.failure("Expected a JSON array of Sentry events"); + } + rows = new ArrayList<>(root.size()); + for (JsonNode event : root) { + if (!event.isObject()) { + throw SentryClient.failure("Expected an event object"); + } + rows.add( + new SeaTunnelRow( + new Object[] { + text(event, "eventID", true), text(event, "groupID", false), + text(event, "projectID", false), + text(event, "dateCreated", false), text(event, "title", false), + text(event, "message", false), + text(event, "platform", false), event.toString() + })); + } + } + + private static String text(JsonNode event, String field, boolean required) { + JsonNode value = event.get(field); + if (value == null || value.isNull()) { + if (required) { + throw SentryClient.failure("Missing event field " + field); + } + return null; + } + if (!value.isTextual() || (required && value.textValue().isEmpty())) { + throw SentryClient.failure("Invalid event field " + field + " (value withheld)"); + } + return value.textValue(); + } + + /** Use only the cursor, never a server-provided URL or rewritten query bounds. */ + static String nextCursor(String header, String endpoint) { + if (header == null || header.length() > 16384) { + throw SentryClient.failure("Missing or oversized Sentry Link header"); + } + boolean found = false; + String next = null; + for (String part : header.split(",(?=\\s*<)")) { + Matcher link = LINK.matcher(part); + if (!link.matches()) { + throw SentryClient.failure("Malformed Sentry Link header"); + } + Map<String, String> parameters = new HashMap<>(); + Matcher parameter = PARAMETER.matcher(link.group(2)); + int position = 0; + while (parameter.find()) { + if (parameter.start() != position + || parameters.put(parameter.group(1), parameter.group(2)) != null) { + throw SentryClient.failure("Malformed Sentry Link parameters"); + } + position = parameter.end(); + } + if (position != link.group(2).length()) { + throw SentryClient.failure("Malformed Sentry Link parameters"); + } + if (!"next".equals(parameters.get("rel"))) { + continue; + } + if (found) { + throw SentryClient.failure("Duplicate next-page links"); + } + found = true; + String results = parameters.get("results"); + if (!"true".equals(results) && !"false".equals(results)) { + throw SentryClient.failure("Missing or invalid next-page results flag"); + } + URI uri; + try { + uri = URI.create(link.group(1)); + } catch (RuntimeException e) { + throw SentryClient.failure("Invalid next-page URI"); + } + URI expected = URI.create(endpoint); + if (!expected.getScheme().equals(uri.getScheme()) + || !expected.getRawAuthority().equals(uri.getRawAuthority()) + || !expected.getRawPath().equals(uri.getRawPath()) + || uri.getRawFragment() != null) { + throw SentryClient.failure("Next-page URI changed Sentry origin or endpoint"); + } + if ("false".equals(results)) { + continue; + } + String cursor = null; + boolean cursorFound = false; + for (NameValuePair query : URLEncodedUtils.parse(uri, StandardCharsets.UTF_8)) { + if ("cursor".equals(query.getName())) { + if (cursorFound) { + throw SentryClient.failure("Duplicate next-page cursor"); + } + cursorFound = true; + cursor = query.getValue(); + } + } + if (cursor == null + || cursor.isEmpty() + || cursor.length() > 2048 + || cursor.chars().anyMatch(c -> c < 33 || c > 126) + || (parameters.containsKey("cursor") + && !cursor.equals(parameters.get("cursor")))) { + throw SentryClient.failure("Missing or invalid next-page cursor"); + } + next = cursor; + } + if (!found) { + throw SentryClient.failure("Missing next-page relation in Sentry Link header"); + } + return next; + } +} diff --git a/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySource.java b/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySource.java new file mode 100644 index 0000000000..a0a4b748ac --- /dev/null +++ b/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySource.java @@ -0,0 +1,99 @@ +/* + * 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.sentry.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.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; + +/** Bounded error-event extraction for one project, without remote snapshot guarantees. */ +public class SentrySource extends AbstractSingleSplitSource<SeaTunnelRow> { + private final SentrySourceConfig config; + + public SentrySource(ReadonlyConfig options) { + config = new SentrySourceConfig(options); + } + + @Override + public String getPluginName() { + return "Sentry"; + } + + @Override + public Boundedness getBoundedness() { + return Boundedness.BOUNDED; + } + + @Override + public void setJobContext(JobContext context) { + if (context.getJobMode() != JobMode.BATCH) { + throw SentrySourceConfig.failure("Sentry source supports BATCH only"); + } + } + + @Override + public List<CatalogTable> getProducedCatalogTables() { + TableSchema.Builder schema = TableSchema.builder(); + String[] fields = { + "event_id", + "group_id", + "project_id", + "date_created", + "title", + "message", + "platform", + "content" + }; + for (String name : fields) { + schema.column( + PhysicalColumn.of( + name, + BasicType.STRING_TYPE, + 0, + !name.equals("event_id") && !name.equals("content"), + null, + null)); + } + return Collections.singletonList( + CatalogTable.of( + TableIdentifier.of("Sentry", TablePath.DEFAULT), + schema.build(), + Collections.emptyMap(), + Collections.emptyList(), + null)); + } + + @Override + public AbstractSingleSplitReader<SeaTunnelRow> createReader(SingleSplitReaderContext context) { + return new SentrySourceReader(config, context); + } +} diff --git a/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySourceConfig.java b/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySourceConfig.java new file mode 100644 index 0000000000..2c82881e6a --- /dev/null +++ b/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySourceConfig.java @@ -0,0 +1,177 @@ +/* + * 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.sentry.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 org.apache.seatunnel.api.options.MultiTableCommonOptions; +import org.apache.seatunnel.api.options.MultiTableFailurePolicy; +import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated; +import org.apache.seatunnel.connectors.seatunnel.sentry.exception.SentryConnectorException; + +import java.io.Serializable; +import java.net.URI; +import java.time.Instant; +import java.time.OffsetDateTime; +import java.util.HashMap; +import java.util.Map; + +final class SentrySourceConfig implements Serializable { + private static final long serialVersionUID = 1L; + final String token; + final String endpoint; + final Instant start; + final Instant end; + final int pageSize; + final int maxPages; + final int retries; + final int retryDelay; + final int timeout; + final int maxBytes; + + SentrySourceConfig(ReadonlyConfig options) { + // Zeta injects this engine-owned policy even for single-table sources. + if (get(options, MultiTableCommonOptions.MULTI_TABLE_FAILURE_POLICY) + != MultiTableFailurePolicy.FAIL_FAST) { + throw failure("Sentry source supports only multi_table.failure_policy=FAIL_FAST"); + } + Map<String, Object> validation = new HashMap<>(options.getSourceMap()); + validation.remove(MultiTableCommonOptions.MULTI_TABLE_FAILURE_POLICY.key()); + if (validation.get("multi_table") instanceof Map) { + Map<String, Object> nested = + new HashMap<>((Map<String, Object>) validation.get("multi_table")); + nested.remove("failure_policy"); + if (nested.isEmpty()) { + validation.remove("multi_table"); + } else { + validation.put("multi_table", nested); + } + } + ConfigValidator.validateUnknownKeys( + ReadonlyConfig.fromMap(validation), + new SentrySourceFactory().optionRule(), + "Sentry"); + if (options.getSourceMap().containsKey("schema")) { + throw failure("Sentry source uses a fixed schema"); + } + if (get(options, EnvCommonOptions.PARALLELISM) != 1) { + throw failure("Sentry source requires parallelism=1"); + } + token = get(options, SentrySourceOptions.TOKEN); + if (token == null || token.length() > 8192 || !token.matches("[A-Za-z0-9._~+/-]+=*")) { + throw failure("Invalid token (value withheld)"); + } + boolean mock = get(options, SentrySourceOptions.MOCK_MODE); + if (mock && !token.equals("mock-token")) { + throw failure("mock_mode requires token=mock-token"); + } + String origin = origin(get(options, SentrySourceOptions.API_BASE_URL), mock); + endpoint = + origin + + "/api/0/projects/" + + slug(get(options, SentrySourceOptions.ORGANIZATION), "organization") + + "/" + + slug(get(options, SentrySourceOptions.PROJECT), "project") + + "/events/"; + start = date(get(options, SentrySourceOptions.START_TIME)); + end = date(get(options, SentrySourceOptions.END_TIME)); + if (!start.isBefore(end)) { + throw failure("start_time must be earlier than end_time"); + } + pageSize = range(get(options, SentrySourceOptions.PAGE_SIZE), 1, 100, "page_size"); + maxPages = range(get(options, SentrySourceOptions.MAX_PAGES), 1, 100000, "max_pages"); + retries = range(get(options, SentrySourceOptions.MAX_RETRIES), 0, 5, "max_retries"); + retryDelay = + range(get(options, SentrySourceOptions.RETRY_DELAY_MS), 1, 60000, "retry_delay_ms"); + timeout = + range( + get(options, SentrySourceOptions.REQUEST_TIMEOUT_MS), + 1, + 120000, + "request_timeout_ms"); + maxBytes = + range( + get(options, SentrySourceOptions.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 failure("Invalid option " + option.key() + " (value withheld)"); + } + } + + private static String origin(String value, boolean mock) { + try { + URI uri = URI.create(value); + if (uri.getHost() == null + || uri.getPort() == 0 + || uri.getPort() > 65535 + || uri.getRawUserInfo() != null + || uri.getRawQuery() != null + || uri.getRawFragment() != null + || !uri.getRawPath().isEmpty() + || !("https".equals(uri.getScheme()) + || (mock && "http".equals(uri.getScheme())))) { + throw new IllegalArgumentException(); + } + return value; + } catch (RuntimeException e) { + throw failure( + "api_base_url must be an HTTPS origin without user info, path, query or fragment"); + } + } + + private static String slug(String value, String name) { + if (value == null || !value.matches("[A-Za-z0-9_-]{1,200}")) { + throw failure("Invalid " + name + " ID or slug"); + } + return value; + } + + private 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).toInstant(); + } catch (RuntimeException e) { + throw failure( + "start_time and end_time must be absolute RFC3339 timestamps with offsets"); + } + } + + private static int range(int value, int min, int max, String name) { + if (value < min || value > max) { + throw failure(name + " must be between " + min + " and " + max); + } + return value; + } + + static SentryConnectorException failure(String message) { + return new SentryConnectorException(CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT, message); + } +} diff --git a/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySourceFactory.java b/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySourceFactory.java new file mode 100644 index 0000000000..da80a39a7f --- /dev/null +++ b/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySourceFactory.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.sentry.source; + +import org.apache.seatunnel.api.configuration.util.OptionRule; +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 SentrySourceFactory implements TableSourceFactory { + @Override + public String factoryIdentifier() { + return "Sentry"; + } + + @Override + public OptionRule optionRule() { + return OptionRule.builder() + .required( + SentrySourceOptions.TOKEN, + SentrySourceOptions.ORGANIZATION, + SentrySourceOptions.PROJECT, + SentrySourceOptions.START_TIME, + SentrySourceOptions.END_TIME) + .optional( + SentrySourceOptions.API_BASE_URL, + SentrySourceOptions.PAGE_SIZE, + SentrySourceOptions.MAX_PAGES, + SentrySourceOptions.MAX_RETRIES, + SentrySourceOptions.RETRY_DELAY_MS, + SentrySourceOptions.REQUEST_TIMEOUT_MS, + SentrySourceOptions.MAX_RESPONSE_BYTES, + SentrySourceOptions.MOCK_MODE) + .build(); + } + + @Override + public <T, SplitT extends SourceSplit, StateT extends Serializable> + TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) { + return () -> (SeaTunnelSource<T, SplitT, StateT>) new SentrySource(context.getOptions()); + } + + @Override + public Class<? extends SeaTunnelSource> getSourceClass() { + return SentrySource.class; + } +} diff --git a/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySourceOptions.java b/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySourceOptions.java new file mode 100644 index 0000000000..7eb5c3b808 --- /dev/null +++ b/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySourceOptions.java @@ -0,0 +1,96 @@ +/* + * 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.sentry.source; + +import org.apache.seatunnel.api.configuration.Option; +import org.apache.seatunnel.api.configuration.Options; + +public final class SentrySourceOptions { + public static final Option<String> TOKEN = + Options.key("token") + .stringType() + .noDefaultValue() + .withDescription("Bearer token with project:read access; not a sink DSN."); + public static final Option<String> ORGANIZATION = + Options.key("organization") + .stringType() + .noDefaultValue() + .withDescription("Sentry organization ID or slug."); + public static final Option<String> PROJECT = + Options.key("project") + .stringType() + .noDefaultValue() + .withDescription("Sentry project ID or slug."); + public static final Option<String> START_TIME = + Options.key("start_time") + .stringType() + .noDefaultValue() + .withDescription("Absolute RFC3339 start of the event query."); + public static final Option<String> END_TIME = + Options.key("end_time") + .stringType() + .noDefaultValue() + .withDescription("Absolute RFC3339 end of the event query."); + public static final Option<String> API_BASE_URL = + Options.key("api_base_url") + .stringType() + .defaultValue("https://sentry.io";) + .withDescription( + "HTTPS origin of Sentry or a self-hosted deployment; no path."); + public static final Option<Integer> PAGE_SIZE = + Options.key("page_size") + .intType() + .defaultValue(100) + .withDescription("Number of events requested per page, 1 to 100."); + public static final Option<Integer> MAX_PAGES = + Options.key("max_pages") + .intType() + .defaultValue(10000) + .withDescription( + "Fail rather than truncate if the query requires more pages, 1 to 100000."); + public static final Option<Integer> MAX_RETRIES = + Options.key("max_retries") + .intType() + .defaultValue(3) + .withDescription( + "Additional attempts for transport failures and HTTP 429/500/502/503/504, 0 to 5."); + public static final Option<Integer> RETRY_DELAY_MS = + Options.key("retry_delay_ms") + .intType() + .defaultValue(1000) + .withDescription("Minimum retry delay, 1 to 60000 milliseconds."); + public static final Option<Integer> REQUEST_TIMEOUT_MS = + Options.key("request_timeout_ms") + .intType() + .defaultValue(30000) + .withDescription( + "Per-attempt HTTP connect/socket timeout and abort deadline, 1 to 120000 milliseconds; not a total job deadline."); + public static final Option<Integer> MAX_RESPONSE_BYTES = + Options.key("max_response_bytes") + .intType() + .defaultValue(8388608) + .withDescription("Maximum uncompressed response size, 1024 to 16777216 bytes."); + public static final Option<Boolean> MOCK_MODE = + Options.key("mock_mode") + .booleanType() + .defaultValue(false) + .withDescription( + "Allow HTTP fixtures only with token=mock-token; never use with real credentials."); + + private SentrySourceOptions() {} +} diff --git a/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySourceReader.java b/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySourceReader.java new file mode 100644 index 0000000000..cf1f37a022 --- /dev/null +++ b/seatunnel-connectors-v2/connector-sentry/src/main/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySourceReader.java @@ -0,0 +1,97 @@ +/* + * 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.sentry.source; + +import org.apache.seatunnel.api.source.Collector; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated; +import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader; +import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext; +import org.apache.seatunnel.connectors.seatunnel.sentry.exception.SentryConnectorException; + +import java.io.IOException; +import java.util.HashSet; +import java.util.Set; + +final class SentrySourceReader extends AbstractSingleSplitReader<SeaTunnelRow> { + static SentryConnectorException failure(String message) { + return new SentryConnectorException( + CommonErrorCodeDeprecated.READER_OPERATION_FAILED, message); + } + + private final SentrySourceConfig config; + private final SingleSplitReaderContext context; + private volatile boolean closed; + private SentryClient client; + + SentrySourceReader(SentrySourceConfig config, SingleSplitReaderContext context) { + this.config = config; + this.context = context; + } + + /** Create HTTP resources on the worker, not inside serialized configuration or split state. */ + @Override + public synchronized void open() { + if (closed || client != null) { + throw failure("Reader closed or already opened"); + } + client = new SentryClient(config); + } + + /** Read a complete bounded window. Recovery replays the window, not unstable remote offsets. */ + @Override + public void internalPollNext(Collector<SeaTunnelRow> output) throws Exception { + Set<String> visited = new HashSet<>(); + String cursor = null; + for (int page = 0; ; page++) { + checkOpen(); + if (page >= config.maxPages) { + throw failure("Sentry query exceeds max_pages; use a smaller time window"); + } + SentryPage result = client.page(cursor); + if (result.nextCursor != null && !visited.add(result.nextCursor)) { + throw failure("Sentry returned a repeated pagination cursor"); + } + for (SeaTunnelRow row : result.rows) { + checkOpen(); + output.collect(row); + } + cursor = result.nextCursor; + if (cursor == null) { + break; + } + } + checkOpen(); + context.signalNoMoreElement(); + } + + private void checkOpen() { + if (closed || Thread.currentThread().isInterrupted() || client == null) { + throw failure("Sentry reader is not open or was cancelled"); + } + } + + /** Abort in-flight requests, wake retry waits, and release all worker-local resources. */ + @Override + public synchronized void close() throws IOException { + closed = true; + if (client != null) { + client.close(); + } + } +} diff --git a/seatunnel-connectors-v2/connector-sentry/src/test/java/org/apache/seatunnel/connectors/seatunnel/sentry/SentryFactoryTest.java b/seatunnel-connectors-v2/connector-sentry/src/test/java/org/apache/seatunnel/connectors/seatunnel/sentry/SentryFactoryTest.java index 3bb59eb44c..316edc5fde 100644 --- a/seatunnel-connectors-v2/connector-sentry/src/test/java/org/apache/seatunnel/connectors/seatunnel/sentry/SentryFactoryTest.java +++ b/seatunnel-connectors-v2/connector-sentry/src/test/java/org/apache/seatunnel/connectors/seatunnel/sentry/SentryFactoryTest.java @@ -21,6 +21,8 @@ import org.apache.seatunnel.api.configuration.ReadonlyConfig; import org.apache.seatunnel.api.configuration.util.ConfigValidator; import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.api.configuration.util.OptionValidationException; +import org.apache.seatunnel.api.table.factory.Factory; +import org.apache.seatunnel.api.table.factory.TableSourceFactory; import org.apache.seatunnel.connectors.seatunnel.sentry.config.SentrySinkOptions; import org.apache.seatunnel.connectors.seatunnel.sentry.sink.SentrySinkFactory; @@ -29,9 +31,22 @@ import org.junit.jupiter.api.Test; import java.util.HashMap; import java.util.Map; +import java.util.ServiceLoader; class SentryFactoryTest { + @Test + void testSourceDiscovery() { + boolean found = false; + for (Factory factory : ServiceLoader.load(Factory.class)) { + if (factory instanceof TableSourceFactory + && "Sentry".equals(factory.factoryIdentifier())) { + found = true; + } + } + Assertions.assertTrue(found, "Sentry must expose a source factory"); + } + private final OptionRule optionRule = new SentrySinkFactory().optionRule(); @Test diff --git a/seatunnel-connectors-v2/connector-sentry/src/test/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySourceTest.java b/seatunnel-connectors-v2/connector-sentry/src/test/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySourceTest.java new file mode 100644 index 0000000000..b2ca77a8f2 --- /dev/null +++ b/seatunnel-connectors-v2/connector-sentry/src/test/java/org/apache/seatunnel/connectors/seatunnel/sentry/source/SentrySourceTest.java @@ -0,0 +1,685 @@ +/* + * 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.sentry.source; + +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.Collector; +import org.apache.seatunnel.api.source.SourceReader; +import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.common.constants.JobMode; +import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated; +import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader; +import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplit; +import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitEnumeratorState; +import org.apache.seatunnel.connectors.seatunnel.sentry.exception.SentryConnectorException; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +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.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class SentrySourceTest { + + @Test + void distinguishesConfigurationErrorsFromReaderFailures() throws Exception { + Map<String, Object> invalid = options(); + invalid.put("page_size", 0); + SentryConnectorException configError = + Assertions.assertThrows(SentryConnectorException.class, () -> source(invalid)); + Assertions.assertEquals( + CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT, configError.getSeaTunnelErrorCode()); + AbstractSingleSplitReader<SeaTunnelRow> reader = + reader(options(), mock(SourceReader.Context.class)); + reader.close(); + SentryConnectorException readerError = + Assertions.assertThrows( + SentryConnectorException.class, + () -> reader.pollNext(collector(new ArrayList<>()))); + Assertions.assertEquals( + CommonErrorCodeDeprecated.READER_OPERATION_FAILED, + readerError.getSeaTunnelErrorCode()); + } + + @Test + void acceptsEngineInjectedFailFastButRejectsUnsupportedFailurePolicy() { + ReadonlyConfig injected = + MultiTableFailureHelper.withMultiTableFailurePolicy( + ReadonlyConfig.fromMap(options()), + ReadonlyConfig.fromMap(Collections.emptyMap())); + Assertions.assertNotNull( + new SentrySourceFactory() + .createSource( + new TableSourceFactoryContext( + injected, getClass().getClassLoader())) + .createSource()); + Map<String, Object> config = options(); + config.put("multi_table.failure_policy", "CONTINUE_OTHER_TABLES"); + assertFailure(() -> source(config), "FAIL_FAST"); + config.put("multi_table.failure_policy", "invalid"); + assertFailure(() -> source(config), "Invalid option multi_table.failure_policy"); + } + + private HttpServer server; + private ExecutorService executor; + private final List<AbstractSingleSplitReader<SeaTunnelRow>> readers = new ArrayList<>(); + private static final String PATH = "/api/0/projects/acme/demo/events/"; + + @AfterEach + void close() throws Exception { + for (AbstractSingleSplitReader<SeaTunnelRow> reader : readers) { + reader.close(); + } + if (server != null) { + server.stop(0); + } + if (executor != null) { + executor.shutdownNow(); + Assertions.assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS)); + } + } + + @Test + void factoryCreatesSourceWithFixedSchemaAndSupportedClass() throws Exception { + SentrySource source = source(options()); + Assertions.assertEquals(SentrySource.class, new SentrySourceFactory().getSourceClass()); + Assertions.assertEquals( + 8, source.getProducedCatalogTables().get(0).getTableSchema().getColumns().size()); + Assertions.assertEquals( + "event_id", + source.getProducedCatalogTables() + .get(0) + .getTableSchema() + .getColumns() + .get(0) + .getName()); + JobContext job = mock(JobContext.class); + when(job.getJobMode()).thenReturn(JobMode.STREAMING); + assertFailure(() -> source.setJobContext(job), "BATCH only"); + } + + @Test + void rejectsIgnoredAndInvalidOptionsWithoutLeakingValues() { + for (String key : + Arrays.asList("schema", "query", "multi_table_failure_policy", "dsn", "full")) { + Map<String, Object> options = options(); + options.put(key, "unsupported"); + Assertions.assertThrows(RuntimeException.class, () -> source(options)); + } + Map<String, Object> options = options(); + options.put("token", "secret\r\nvalue"); + SentryConnectorException error = + Assertions.assertThrows(SentryConnectorException.class, () -> source(options)); + Assertions.assertFalse(error.toString().contains("secret")); + options.put("token", "real-token"); + options.put("mock_mode", true); + assertFailure(() -> source(options), "mock_mode"); + } + + @Test + void rejectsInvalidBoundsOriginsAndParallelism() { + Map<String, Object> invalid = new HashMap<>(); + invalid.put("page_size", 101); + invalid.put("max_pages", 0); + invalid.put("max_retries", 6); + invalid.put("retry_delay_ms", 0); + invalid.put("request_timeout_ms", 0); + invalid.put("max_response_bytes", Integer.MAX_VALUE); + invalid.put("parallelism", 2); + invalid.put("organization", "../other"); + invalid.put("project", "a/b"); + invalid.put("start_time", "2026-01-03T00:00:00Z"); + invalid.put("end_time", "bad"); + invalid.put("api_base_url", "https://user:[email protected]/private";); + for (Map.Entry<String, Object> entry : invalid.entrySet()) { + Map<String, Object> options = options(); + options.put(entry.getKey(), entry.getValue()); + Assertions.assertThrows( + SentryConnectorException.class, () -> source(options), entry.getKey()); + } + for (String url : + Arrays.asList( + "http://example.com";, + "https://example.com/";, + "https://example.com?secret=1";, + "https://example.com#x";)) { + Map<String, Object> options = options(); + options.put("api_base_url", url); + assertFailure(() -> source(options), "HTTPS origin"); + } + } + + @Test + void readsTwoPagesThroughFactoryAndReaderAndStopsOnFalseResults() throws Exception { + AtomicInteger requests = new AtomicInteger(); + start( + exchange -> { + int page = requests.incrementAndGet(); + Assertions.assertEquals( + "Bearer mock-token", + exchange.getRequestHeaders().getFirst("Authorization")); + String query = exchange.getRequestURI().getRawQuery(); + Assertions.assertTrue(query.contains("start=2026-01-01T00%3A00%3A00Z")); + Assertions.assertTrue(query.contains("end=2026-01-02T00%3A00%3A00Z")); + Assertions.assertTrue(query.contains("per_page=2")); + Assertions.assertTrue(query.contains("full=false")); + if (page == 1) { + Assertions.assertFalse(query.contains("cursor=")); + } else { + Assertions.assertTrue(query.contains("cursor=0%3A2%3A0")); + } + reply( + exchange, + 200, + page == 1 + ? "[" + event("a") + "," + event("b") + "]" + : "[" + event("c") + "]", + link(page == 1, "0:2:0")); + }); + SourceReader.Context context = mock(SourceReader.Context.class); + AbstractSingleSplitReader<SeaTunnelRow> reader = reader(localOptions(), context); + List<SeaTunnelRow> rows = new ArrayList<>(); + Collector<SeaTunnelRow> collector = collector(rows); + reader.pollNext(collector); + reader.pollNext(collector); + Assertions.assertEquals(2, requests.get()); + Assertions.assertEquals( + Arrays.asList("a", "b", "c"), + Arrays.asList( + rows.get(0).getField(0), rows.get(1).getField(0), rows.get(2).getField(0))); + Assertions.assertEquals("Résumé ✓", rows.get(0).getField(4)); + Assertions.assertNull(rows.get(0).getField(5)); + Assertions.assertTrue( + rows.get(0).getField(7).toString().contains("0.12345678901234567890123456789")); + verify(context).signalNoMoreElement(); + } + + @Test + void emptyFinalPageCompletesWithoutExtraRequest() throws Exception { + AtomicInteger count = new AtomicInteger(); + start( + exchange -> { + count.incrementAndGet(); + reply(exchange, 200, "[]", link(false, "0:0:0")); + }); + SourceReader.Context context = mock(SourceReader.Context.class); + List<SeaTunnelRow> rows = new ArrayList<>(); + reader(localOptions(), context).pollNext(collector(rows)); + Assertions.assertTrue(rows.isEmpty()); + Assertions.assertEquals(1, count.get()); + verify(context).signalNoMoreElement(); + } + + @Test + void retriesTransientFailureWithoutEmittingDuplicates() throws Exception { + AtomicInteger count = new AtomicInteger(); + start( + exchange -> { + if (count.incrementAndGet() == 1) { + exchange.getResponseHeaders().add("Retry-After", "0"); + reply(exchange, 429, "private", null); + } else { + reply(exchange, 200, "[" + event("a") + "]", link(false, "0:0:0")); + } + }); + List<SeaTunnelRow> rows = new ArrayList<>(); + reader(localOptions(), mock(SourceReader.Context.class)).pollNext(collector(rows)); + Assertions.assertEquals(2, count.get()); + Assertions.assertEquals(1, rows.size()); + } + + @Test + void doesNotRetryPermissionFailuresOrRedirectCredentials() throws Exception { + AtomicInteger count = new AtomicInteger(); + start( + exchange -> { + count.incrementAndGet(); + exchange.getResponseHeaders().add("Location", origin() + "/unexpected"); + reply(exchange, 302, "private-event-and-token", null); + }); + assertFailure( + () -> + reader(localOptions(), mock(SourceReader.Context.class)) + .pollNext(collector(new ArrayList<>())), + "HTTP 302"); + Assertions.assertEquals(1, count.get()); + } + + @Test + void boundsRetryBudgetAndWithholdsRemoteBody() throws Exception { + AtomicInteger count = new AtomicInteger(); + start( + exchange -> { + count.incrementAndGet(); + reply(exchange, 503, "private-event-and-token", null); + }); + Map<String, Object> options = localOptions(); + options.put("max_retries", 2); + SentryConnectorException error = + Assertions.assertThrows( + SentryConnectorException.class, + () -> + reader(options, mock(SourceReader.Context.class)) + .pollNext(collector(new ArrayList<>()))); + Assertions.assertTrue(error.getMessage().contains("retry budget")); + Assertions.assertEquals( + CommonErrorCodeDeprecated.HTTP_OPERATION_FAILED, error.getSeaTunnelErrorCode()); + Assertions.assertFalse(error.toString().contains("private-event")); + Assertions.assertNull(error.getCause()); + Assertions.assertEquals(3, count.get()); + } + + @Test + void permissionFailureIsNotRetried() throws Exception { + AtomicInteger count = new AtomicInteger(); + start( + exchange -> { + count.incrementAndGet(); + reply(exchange, 403, "private", null); + }); + assertFailure( + () -> + reader(localOptions(), mock(SourceReader.Context.class)) + .pollNext(collector(new ArrayList<>())), + "HTTP 403"); + Assertions.assertEquals(1, count.get()); + } + + @Test + void rejectsMalformedPayloadBeforeEmittingRows() { + String endpoint = "https://sentry.io"; + PATH; + String link = "<" + endpoint + "?cursor=0:0:0>; rel=\"next\"; results=\"false\""; + for (String body : + Arrays.asList( + "{}", + "null", + "[null]", + "[{}]", + "[" + event("a") + ",{}]", + "[{\"eventID\":3}]", + "[{\"eventID\":\"a\",\"eventID\":\"b\"}]", + "[" + event("a") + "] {}")) { + SentryConnectorException error = + Assertions.assertThrows( + SentryConnectorException.class, + () -> + new SentryPage( + body.getBytes(StandardCharsets.UTF_8), link, endpoint), + body); + Assertions.assertEquals( + CommonErrorCodeDeprecated.HTTP_OPERATION_FAILED, error.getSeaTunnelErrorCode()); + } + } + + @Test + void validatesLinkContractAndNeverTrustsForeignUrls() { + String endpoint = "https://sentry.io"; + PATH; + for (String header : + Arrays.asList( + null, + "", + "<" + endpoint + ">; rel=\"next\"", + "<" + endpoint + ">; rel=\"previous\"; results=\"false\"", + "<https://evil.example"; + + PATH + + "?cursor=a>; rel=\"next\"; results=\"true\"", + "<" + endpoint + "?cursor=a>; rel=\"next\"; results=\"true\"; cursor=\"b\"", + "<" + endpoint + "?cursor=a&cursor=b>; rel=\"next\"; results=\"true\"", + "<" + endpoint + "?cursor&cursor=b>; rel=\"next\"; results=\"true\"", + "<" + endpoint + "?cursor=a>; rel=\"next\"; results=\"maybe\"", + "<" + + endpoint + + "?cursor=a>; rel=\"next\"; results=\"true\", <" + + endpoint + + "?cursor=b>; rel=\"next\"; results=\"true\"")) { + Assertions.assertThrows( + SentryConnectorException.class, () -> SentryPage.nextCursor(header, endpoint)); + } + Assertions.assertEquals( + "0:100:0", + SentryPage.nextCursor( + "<" + + endpoint + + "?cursor=0%3A100%3A0>; rel=\"next\"; results=\"true\"; cursor=\"0:100:0\"", + endpoint)); + } + + @Test + void repeatedCursorFailsInsteadOfCompleting() throws Exception { + start(exchange -> reply(exchange, 200, "[" + event("a") + "]", link(true, "0:2:0"))); + SourceReader.Context context = mock(SourceReader.Context.class); + assertFailure( + () -> reader(localOptions(), context).pollNext(collector(new ArrayList<>())), + "repeated pagination cursor"); + verify(context, never()).signalNoMoreElement(); + } + + @Test + void maxPagesFailsRatherThanTruncating() throws Exception { + start(exchange -> reply(exchange, 200, "[]", link(true, "0:2:0"))); + Map<String, Object> options = localOptions(); + options.put("max_pages", 1); + SourceReader.Context context = mock(SourceReader.Context.class); + assertFailure( + () -> reader(options, context).pollNext(collector(new ArrayList<>())), "max_pages"); + verify(context, never()).signalNoMoreElement(); + } + + @Test + void limitsChunkedBodyAndDoesNotCompleteOnMissingPagination() throws Exception { + start( + exchange -> { + exchange.getResponseHeaders().add("Link", link(false, "0:0:0")); + exchange.sendResponseHeaders(200, 0); + exchange.getResponseBody().write(new byte[2048]); + exchange.close(); + }); + Map<String, Object> options = localOptions(); + options.put("max_response_bytes", 1024); + assertFailure( + () -> + reader(options, mock(SourceReader.Context.class)) + .pollNext(collector(new ArrayList<>())), + "max_response_bytes"); + } + + @Test + void respectsRetryAfterBoundsAndHttpDate() { + Instant now = Instant.parse("2026-01-01T00:00:00Z"); + Assertions.assertEquals(1000, SentryClient.retryDelay("0", 1000, now)); + Assertions.assertEquals( + 2000, SentryClient.retryDelay("Thu, 1 Jan 2026 00:00:02 GMT", 1, now)); + assertFailure(() -> SentryClient.retryDelay("61", 1, now), "Retry-After"); + assertFailure(() -> SentryClient.retryDelay("9223372036854775807", 1, now), "Retry-After"); + assertFailure(() -> SentryClient.retryDelay("private-token", 1, now), "Retry-After"); + } + + @Test + void rejectsMissingLinkBeforeAnyRowsOrCompletion() throws Exception { + start(exchange -> reply(exchange, 200, "[" + event("a") + "]", null)); + SourceReader.Context context = mock(SourceReader.Context.class); + List<SeaTunnelRow> rows = new ArrayList<>(); + assertFailure( + () -> reader(localOptions(), context).pollNext(collector(rows)), "Link header"); + Assertions.assertTrue(rows.isEmpty()); + verify(context, never()).signalNoMoreElement(); + } + + @Test + void rejectsOversizedHttpHeadersWithBoundedRedactedFailure() throws Exception { + start( + exchange -> { + char[] padding = new char[20000]; + Arrays.fill(padding, 'x'); + exchange.getResponseHeaders().add("X-Example", new String(padding)); + reply(exchange, 200, "[]", link(false, "0:0:0")); + }); + Map<String, Object> options = localOptions(); + options.put("max_retries", 0); + assertFailure( + () -> + reader(options, mock(SourceReader.Context.class)) + .pollNext(collector(new ArrayList<>())), + "retry budget"); + } + + @Test + void closeWakesLongRetryDelayWithoutCompleting() throws Exception { + CountDownLatch responseSent = new CountDownLatch(1); + start( + exchange -> { + exchange.sendResponseHeaders(503, -1); + exchange.close(); + responseSent.countDown(); + }); + Map<String, Object> options = localOptions(); + options.put("retry_delay_ms", 60000); + SourceReader.Context context = mock(SourceReader.Context.class); + AbstractSingleSplitReader<SeaTunnelRow> reader = reader(options, context); + executor = Executors.newSingleThreadExecutor(); + AtomicReference<Thread> pollingThread = new AtomicReference<>(); + Future<?> future = + executor.submit( + () -> { + pollingThread.set(Thread.currentThread()); + return Assertions.assertThrows( + Exception.class, + () -> reader.pollNext(collector(new ArrayList<>()))); + }); + Assertions.assertTrue(responseSent.await(5, TimeUnit.SECONDS)); + org.awaitility.Awaitility.await() + .atMost(5, TimeUnit.SECONDS) + .until( + () -> + pollingThread.get() != null + && pollingThread.get().getState() + == Thread.State.TIMED_WAITING + && Arrays.stream(pollingThread.get().getStackTrace()) + .anyMatch( + frame -> + frame.getMethodName() + .equals("retry"))); + reader.close(); + future.get(5, TimeUnit.SECONDS); + verify(context, never()).signalNoMoreElement(); + } + + @Test + void closingCancelsInFlightRequestAndPreventsCompletion() throws Exception { + CountDownLatch received = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + start( + exchange -> { + received.countDown(); + try { + release.await(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + exchange.close(); + }); + SourceReader.Context context = mock(SourceReader.Context.class); + AbstractSingleSplitReader<SeaTunnelRow> reader = reader(localOptions(), context); + executor = Executors.newSingleThreadExecutor(); + Future<?> future = + executor.submit( + () -> { + Assertions.assertThrows( + Exception.class, + () -> reader.pollNext(collector(new ArrayList<>()))); + }); + Assertions.assertTrue(received.await(5, TimeUnit.SECONDS)); + reader.close(); + future.get(5, TimeUnit.SECONDS); + release.countDown(); + verify(context, never()).signalNoMoreElement(); + } + + @Test + void requestDeadlineBoundsStalledResponse() throws Exception { + CountDownLatch release = new CountDownLatch(1); + start( + exchange -> { + try { + release.await(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + exchange.close(); + }); + Map<String, Object> options = localOptions(); + options.put("request_timeout_ms", 50); + options.put("max_retries", 0); + try { + assertFailure( + () -> + reader(options, mock(SourceReader.Context.class)) + .pollNext(collector(new ArrayList<>())), + "retry budget"); + } finally { + release.countDown(); + } + } + + @Test + void checkpointContainsNoCredentialsAndRestorationReplaysWindow() throws Exception { + AtomicInteger count = new AtomicInteger(); + start( + exchange -> { + count.incrementAndGet(); + reply(exchange, 200, "[" + event("a") + "]", link(false, "0:0:0")); + }); + AbstractSingleSplitReader<SeaTunnelRow> first = + reader(localOptions(), mock(SourceReader.Context.class)); + List<SingleSplit> state = first.snapshotState(1); + Assertions.assertNull(state.get(0).getState()); + first.pollNext(collector(new ArrayList<>())); + AbstractSingleSplitReader<SeaTunnelRow> restored = + reader(localOptions(), mock(SourceReader.Context.class)); + restored.addSplits(state); + List<SeaTunnelRow> rows = new ArrayList<>(); + restored.pollNext(collector(rows)); + Assertions.assertEquals(2, count.get()); + Assertions.assertEquals(1, rows.size()); + } + + private SentrySource source(Map<String, Object> options) { + return (SentrySource) + new SentrySourceFactory() + .<SeaTunnelRow, SingleSplit, SingleSplitEnumeratorState>createSource( + new TableSourceFactoryContext( + ReadonlyConfig.fromMap(options), + getClass().getClassLoader())) + .createSource(); + } + + private AbstractSingleSplitReader<SeaTunnelRow> reader( + Map<String, Object> options, SourceReader.Context context) throws Exception { + AbstractSingleSplitReader<SeaTunnelRow> reader = source(options).createReader(context); + readers.add(reader); + reader.open(); + return reader; + } + + private Collector<SeaTunnelRow> collector(List<SeaTunnelRow> rows) throws Exception { + Collector<SeaTunnelRow> collector = mock(Collector.class); + when(collector.getCheckpointLock()).thenReturn(new Object()); + doAnswer( + invocation -> { + rows.add(invocation.getArgument(0)); + return null; + }) + .when(collector) + .collect(any(SeaTunnelRow.class)); + return collector; + } + + private Map<String, Object> options() { + Map<String, Object> options = new HashMap<>(); + options.put("token", "mock-token"); + options.put("organization", "acme"); + options.put("project", "demo"); + options.put("start_time", "2026-01-01T00:00:00Z"); + options.put("end_time", "2026-01-02T00:00:00Z"); + options.put("parallelism", 1); + options.put("retry_delay_ms", 1); + options.put("page_size", 2); + return options; + } + + private Map<String, Object> localOptions() { + Map<String, Object> options = options(); + options.put("mock_mode", true); + options.put("api_base_url", origin()); + return options; + } + + private void start(com.sun.net.httpserver.HttpHandler handler) throws IOException { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext(PATH, handler); + server.start(); + } + + private String origin() { + return "http://127.0.0.1:"; + server.getAddress().getPort(); + } + + private String link(boolean more, String cursor) { + return "<" + + origin() + + PATH + + "?cursor=" + + cursor + + ">; rel=\"next\"; results=\"" + + more + + "\""; + } + + private static String event(String id) { + return "{\"eventID\":\"" + + id + + "\",\"groupID\":\"42\",\"projectID\":\"7\",\"dateCreated\":\"2026-01-01T12:00:00Z\",\"title\":\"Résumé ✓\",\"message\":null,\"platform\":\"java\",\"value\":0.12345678901234567890123456789}"; + } + + private static void reply(HttpExchange exchange, int status, String body, String link) + throws IOException { + if (link != null) { + exchange.getResponseHeaders().add("Link", link); + } + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(status, bytes.length); + exchange.getResponseBody().write(bytes); + exchange.close(); + } + + private static void assertFailure( + org.junit.jupiter.api.function.Executable action, String message) { + SentryConnectorException error = + Assertions.assertThrows(SentryConnectorException.class, action); + Assertions.assertTrue(error.getMessage().contains(message), error.getMessage()); + } +} diff --git a/seatunnel-dist/release-docs/LICENSE b/seatunnel-dist/release-docs/LICENSE index 7f6ece4436..1a601c2b1d 100644 --- a/seatunnel-dist/release-docs/LICENSE +++ b/seatunnel-dist/release-docs/LICENSE @@ -488,3 +488,7 @@ The Google Analytics 4 connector also bundles these components: The existing CDDL entry for javax.annotation:javax.annotation-api:1.3.2 applies to this connector as well. Its license is retained in the connector JAR. + +Sentry source HTTP transport: + (Apache-2.0) httpclient (org.apache.httpcomponents:httpclient:4.5.13 - https://github.com/apache/httpcomponents-client) + (Apache-2.0) httpcore (org.apache.httpcomponents:httpcore:4.4.16 - https://hc.apache.org/) diff --git a/seatunnel-dist/release-docs/NOTICE b/seatunnel-dist/release-docs/NOTICE index 6675d77591..3b29d17841 100644 --- a/seatunnel-dist/release-docs/NOTICE +++ b/seatunnel-dist/release-docs/NOTICE @@ -869,6 +869,22 @@ The initial codebase was donated to the ASF by HugeGraph Authors, copyright 2017 + Copyright 2012-2023 Aerospike, Inc. + Licensed under the Apache License, Version 2.0 +======================================================================== +Sentry source HTTP transport notices +======================================================================== + +Apache HttpClient +Copyright 1999-2020 The Apache Software Foundation + +This product includes software developed at +The Apache Software Foundation (http://www.apache.org/). + +Apache HttpCore +Copyright 2005-2022 The Apache Software Foundation + +This product includes software developed at +The Apache Software Foundation (http://www.apache.org/). + ========================================================================= +=============================================================================== diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/pom.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/pom.xml new file mode 100644 index 0000000000..4c4012800a --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/pom.xml @@ -0,0 +1,52 @@ +<?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-sentry-e2e</artifactId> + <name>SeaTunnel : E2E : Connector V2 : Sentry</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-sentry</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-sentry-e2e/src/test/java/org/apache/seatunnel/e2e/connector/sentry/SentryFixtureTest.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/src/test/java/org/apache/seatunnel/e2e/connector/sentry/SentryFixtureTest.java new file mode 100644 index 0000000000..b0fb306659 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/src/test/java/org/apache/seatunnel/e2e/connector/sentry/SentryFixtureTest.java @@ -0,0 +1,37 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.e2e.connector.sentry; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockserver.configuration.Configuration; +import org.mockserver.logging.MockServerLogger; +import org.mockserver.matchers.HttpRequestPropertiesMatcher; + +class SentryFixtureTest { + @Test + void firstPageMatcherExcludesCursorRequests() { + HttpRequestPropertiesMatcher matcher = + new HttpRequestPropertiesMatcher(new Configuration(), new MockServerLogger()); + matcher.apply(SentryIT.firstRequest()); + Assertions.assertTrue(matcher.matches(null, SentryIT.request())); + Assertions.assertFalse( + matcher.matches( + null, SentryIT.request().withQueryStringParameter("cursor", "0:2:0"))); + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/src/test/java/org/apache/seatunnel/e2e/connector/sentry/SentryIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/src/test/java/org/apache/seatunnel/e2e/connector/sentry/SentryIT.java new file mode 100644 index 0000000000..ddff08d1fc --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/src/test/java/org/apache/seatunnel/e2e/connector/sentry/SentryIT.java @@ -0,0 +1,130 @@ +/* + * 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.sentry; + +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.model.NottableOptionalString; +import org.mockserver.model.NottableString; +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; + +/** Engine integration with a deterministic API fixture, not a live Sentry deployment. */ +public class SentryIT extends TestSuiteBase implements TestResource { + private static final String ENDPOINT = + "http://sentry-fixture:1080/api/0/projects/acme/demo/events/";; + private GenericContainer<?> fixture; + private MockServerClient client; + + @BeforeAll + @Override + public void startUp() { + fixture = + new GenericContainer<>(DockerImageName.parse("mockserver/mockserver:5.14.0")) + .withNetwork(NETWORK) + .withNetworkAliases("sentry-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 readsAllPagesAfterRateLimitThroughEngine(TestContainer container) throws Exception { + client.reset(); + HttpRequest first = firstRequest(); + HttpRequest second = request().withQueryStringParameter("cursor", "0:2:0"); + client.when(first, Times.exactly(1)) + .respond( + HttpResponse.response().withStatusCode(429).withHeader("Retry-After", "0")); + client.when(first).respond(response(true, "[" + event("a") + "," + event("b") + "]")); + client.when(second).respond(response(false, "[" + event("c") + "]")); + Container.ExecResult result = container.executeJob("/sentry_to_assert.conf"); + Assertions.assertEquals(0, result.getExitCode(), result.getStderr()); + client.verify(first, VerificationTimes.exactly(2)); + client.verify(second, VerificationTimes.exactly(1)); + client.verify( + HttpRequest.request().withPath("/api/0/projects/acme/demo/events/"), + VerificationTimes.exactly(3)); + } + + static HttpRequest firstRequest() { + return request() + .withQueryStringParameter( + NottableOptionalString.optional("cursor"), NottableString.not(".*")); + } + + static HttpRequest request() { + return HttpRequest.request() + .withMethod("GET") + .withPath("/api/0/projects/acme/demo/events/") + .withHeader("Authorization", "Bearer mock-token") + .withQueryStringParameter("start", "2026-01-01T00:00:00Z") + .withQueryStringParameter("end", "2026-01-02T00:00:00Z") + .withQueryStringParameter("per_page", "2") + .withQueryStringParameter("full", "false") + .withQueryStringParameter("sample", "false"); + } + + private HttpResponse response(boolean more, String body) { + return HttpResponse.response() + .withHeader("Content-Type", "application/json") + .withHeader( + "Link", + "<" + ENDPOINT + "?cursor=0:2:0>; rel=\"next\"; results=\"" + more + "\"") + .withBody(body); + } + + private String event(String id) { + return "{\"eventID\":\"" + + id + + "\",\"groupID\":\"42\",\"projectID\":\"7\",\"dateCreated\":\"2026-01-01T12:00:00Z\",\"title\":\"Sentry event\",\"message\":null,\"platform\":\"java\"}"; + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/src/test/java/org/apache/seatunnel/e2e/connector/sentry/SentryMaskingTest.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/src/test/java/org/apache/seatunnel/e2e/connector/sentry/SentryMaskingTest.java new file mode 100644 index 0000000000..14d72ca6cf --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/src/test/java/org/apache/seatunnel/e2e/connector/sentry/SentryMaskingTest.java @@ -0,0 +1,101 @@ +/* + * 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.sentry; + +import org.apache.seatunnel.shade.com.typesafe.config.Config; + +import org.apache.seatunnel.core.starter.utils.ConfigBuilder; + +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.Logger; +import org.apache.logging.log4j.core.LoggerContext; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.Property; +import org.apache.logging.log4j.core.impl.Log4jContextFactory; +import org.apache.logging.log4j.core.layout.PatternLayout; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SentryMaskingTest { + @Test + void startupLogsMaskTokenWithoutChangingExecutableConfiguration() { + List<String> messages = new ArrayList<>(); + AbstractAppender appender = + new AbstractAppender( + "sentry-config-test", + null, + PatternLayout.createDefaultLayout(), + true, + Property.EMPTY_ARRAY) { + @Override + public void append(LogEvent event) { + messages.add(event.getMessage().getFormattedMessage()); + } + }; + // Initialize the actual SLF4J logging context before attaching the capture. Java 8 and 11 + // can select different caller-classloader contexts in the shaded test classpath. + ConfigBuilder.of(Collections.emptyMap()); + Set<LoggerContext> contexts = + new HashSet<>( + ((Log4jContextFactory) LogManager.getFactory()) + .getSelector() + .getLoggerContexts()); + contexts.add((LoggerContext) LogManager.getContext(false)); + Map<Logger, Level> loggers = new HashMap<>(); + appender.start(); + for (LoggerContext context : contexts) { + Logger logger = context.getLogger(ConfigBuilder.class.getName()); + loggers.put(logger, logger.getLevel()); + logger.addAppender(appender); + logger.setLevel(Level.INFO); + } + try { + Map<String, Object> source = Collections.singletonMap("token", "DO-NOT-LOG-SENTRY"); + Config config = + ConfigBuilder.of( + Collections.singletonMap("source", Collections.singletonList(source))); + assertEquals( + "DO-NOT-LOG-SENTRY", config.getConfigList("source").get(0).getString("token")); + assertTrue( + messages.stream().anyMatch(message -> message.contains("Parsed config file"))); + assertFalse(messages.toString().contains("DO-NOT-LOG-SENTRY")); + assertTrue(messages.toString().contains("******")); + } finally { + loggers.forEach( + (logger, level) -> { + logger.removeAppender(appender); + logger.setLevel(level); + }); + appender.stop(); + } + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/src/test/resources/logback-test.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/src/test/resources/logback-test.xml new file mode 100644 index 0000000000..fa971259e4 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/src/test/resources/logback-test.xml @@ -0,0 +1,27 @@ +<?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. +--> +<configuration> + <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender"> + <encoder> + <pattern>%d{HH:mm:ss.SSS} %-5level %logger - %msg%n</pattern> + </encoder> + </appender> + <logger name="com.github.dockerjava" level="WARN" /> + <logger name="org.mockserver" level="WARN" /> + <root level="INFO"> + <appender-ref ref="CONSOLE" /> + </root> +</configuration> diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/src/test/resources/sentry_to_assert.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/src/test/resources/sentry_to_assert.conf new file mode 100644 index 0000000000..2c68eb6fc2 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-sentry-e2e/src/test/resources/sentry_to_assert.conf @@ -0,0 +1,53 @@ +# +# 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 { + Sentry { + plugin_output = "events" + token = "mock-token" + mock_mode = true + api_base_url = "http://sentry-fixture:1080"; + organization = "acme" + project = "demo" + start_time = "2026-01-01T00:00:00Z" + end_time = "2026-01-02T00:00:00Z" + page_size = 2 + retry_delay_ms = 1 + } +} +sink { + Assert { + plugin_input = "events" + rules { + row_rules = [ + {rule_type = MIN_ROW, rule_value = 3}, + {rule_type = MAX_ROW, rule_value = 3} + ] + field_rules = [ + {field_name = event_id, field_type = string, field_value = [{rule_type = NOT_NULL}]}, + {field_name = group_id, field_type = string, field_value = [{rule_type = NOT_NULL, equals_to = "42"}]}, + {field_name = project_id, field_type = string, field_value = [{rule_type = NOT_NULL, equals_to = "7"}]}, + {field_name = date_created, field_type = string, field_value = [{rule_type = NOT_NULL, equals_to = "2026-01-01T12:00:00Z"}]}, + {field_name = title, field_type = string, field_value = [{rule_type = NOT_NULL, equals_to = "Sentry event"}]}, + {field_name = message, field_type = string, field_value = [{rule_type = NULL}]}, + {field_name = platform, field_type = string, field_value = [{rule_type = NOT_NULL, equals_to = "java"}]}, + {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 8e41797e13..b42a2f3ce9 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml @@ -96,6 +96,7 @@ <module>connector-milvus-e2e</module> <module>connector-activemq-e2e</module> <module>connector-prometheus-e2e</module> + <module>connector-sentry-e2e</module> <module>connector-qdrant-e2e</module> <module>connector-sls-e2e</module> <module>connector-socket-e2e</module>
