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-12719-a7768a13285a53853acbaa2db66fca47530a46af in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 613803559a11a63b0bd6a138122d323a935e5def Author: Saloni Gupta <[email protected]> AuthorDate: Thu Oct 15 07:13:26 2026 +0000 [Connector-V2] [Asana] Asana Source Connector (#12719) --- config/plugin_config | 1 + .../connectors/changelog/connector-http-asana.md | 3 + docs/en/connectors/source/Asana.md | 83 ++++++++ .../connectors/changelog/connector-http-asana.md | 3 + docs/zh/connectors/source/Asana.md | 76 ++++++++ plugin-mapping.properties | 1 + .../{ => connector-http-asana}/pom.xml | 49 +++-- .../connectors/seatunnel/asana/AsanaSource.java | 128 +++++++++++++ .../seatunnel/asana/AsanaSourceFactory.java | 66 +++++++ .../seatunnel/asana/AsanaSourceReader.java | 97 ++++++++++ .../seatunnel/asana/config/AsanaSourceOptions.java | 50 +++++ .../asana/config/AsanaSourceParameter.java | 85 +++++++++ .../seatunnel/asana/AsanaSourceFactoryTest.java | 212 +++++++++++++++++++++ .../seatunnel/asana/AsanaSourceReaderTest.java | 163 ++++++++++++++++ seatunnel-connectors-v2/connector-http/pom.xml | 1 + seatunnel-dist/pom.xml | 6 + .../connector-asana-e2e}/pom.xml | 61 +++--- .../seatunnel/e2e/connector/asana/AsanaIT.java | 118 ++++++++++++ .../src/test/resources/asana_source_to_assert.conf | 57 ++++++ seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml | 1 + 20 files changed, 1202 insertions(+), 59 deletions(-) diff --git a/config/plugin_config b/config/plugin_config index 6f1ee3fac2..7600b6afc3 100644 --- a/config/plugin_config +++ b/config/plugin_config @@ -71,6 +71,7 @@ connector-graphql connector-hive connector-http-base connector-mem0 +connector-http-asana connector-http-feishu connector-http-gitlab connector-http-github diff --git a/docs/en/connectors/changelog/connector-http-asana.md b/docs/en/connectors/changelog/connector-http-asana.md new file mode 100644 index 0000000000..892bf65296 --- /dev/null +++ b/docs/en/connectors/changelog/connector-http-asana.md @@ -0,0 +1,3 @@ +| Change | Commit | Version | +| --- | --- | --- | +| [Feature][connector-http-asana] Add Asana source connector | https://github.com/apache/seatunnel | dev | \ No newline at end of file diff --git a/docs/en/connectors/source/Asana.md b/docs/en/connectors/source/Asana.md new file mode 100644 index 0000000000..c7c54a38b9 --- /dev/null +++ b/docs/en/connectors/source/Asana.md @@ -0,0 +1,83 @@ +import ChangeLog from '../changelog/connector-http-asana.md'; + +# Http-Asana + +> Asana source connector + +## Description + +Reads tasks from the Asana REST API (`GET /tasks`) for one project. Built on +`connector-http-base`. Batch mode only. + +## Key Features + +- [x] Batch +- [ ] Stream +- [ ] Exactly-once +- [ ] Column projection +- [ ] Parallelism + +## Options + +| Name | Type | Required | Default | Description | +| --- | --- | --- | --- | --- | +| api_key | String | Yes | - | Asana Personal Access Token (raw token, no `Bearer ` prefix). Needs `tasks:read`. | +| project_gid | String | Yes | - | Project whose tasks are read. | +| modified_since | String | No | - | Optional ISO 8601 timestamp, e.g. `2026-10-01T00:00:00Z`. | +| base_url | String | No | `https://app.asana.com/api/1.0` | Asana REST API base URL. | +| retry | Int | No | `3` | Max retries on HTTP 429 and 5xx. | +| retry_backoff_multiplier_ms | Int | No | `100` | Backoff multiplier in milliseconds. | +| retry_backoff_max_ms | Int | No | `10000` | Maximum backoff time in milliseconds. | + +### api_key +Asana Personal Access Token (raw token, no `Bearer ` prefix). Needs `tasks:read`. + +### project_gid +Project whose tasks are read. + +### modified_since +Optional ISO 8601 timestamp, e.g. `2026-10-01T00:00:00Z`. Asana counts a task as +modified when its own properties or associations change, but not when only a +subtask changes. The connector keeps no state between runs. + +### retry +Max retries on HTTP 429 and 5xx (and, in the base client, IOExceptions). +Backoff doubles from `retry_backoff_multiplier_ms` up to `retry_backoff_max_ms`. + +## Output columns + +All columns are STRING: `gid`, `name`, `completed`, `completed_at`, `created_at`, +`modified_at`, `due_on`, `assignee_gid`, `assignee_name`, `permalink_url`. +`completed_at`, `due_on` and the assignee columns can be null. + +## Example + +```hocon +env { + execution.parallelism = 1 + job.mode = BATCH +} + +source { + Asana { + api_key = "<PAT>" + project_gid = "1200000000000000" + modified_since = "2026-10-01T00:00:00Z" + plugin_output = "asana_tasks" + } +} + +sink { + Console { + plugin_input = "asana_tasks" + } +} +``` + +## Notes + +Each run reads the project from the first page. After a failure, a restarted job re-reads from the beginning and may emit rows that were already sent; +tasks edited during a run can also appear twice. Every row includes the task `gid`, which can assist an idempotent downstream sink in deduplicating or upserting, +but the connector provides no delivery guarantee. Rate-limit handling is bounded backoff only; `Retry-After` is not honored. + +<ChangeLog /> diff --git a/docs/zh/connectors/changelog/connector-http-asana.md b/docs/zh/connectors/changelog/connector-http-asana.md new file mode 100644 index 0000000000..0f8d1522a3 --- /dev/null +++ b/docs/zh/connectors/changelog/connector-http-asana.md @@ -0,0 +1,3 @@ +| Change | Commit | Version | +|------------------------------------------------------------| --- | --- | +| [Feature][connector-http-asana] Add Asana source connector | https://github.com/apache/seatunnel | dev | \ No newline at end of file diff --git a/docs/zh/connectors/source/Asana.md b/docs/zh/connectors/source/Asana.md new file mode 100644 index 0000000000..8296140169 --- /dev/null +++ b/docs/zh/connectors/source/Asana.md @@ -0,0 +1,76 @@ +import ChangeLog from '../changelog/connector-http-asana.md'; + +# Http-Asana + +> Asana 源连接器 + +## 描述 + +从 Asana REST API (`GET /tasks`) 读取单个项目的任务。基于 `connector-http-base` 构建。仅支持批处理模式。 + +## 主要特性 + +- [x] 批处理 +- [ ] 流处理 +- [ ] 精确一次 +- [ ] 列投影 +- [ ] 并行度 + +## 选项 + +| 名称 | 类型 | 是否必填 | 默认值 | 描述 | +| --- | --- | --- | --- | --- | +| api_key | String | 是 | - | Asana 个人访问令牌（原始令牌，不含 `Bearer ` 前缀）。需要 `tasks:read` 权限。 | +| project_gid | String | 是 | - | 要读取其任务的项目。 | +| modified_since | String | 否 | - | 可选的 ISO 8601 时间戳，例如 `2026-10-01T00:00:00Z`。 | +| base_url | String | 否 | `https://app.asana.com/api/1.0` | Asana REST API 基础 URL。 | +| retry | Int | 否 | `3` | HTTP 429 和 5xx 错误的最大重试次数。 | +| retry_backoff_multiplier_ms | Int | 否 | `100` | 退避乘数（毫秒）。 | +| retry_backoff_max_ms | Int | 否 | `10000` | 最大退避时间（毫秒）。 | + +### api_key +Asana 个人访问令牌（原始令牌，不含 `Bearer ` 前缀）。需要 `tasks:read` 权限。 + +### project_gid +要读取其任务的项目。 + +### modified_since +可选的 ISO 8601 时间戳，例如 `2026-10-01T00:00:00Z`。当任务自身的属性或关联发生变化时，Asana 会将其视为修改，但仅子任务发生变化时不计入。连接器在运行之间不保留状态。 + +### retry +HTTP 429 和 5xx（以及基础客户端中的 IOExceptions）的最大重试次数。退避时间从 `retry_backoff_multiplier_ms` 开始翻倍，最高可达 `retry_backoff_max_ms`。 + +## 输出列 + +所有列均为 STRING 类型：`gid`、`name`、`completed`、`completed_at`、`created_at`、`modified_at`、`due_on`、`assignee_gid`、`assignee_name`、`permalink_url`。 +`completed_at`、`due_on` 以及负责人列可以为 null。 + +## 示例 + +```hocon +env { + execution.parallelism = 1 + job.mode = BATCH +} + +source { + Asana { + api_key = "<PAT>" + project_gid = "1200000000000000" + modified_since = "2026-10-01T00:00:00Z" + plugin_output = "asana_tasks" + } +} + +sink { + Console { + plugin_input = "asana_tasks" + } +} +``` + +## 注意事项 + +每次运行都从第一页开始读取项目。发生故障后，重启的任务会从头重新读取，并可能发出已经发送过的行；在运行期间编辑的任务也可能会出现两次。每一行都包含任务 `gid`，这可以协助幂等的下游 Sink 进行去重或更新（upsert），但连接器不提供送达保证。速率限制处理仅采用有界退避机制；不遵循 `Retry-After`。 + +<ChangeLog /> diff --git a/plugin-mapping.properties b/plugin-mapping.properties index 65c23b908c..cf0aa84a45 100644 --- a/plugin-mapping.properties +++ b/plugin-mapping.properties @@ -207,6 +207,7 @@ seatunnel.sink.NatsJetStream = connector-nats-jetstream seatunnel.source.Linear = connector-http-linear seatunnel.source.Splunk = connector-http-splunk seatunnel.source.WebSocket = connector-websocket +seatunnel.source.Asana = connector-http-asana # For custom transforms, make sure to use the seatunnel.transform.[PluginIdentifier]=[JarPerfix] naming convention. For example: # seatunnel.transform.Sql = seatunnel-transforms-v2 diff --git a/seatunnel-connectors-v2/connector-http/pom.xml b/seatunnel-connectors-v2/connector-http/connector-http-asana/pom.xml similarity index 50% copy from seatunnel-connectors-v2/connector-http/pom.xml copy to seatunnel-connectors-v2/connector-http/connector-http-asana/pom.xml index 92286f0525..2814a84e7b 100644 --- a/seatunnel-connectors-v2/connector-http/pom.xml +++ b/seatunnel-connectors-v2/connector-http/connector-http-asana/pom.xml @@ -22,35 +22,30 @@ <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.apache.seatunnel</groupId> - <artifactId>seatunnel-connectors-v2</artifactId> + <artifactId>connector-http</artifactId> <version>${revision}</version> </parent> - <artifactId>connector-http</artifactId> - <packaging>pom</packaging> - <name>SeaTunnel : Connectors V2 : Http :</name> - <modules> - <module>connector-http-base</module> - <module>connector-http-feishu</module> - <module>connector-http-wechat</module> - <module>connector-http-myhours</module> - <module>connector-http-lemlist</module> - <module>connector-http-klaviyo</module> - <module>connector-http-onesignal</module> - <module>connector-http-jira</module> - <module>connector-http-gitlab</module> - <module>connector-http-github</module> - <module>connector-http-notion</module> - <module>connector-http-persistiq</module> - <module>connector-http-airtable</module> - <module>connector-http-posthog</module> - <module>connector-http-shopify</module> - <module>connector-http-zendesk</module> - <module>connector-http-stripe</module> - <module>connector-http-paypal</module> - <module>connector-http-linear</module> - <module>connector-http-splunk</module> - <module>connector-mem0</module> - </modules> + <artifactId>connector-http-asana</artifactId> + <name>SeaTunnel : Connectors V2 : Http : Asana</name> + + <dependencies> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-base</artifactId> + <version>${project.version}</version> + </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>seatunnel-common</artifactId> + <version>${project.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>com.squareup.okhttp3</groupId> + <artifactId>mockwebserver</artifactId> + <scope>test</scope> + </dependency> + </dependencies> </project> diff --git a/seatunnel-connectors-v2/connector-http/connector-http-asana/src/main/java/org/apache/seatunnel/connectors/seatunnel/asana/AsanaSource.java b/seatunnel-connectors-v2/connector-http/connector-http-asana/src/main/java/org/apache/seatunnel/connectors/seatunnel/asana/AsanaSource.java new file mode 100644 index 0000000000..ddedfed126 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-asana/src/main/java/org/apache/seatunnel/connectors/seatunnel/asana/AsanaSource.java @@ -0,0 +1,128 @@ +/* + * 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.asana; + +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.common.exception.CommonErrorCodeDeprecated; +import org.apache.seatunnel.connectors.seatunnel.asana.config.AsanaSourceParameter; +import org.apache.seatunnel.connectors.seatunnel.common.source.AbstractSingleSplitReader; +import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext; +import org.apache.seatunnel.connectors.seatunnel.http.config.HttpPaginationType; +import org.apache.seatunnel.connectors.seatunnel.http.config.HttpSourceOptions; +import org.apache.seatunnel.connectors.seatunnel.http.config.JsonField; +import org.apache.seatunnel.connectors.seatunnel.http.config.PageInfo; +import org.apache.seatunnel.connectors.seatunnel.http.exception.HttpConnectorException; +import org.apache.seatunnel.connectors.seatunnel.http.source.HttpSource; +import org.apache.seatunnel.format.json.JsonDeserializationSchema; + +import lombok.extern.slf4j.Slf4j; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * Asana tasks source. Reuses {@link HttpSource}, but after super(pluginConfig) replaces the schema, + * JSONPath fields and paging so the ten fixed Asana columns and the cursor pagination + * ("next_page.offset") are used instead of user-supplied config. + */ +@Slf4j +public class AsanaSource extends HttpSource { + private final AsanaSourceParameter asanaSourceParameter = new AsanaSourceParameter(); + + public AsanaSource(ReadonlyConfig pluginConfig) { + super(pluginConfig); + asanaSourceParameter.buildWithConfig(pluginConfig); + + TableSchema.Builder schema = TableSchema.builder(); + Map<String, String> fields = new LinkedHashMap<>(); + + fields.put("gid", "$.data[*].gid"); + fields.put("name", "$.data[*].name"); + fields.put("completed", "$.data[*].completed"); + fields.put("completed_at", "$.data[*].completed_at"); + fields.put("created_at", "$.data[*].created_at"); + fields.put("modified_at", "$.data[*].modified_at"); + fields.put("due_on", "$.data[*].due_on"); + fields.put("assignee_gid", "$.data[*].assignee.gid"); + fields.put("assignee_name", "$.data[*].assignee.name"); + fields.put("permalink_url", "$.data[*].permalink_url"); + + fields.keySet() + .forEach( + c -> + schema.column( + PhysicalColumn.of( + c, BasicType.STRING_TYPE, 0, true, null, null))); + this.catalogTable = + CatalogTable.of( + TableIdentifier.of("Asana", TablePath.DEFAULT), + schema.build(), + Collections.emptyMap(), + Collections.emptyList(), + null); + this.deserializationSchema = new JsonDeserializationSchema(catalogTable, false, false); + this.jsonField = JsonField.builder().fields(fields).build(); + + PageInfo info = new PageInfo(); + info.setPageType(HttpPaginationType.CURSOR.getCode()); + info.setPageCursorFieldName("offset"); + info.setPageCursorResponseField("$.next_page.offset"); + info.setPageIndex(HttpSourceOptions.START_PAGE_NUMBER.defaultValue()); + info.setBatchSize(HttpSourceOptions.BATCH_SIZE.defaultValue()); + info.setTotalPageSize(HttpSourceOptions.TOTAL_PAGE_SIZE.defaultValue()); + info.setUsePlaceholderReplacement(false); + this.pageInfo = info; + } + + @Override + public AbstractSingleSplitReader<SeaTunnelRow> createReader( + SingleSplitReaderContext readerContext) throws Exception { + return new AsanaSourceReader( + this.asanaSourceParameter, + readerContext, + this.deserializationSchema, + jsonField, + contentField, + pageInfo); + } + + @Override + public String getPluginName() { + return "Asana"; + } + + @Override + public Boundedness getBoundedness() { + if (JobMode.STREAMING.equals(jobContext.getJobMode())) { + throw new HttpConnectorException( + CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT, + "Asana source only supports batch mode."); + } + return Boundedness.BOUNDED; + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-asana/src/main/java/org/apache/seatunnel/connectors/seatunnel/asana/AsanaSourceFactory.java b/seatunnel-connectors-v2/connector-http/connector-http-asana/src/main/java/org/apache/seatunnel/connectors/seatunnel/asana/AsanaSourceFactory.java new file mode 100644 index 0000000000..68ec01ff2a --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-asana/src/main/java/org/apache/seatunnel/connectors/seatunnel/asana/AsanaSourceFactory.java @@ -0,0 +1,66 @@ +/* + * 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.asana; + +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.TableSourceFactoryContext; +import org.apache.seatunnel.connectors.seatunnel.asana.config.AsanaSourceOptions; +import org.apache.seatunnel.connectors.seatunnel.http.source.HttpSourceFactory; + +import com.google.auto.service.AutoService; + +import java.io.Serializable; + +/** + * Factory that registers the Asana source under the identifier "Asana" and declares its options. + */ +@AutoService(Factory.class) +public class AsanaSourceFactory extends HttpSourceFactory { + @Override + public String factoryIdentifier() { + return "Asana"; + } + + @Override + public <T, SplitT extends SourceSplit, StateT extends Serializable> + TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) { + return () -> (SeaTunnelSource<T, SplitT, StateT>) new AsanaSource(context.getOptions()); + } + + @Override + public OptionRule optionRule() { + return OptionRule.builder() + .required(AsanaSourceOptions.API_KEY, AsanaSourceOptions.PROJECT_GID) + .optional( + AsanaSourceOptions.MODIFIED_SINCE, + AsanaSourceOptions.BASE_URL, + AsanaSourceOptions.RETRY, + AsanaSourceOptions.RETRY_BACKOFF_MULTIPLIER_MS, + AsanaSourceOptions.RETRY_BACKOFF_MAX_MS) + .build(); + } + + @Override + public Class<? extends SeaTunnelSource> getSourceClass() { + return AsanaSource.class; + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-asana/src/main/java/org/apache/seatunnel/connectors/seatunnel/asana/AsanaSourceReader.java b/seatunnel-connectors-v2/connector-http/connector-http-asana/src/main/java/org/apache/seatunnel/connectors/seatunnel/asana/AsanaSourceReader.java new file mode 100644 index 0000000000..d902744602 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-asana/src/main/java/org/apache/seatunnel/connectors/seatunnel/asana/AsanaSourceReader.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.asana; + +import org.apache.seatunnel.api.serialization.DeserializationSchema; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated; +import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext; +import org.apache.seatunnel.connectors.seatunnel.http.client.HttpResponse; +import org.apache.seatunnel.connectors.seatunnel.http.config.HttpParameter; +import org.apache.seatunnel.connectors.seatunnel.http.config.JsonField; +import org.apache.seatunnel.connectors.seatunnel.http.config.PageInfo; +import org.apache.seatunnel.connectors.seatunnel.http.exception.HttpConnectorException; +import org.apache.seatunnel.connectors.seatunnel.http.source.HttpSourceReader; + +import lombok.extern.slf4j.Slf4j; + +/** + * Reader that re-implements executeRequest to retry HTTP 429 and 5xx with exponential backoff + * before handing the response to the base reader. + */ +@Slf4j +public class AsanaSourceReader extends HttpSourceReader { + + public AsanaSourceReader( + HttpParameter httpParameter, + SingleSplitReaderContext readerContext, + DeserializationSchema<SeaTunnelRow> deserializationSchema, + JsonField jsonField, + String contentField, + PageInfo pageInfo) { + super( + httpParameter, + readerContext, + deserializationSchema, + jsonField, + contentField, + pageInfo); + } + + @Override + protected HttpResponse executeRequest() throws Exception { + int maxRetries = httpParameter.getRetry(); + long waitMillis = httpParameter.getRetryBackoffMultiplierMillis(); + long maxWaitMillis = httpParameter.getRetryBackoffMaxMillis(); + int attempt = 1; + while (true) { + HttpResponse response = super.executeRequest(); + int code = response.getCode(); + boolean retryable = code == 429 || (code >= 500 && code < 600); + if (!retryable) { + return response; + } + if (attempt > maxRetries) { + throw new HttpConnectorException( + CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT, + String.format( + "Asana request failed with HTTP %d after %d attempt(s) (retry=%d)", + code, attempt, maxRetries)); + } + log.warn( + "Asana returned HTTP {}; retry {}/{} in {} ms", + code, + attempt, + maxRetries, + waitMillis); + + try { + Thread.sleep(waitMillis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException( + String.format( + "Interrupted while backing off after Asana HTTP %d (attempt %d)", + code, attempt), + e); + } + waitMillis = Math.min(waitMillis * 2, maxWaitMillis); + attempt++; + } + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-asana/src/main/java/org/apache/seatunnel/connectors/seatunnel/asana/config/AsanaSourceOptions.java b/seatunnel-connectors-v2/connector-http/connector-http-asana/src/main/java/org/apache/seatunnel/connectors/seatunnel/asana/config/AsanaSourceOptions.java new file mode 100644 index 0000000000..4d968daf04 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-asana/src/main/java/org/apache/seatunnel/connectors/seatunnel/asana/config/AsanaSourceOptions.java @@ -0,0 +1,50 @@ +/* + * 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.asana.config; + +import org.apache.seatunnel.api.configuration.Option; +import org.apache.seatunnel.api.configuration.Options; +import org.apache.seatunnel.connectors.seatunnel.http.config.HttpCommonOptions; + +/** Config options specific to the Asana source, on top of the shared HTTP options. */ +public class AsanaSourceOptions extends HttpCommonOptions { + public static final Option<String> API_KEY = + Options.key("api_key") + .stringType() + .noDefaultValue() + .withDescription("Asana Personal Access Token (PAT)"); + + public static final Option<String> PROJECT_GID = + Options.key("project_gid") + .stringType() + .noDefaultValue() + .withDescription("Asana Project GID to scope the tasks endpoint"); + + public static final Option<String> MODIFIED_SINCE = + Options.key("modified_since") + .stringType() + .noDefaultValue() + .withDescription("Optional ISO timestamp to filter modified tasks"); + + public static final Option<String> BASE_URL = + Options.key("base_url") + .stringType() + .defaultValue("https://app.asana.com/api/1.0";) + .withDescription( + "Asana API base URL. Override only to point tests at a mock server."); +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-asana/src/main/java/org/apache/seatunnel/connectors/seatunnel/asana/config/AsanaSourceParameter.java b/seatunnel-connectors-v2/connector-http/connector-http-asana/src/main/java/org/apache/seatunnel/connectors/seatunnel/asana/config/AsanaSourceParameter.java new file mode 100644 index 0000000000..1a7faa0488 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-asana/src/main/java/org/apache/seatunnel/connectors/seatunnel/asana/config/AsanaSourceParameter.java @@ -0,0 +1,85 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.asana.config; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.connectors.seatunnel.http.config.HttpParameter; +import org.apache.seatunnel.connectors.seatunnel.http.config.HttpRequestMethod; + +import java.util.HashMap; + +/** + * Builds the HTTP request parameters for the Asana tasks endpoint: the tasks URL, the Bearer + * Authorization header, query params (project, page limit, fields, optional modified_since) and + * retry defaults. Validates that api_key and project_gid are not blank. + */ +public class AsanaSourceParameter extends HttpParameter { + + private static String requireNonBlank(String value, String optionKey) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException( + String.format("Asana option '%s' must not be blank.", optionKey)); + } + return value.trim(); + } + + /** + * Builds the Asana request: tasks URL, Bearer Authorization header, query params and retry + * defaults. Validates that api_key and project_gid are not blank. + */ + public void buildWithConfig(ReadonlyConfig config) { + super.buildWithConfig(config); + String apiKey = requireNonBlank(config.get(AsanaSourceOptions.API_KEY), "api_key"); + String projectGid = + requireNonBlank(config.get(AsanaSourceOptions.PROJECT_GID), "project_gid"); + + String base = config.get(AsanaSourceOptions.BASE_URL); + if (base.endsWith("/")) { + base = base.substring(0, base.length() - 1); + } + setUrl(base + "/tasks"); + setMethod(HttpRequestMethod.GET); + + if (headers == null) { + headers = new HashMap<>(); + } + headers.put("Authorization", "Bearer " + apiKey); + + params = new HashMap<>(); + params.put("project", projectGid); + params.put("limit", "100"); + params.put( + "opt_fields", + "name,completed,completed_at,created_at,modified_at,due_on," + + "assignee.gid,assignee.name,permalink_url"); + config.getOptional(AsanaSourceOptions.MODIFIED_SINCE) + .ifPresent(v -> params.put("modified_since", v)); + setKeepPageParamAsHttpParam(true); + setJsonFiledMissedReturnNull(true); + + if (!config.getOptional(AsanaSourceOptions.RETRY).isPresent()) { + setRetry(3); + } + setRetryBackoffMultiplierMillis(config.get(AsanaSourceOptions.RETRY_BACKOFF_MULTIPLIER_MS)); + setRetryBackoffMaxMillis(config.get(AsanaSourceOptions.RETRY_BACKOFF_MAX_MS)); + + if (getRetryBackoffMultiplierMillis() < 0 || getRetryBackoffMaxMillis() < 0) { + throw new IllegalArgumentException("Asana retry backoff values must not be negative."); + } + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-asana/src/test/java/org/apache/seatunnel/connectors/seatunnel/asana/AsanaSourceFactoryTest.java b/seatunnel-connectors-v2/connector-http/connector-http-asana/src/test/java/org/apache/seatunnel/connectors/seatunnel/asana/AsanaSourceFactoryTest.java new file mode 100644 index 0000000000..eba27cbe7e --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-asana/src/test/java/org/apache/seatunnel/connectors/seatunnel/asana/AsanaSourceFactoryTest.java @@ -0,0 +1,212 @@ +/* + * 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.asana; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.source.Collector; +import org.apache.seatunnel.api.source.SourceReader; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.connectors.seatunnel.asana.config.AsanaSourceParameter; +import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext; +import org.apache.seatunnel.connectors.seatunnel.http.config.HttpRequestMethod; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import okhttp3.mockwebserver.RecordedRequest; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +public class AsanaSourceFactoryTest { + + @Test + public void testFactoryIdentifier() { + AsanaSourceFactory factory = new AsanaSourceFactory(); + Assertions.assertEquals("Asana", factory.factoryIdentifier()); + } + + @Test + public void testParameterBuildingAndAuthHeader() { + HashMap<String, Object> configMap = new HashMap<>(); + configMap.put("api_key", "test-asana-token-123"); + configMap.put("project_gid", "1234567890"); + ReadonlyConfig config = ReadonlyConfig.fromMap(configMap); + + AsanaSourceParameter parameter = new AsanaSourceParameter(); + parameter.buildWithConfig(config); + + Assertions.assertNotNull(parameter.getHeaders()); + Assertions.assertEquals( + "Bearer test-asana-token-123", parameter.getHeaders().get("Authorization")); + Assertions.assertEquals(HttpRequestMethod.GET, parameter.getMethod()); + Assertions.assertEquals("1234567890", parameter.getParams().get("project")); + Assertions.assertNotNull(parameter.getParams().get("opt_fields")); + } + + @Test + public void testModifiedSinceFilterOption() { + HashMap<String, Object> configMap = new HashMap<>(); + configMap.put("api_key", "test-asana-token-123"); + configMap.put("project_gid", "1234567890"); + configMap.put("modified_since", "2026-01-01T00:00:00.000Z"); + ReadonlyConfig config = ReadonlyConfig.fromMap(configMap); + + AsanaSourceParameter parameter = new AsanaSourceParameter(); + parameter.buildWithConfig(config); + + Assertions.assertEquals( + "2026-01-01T00:00:00.000Z", parameter.getParams().get("modified_since")); + } + + @Test + public void testBlankApiKeyValidation() { + HashMap<String, Object> configMap = new HashMap<>(); + configMap.put("api_key", " "); + configMap.put("project_gid", "1234567890"); + ReadonlyConfig config = ReadonlyConfig.fromMap(configMap); + + AsanaSourceParameter parameter = new AsanaSourceParameter(); + Exception exception = + Assertions.assertThrows( + Exception.class, + () -> { + parameter.buildWithConfig(config); + }); + + String msg = exception.getMessage(); + Assertions.assertTrue(msg.contains("api_key"), "Should specify which option failed"); + Assertions.assertFalse( + msg.contains("Authorization"), "Should not contain Authorization keyword"); + } + + @Test + public void testBlankProjectGidValidation() { + HashMap<String, Object> configMap = new HashMap<>(); + configMap.put("api_key", "test-asana-token-123"); + configMap.put("project_gid", ""); + ReadonlyConfig config = ReadonlyConfig.fromMap(configMap); + + AsanaSourceParameter parameter = new AsanaSourceParameter(); + Exception exception = + Assertions.assertThrows( + Exception.class, + () -> { + parameter.buildWithConfig(config); + }); + + String msg = exception.getMessage(); + Assertions.assertTrue(msg.contains("project_gid"), "Should specify which option failed"); + Assertions.assertFalse( + msg.contains("test-asana-token-123"), "Should never leak the secret token"); + } + + @Test + public void testSourceReaderPaginationAndNullAssignee() throws Exception { + try (MockWebServer server = new MockWebServer()) { + String page1 = + "{" + + "\"data\": [" + + " {\"gid\": \"1\", \"name\": \"Task 1\", \"completed\": false, \"assignee\": {\"gid\": \"a1\", \"name\": \"Saloni\"}}," + + " {\"gid\": \"2\", \"name\": \"Task 2\", \"completed\": true, \"assignee\": null}" + + "]," + + "\"next_page\": {\"offset\": \"ab.c_-==%2B\"}" + + "}"; + + String page2 = + "{" + + "\"data\": [" + + " {\"gid\": \"3\", \"name\": \"Task 3\", \"completed\": false, \"assignee\": {\"gid\": \"a2\", \"name\": \"Suhani\"}}" + + "]," + + "\"next_page\": null" + + "}"; + + server.enqueue(new MockResponse().setBody(page1).setResponseCode(200)); + server.enqueue(new MockResponse().setBody(page2).setResponseCode(200)); + server.start(); + + Map<String, Object> configMap = new HashMap<>(); + configMap.put("base_url", server.url("/").toString()); + configMap.put("api_key", "test-token"); + configMap.put("project_gid", "12345"); + ReadonlyConfig config = ReadonlyConfig.fromMap(configMap); + + AsanaSource source = new AsanaSource(config); + + SourceReader.Context ctx = Mockito.mock(SourceReader.Context.class); + AsanaSourceReader reader = + (AsanaSourceReader) source.createReader(new SingleSplitReaderContext(ctx)); + + List<SeaTunnelRow> emittedRows = new ArrayList<>(); + Collector<SeaTunnelRow> collector = + new Collector<SeaTunnelRow>() { + @Override + public void collect(SeaTunnelRow record) { + emittedRows.add(record); + } + + @Override + public Object getCheckpointLock() { + return this; + } + }; + + reader.open(); + try { + reader.pollNext(collector); + } finally { + reader.close(); + } + + RecordedRequest first = server.takeRequest(5, TimeUnit.SECONDS); + Assertions.assertNotNull(first); + Assertions.assertNull(first.getRequestUrl().queryParameter("offset")); + + RecordedRequest second = server.takeRequest(5, TimeUnit.SECONDS); + Assertions.assertNotNull(second); + Assertions.assertEquals("ab.c_-==%2B", second.getRequestUrl().queryParameter("offset")); + Assertions.assertEquals(2, server.getRequestCount()); + + Assertions.assertEquals(3, emittedRows.size()); + Assertions.assertEquals("1", emittedRows.get(0).getField(0)); + Assertions.assertEquals("a1", emittedRows.get(0).getField(7)); + Assertions.assertEquals("2", emittedRows.get(1).getField(0)); + Assertions.assertNull(emittedRows.get(1).getField(7)); + Assertions.assertNull(emittedRows.get(1).getField(8)); + Assertions.assertEquals("3", emittedRows.get(2).getField(0)); + } + } + + @Test + public void testNegativeBackoffRejected() { + HashMap<String, Object> configMap = new HashMap<>(); + configMap.put("api_key", "test-token"); + configMap.put("project_gid", "12345"); + configMap.put("retry_backoff_multiplier_ms", -1); + AsanaSourceParameter parameter = new AsanaSourceParameter(); + Assertions.assertThrows( + IllegalArgumentException.class, + () -> parameter.buildWithConfig(ReadonlyConfig.fromMap(configMap))); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-asana/src/test/java/org/apache/seatunnel/connectors/seatunnel/asana/AsanaSourceReaderTest.java b/seatunnel-connectors-v2/connector-http/connector-http-asana/src/test/java/org/apache/seatunnel/connectors/seatunnel/asana/AsanaSourceReaderTest.java new file mode 100644 index 0000000000..b47da58e83 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-asana/src/test/java/org/apache/seatunnel/connectors/seatunnel/asana/AsanaSourceReaderTest.java @@ -0,0 +1,163 @@ +/* + * 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.asana; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.connectors.seatunnel.asana.config.AsanaSourceParameter; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import okhttp3.mockwebserver.RecordedRequest; + +import java.util.HashMap; +import java.util.Map; + +public class AsanaSourceReaderTest { + + private AsanaSourceReader newReader(MockWebServer server, int retry) { + Map<String, Object> configMap = new HashMap<>(); + configMap.put("base_url", server.url("/api/1.0").toString()); + configMap.put("api_key", "test-token"); + configMap.put("project_gid", "12345"); + configMap.put("retry", retry); + configMap.put("retry_backoff_multiplier_ms", 10); + configMap.put("retry_backoff_max_ms", 50); + ReadonlyConfig config = ReadonlyConfig.fromMap(configMap); + AsanaSourceParameter parameter = new AsanaSourceParameter(); + parameter.buildWithConfig(config); + return new AsanaSourceReader(parameter, null, null, null, null, null); + } + + @Test + public void testRetryOn429And5xx() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.enqueue(new MockResponse().setResponseCode(429)); + server.enqueue(new MockResponse().setResponseCode(500)); + server.enqueue(new MockResponse().setResponseCode(200).setBody("{\"data\":[]}")); + server.start(); + AsanaSourceReader reader = newReader(server, 3); + reader.open(); + try { + Assertions.assertEquals(200, reader.executeRequest().getCode()); + Assertions.assertEquals(3, server.getRequestCount()); + RecordedRequest first = server.takeRequest(); + Assertions.assertEquals("Bearer test-token", first.getHeader("Authorization")); + Assertions.assertTrue(first.getPath().startsWith("/api/1.0/tasks?")); + Assertions.assertTrue(first.getPath().contains("project=12345")); + Assertions.assertTrue(first.getPath().contains("limit=100")); + } finally { + reader.close(); + } + } + } + + @Test + public void testNoRetryOn401() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.enqueue(new MockResponse().setResponseCode(401).setBody("{\"errors\":[]}")); + server.start(); + AsanaSourceReader reader = newReader(server, 3); + reader.open(); + try { + Assertions.assertEquals(401, reader.executeRequest().getCode()); + Assertions.assertEquals(1, server.getRequestCount()); + } finally { + reader.close(); + } + } + } + + @Test + public void testGivesUpAfterRetries() throws Exception { + try (MockWebServer server = new MockWebServer()) { + for (int i = 0; i < 4; i++) { + server.enqueue(new MockResponse().setResponseCode(500)); + } + server.start(); + AsanaSourceReader reader = newReader(server, 3); + reader.open(); + try { + RuntimeException ex = + Assertions.assertThrows( + RuntimeException.class, () -> reader.executeRequest()); + Assertions.assertTrue(ex.getMessage().contains("HTTP 500")); + Assertions.assertEquals(4, server.getRequestCount()); + } finally { + reader.close(); + } + } + } + + @Test + public void testInterruptedBackoffRestoresFlag() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.enqueue(new MockResponse().setResponseCode(429)); + server.start(); + AsanaSourceReader reader = newReader(server, 3); + reader.open(); + + Thread.currentThread().interrupt(); + try { + reader.executeRequest(); + Assertions.fail("Expected exception due to thread interruption"); + } catch (Exception e) { + Assertions.assertTrue( + Thread.currentThread().isInterrupted(), "Interrupt flag must be restored"); + String msg = String.valueOf(e.getMessage()); + Assertions.assertFalse( + msg.contains("test-token"), "Message must not contain token"); + Assertions.assertFalse( + msg.contains("Authorization"), "Message must not contain Authorization"); + } finally { + Thread.interrupted(); + reader.close(); + } + } + } + + @Test + public void testCustomBackoffHonoredWithoutExplicitRetry() { + HashMap<String, Object> configMap = new HashMap<>(); + configMap.put("api_key", "test-token"); + configMap.put("project_gid", "12345"); + configMap.put("retry_backoff_multiplier_ms", 2000); + configMap.put("retry_backoff_max_ms", 3000); + AsanaSourceParameter parameter = new AsanaSourceParameter(); + parameter.buildWithConfig(ReadonlyConfig.fromMap(configMap)); + + Assertions.assertEquals(3, parameter.getRetry()); + Assertions.assertEquals(2000, parameter.getRetryBackoffMultiplierMillis()); + Assertions.assertEquals(3000, parameter.getRetryBackoffMaxMillis()); + } + + @Test + public void testDefaultBackoffWhenNothingConfigured() { + HashMap<String, Object> configMap = new HashMap<>(); + configMap.put("api_key", "test-token"); + configMap.put("project_gid", "12345"); + AsanaSourceParameter parameter = new AsanaSourceParameter(); + parameter.buildWithConfig(ReadonlyConfig.fromMap(configMap)); + + Assertions.assertEquals(3, parameter.getRetry()); + Assertions.assertEquals(100, parameter.getRetryBackoffMultiplierMillis()); + Assertions.assertEquals(10000, parameter.getRetryBackoffMaxMillis()); + } +} diff --git a/seatunnel-connectors-v2/connector-http/pom.xml b/seatunnel-connectors-v2/connector-http/pom.xml index 92286f0525..d190af554b 100644 --- a/seatunnel-connectors-v2/connector-http/pom.xml +++ b/seatunnel-connectors-v2/connector-http/pom.xml @@ -50,6 +50,7 @@ <module>connector-http-paypal</module> <module>connector-http-linear</module> <module>connector-http-splunk</module> + <module>connector-http-asana</module> <module>connector-mem0</module> </modules> diff --git a/seatunnel-dist/pom.xml b/seatunnel-dist/pom.xml index 525b0bf2d3..007279589b 100644 --- a/seatunnel-dist/pom.xml +++ b/seatunnel-dist/pom.xml @@ -248,6 +248,12 @@ <version>${project.version}</version> <scope>provided</scope> </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-asana</artifactId> + <version>${project.version}</version> + <scope>provided</scope> + </dependency> <dependency> <groupId>org.apache.seatunnel</groupId> <artifactId>connector-kafka</artifactId> diff --git a/seatunnel-connectors-v2/connector-http/pom.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-asana-e2e/pom.xml similarity index 50% copy from seatunnel-connectors-v2/connector-http/pom.xml copy to seatunnel-e2e/seatunnel-connector-v2-e2e/connector-asana-e2e/pom.xml index 92286f0525..4d809ed688 100644 --- a/seatunnel-connectors-v2/connector-http/pom.xml +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-asana-e2e/pom.xml @@ -1,56 +1,53 @@ <?xml version="1.0" encoding="UTF-8"?> <!-- - Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at - http://www.apache.org/licenses/LICENSE-2.0 - Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. - --> <project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.apache.seatunnel</groupId> - <artifactId>seatunnel-connectors-v2</artifactId> + <artifactId>seatunnel-connector-v2-e2e</artifactId> <version>${revision}</version> </parent> - <artifactId>connector-http</artifactId> - <packaging>pom</packaging> - <name>SeaTunnel : Connectors V2 : Http :</name> - - <modules> - <module>connector-http-base</module> - <module>connector-http-feishu</module> - <module>connector-http-wechat</module> - <module>connector-http-myhours</module> - <module>connector-http-lemlist</module> - <module>connector-http-klaviyo</module> - <module>connector-http-onesignal</module> - <module>connector-http-jira</module> - <module>connector-http-gitlab</module> - <module>connector-http-github</module> - <module>connector-http-notion</module> - <module>connector-http-persistiq</module> - <module>connector-http-airtable</module> - <module>connector-http-posthog</module> - <module>connector-http-shopify</module> - <module>connector-http-zendesk</module> - <module>connector-http-stripe</module> - <module>connector-http-paypal</module> - <module>connector-http-linear</module> - <module>connector-http-splunk</module> - <module>connector-mem0</module> - </modules> + <artifactId>connector-asana-e2e</artifactId> + <name>SeaTunnel : E2E : Connector V2 : Asana</name> + <dependencies> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-asana</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-base</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-assert</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-common</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> + </dependencies> </project> diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-asana-e2e/src/test/java/org/apache/seatunnel/e2e/connector/asana/AsanaIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-asana-e2e/src/test/java/org/apache/seatunnel/e2e/connector/asana/AsanaIT.java new file mode 100644 index 0000000000..fa12fccf06 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-asana-e2e/src/test/java/org/apache/seatunnel/e2e/connector/asana/AsanaIT.java @@ -0,0 +1,118 @@ +/* + * 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.asana; + +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.BeforeEach; +import org.junit.jupiter.api.TestTemplate; +import org.testcontainers.Testcontainers; +import org.testcontainers.containers.Container; + +import com.sun.net.httpserver.HttpServer; + +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.concurrent.atomic.AtomicInteger; + +public class AsanaIT extends TestSuiteBase implements TestResource { + + private static HttpServer httpServer; + private static int serverPort; + private static final AtomicInteger requestCount = new AtomicInteger(0); + + @BeforeAll + @Override + public void startUp() throws IOException { + httpServer = HttpServer.create(new InetSocketAddress(0), 0); + serverPort = httpServer.getAddress().getPort(); + Testcontainers.exposeHostPorts(serverPort); + + httpServer.createContext( + "/tasks", + exchange -> { + int currentReq = requestCount.incrementAndGet(); + String responseBody; + int statusCode; + + if (currentReq == 1) { + statusCode = 429; + responseBody = "{\"error\": \"Rate limit exceeded\"}"; + } else if (currentReq == 2) { + statusCode = 200; + responseBody = + "{" + + "\"data\": [" + + " {\"gid\": \"1\", \"name\": \"Task 1\", \"completed\": false, \"assignee\": {\"gid\": \"a1\", \"name\": \"Alice\"}}," + + " {\"gid\": \"2\", \"name\": \"Task 2\", \"completed\": true, \"assignee\": null}" + + "]," + + "\"next_page\": {\"offset\": \"ab.c_-==%2B\"}" + + "}"; + } else { + statusCode = 200; + responseBody = + "{" + + "\"data\": [" + + " {\"gid\": \"3\", \"name\": \"Task 3\", \"completed\": false, \"assignee\": null}" + + "]," + + "\"next_page\": null" + + "}"; + } + + byte[] bytes = responseBody.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "application/json"); + exchange.sendResponseHeaders(statusCode, bytes.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(bytes); + } + }); + + httpServer.start(); + } + + @AfterAll + @Override + public void tearDown() { + if (httpServer != null) { + httpServer.stop(0); + } + } + + @BeforeEach + public void resetCounter() { + requestCount.set(0); + } + + @TestTemplate + public void testAsanaSourceE2E(TestContainer container) + throws IOException, InterruptedException { + Container.ExecResult execResult = + container.executeJob( + "/asana_source_to_assert.conf", + Collections.singletonList("asana.test.port=" + serverPort)); + Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr()); + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-asana-e2e/src/test/resources/asana_source_to_assert.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-asana-e2e/src/test/resources/asana_source_to_assert.conf new file mode 100644 index 0000000000..bf78628059 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-asana-e2e/src/test/resources/asana_source_to_assert.conf @@ -0,0 +1,57 @@ +# 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 { + Asana { + base_url = "http://host.testcontainers.internal:${asana.test.port}"; + api_key = "placeholder-pat-token-12345" + project_gid = "123456789" + retry_backoff_multiplier_ms = 10 + retry_backoff_max_ms = 50 + } +} + +sink { + Assert { + rules { + row_rules = [ + { + rule_type = "MAX_ROW" + rule_value = 3 + }, + { + rule_type = "MIN_ROW" + rule_value = 3 + } + ] + field_rules = [ + { + field_name = "gid" + field_type = "string" + field_value = [ + { + rule_type = "NOT_NULL" + } + ] + } + ] + } + } +} \ No newline at end of file diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml index b42a2f3ce9..cf534b2972 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml @@ -51,6 +51,7 @@ <module>connector-file-gcs-e2e</module> <module>connector-cassandra-e2e</module> <module>connector-neo4j-e2e</module> + <module>connector-asana-e2e</module> <module>connector-http-e2e</module> <module>connector-paypal-e2e</module> <module>connector-tiktok-ads-e2e</module>
