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-12844-449225292833a7f8c5e0b0a9169a28b855245c86 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 75f9636c99a7957bf1819e8d0855a5fe92d3d3ee Author: Rishavbhattarai <[email protected]> AuthorDate: Sun Oct 18 13:27:45 2026 +0000 [Feature][Connector-V2] Add Typeform source connector (#12844) Signed-off-by: Rishavbhattarai <[email protected]> Co-authored-by: Claude Opus 5.5 <[email protected]> --- config/plugin_config | 1 + .../changelog/connector-http-typeform.md | 3 + docs/en/connectors/source/Typeform.md | 108 +++++++++ .../changelog/connector-http-typeform.md | 3 + docs/zh/connectors/source/Typeform.md | 80 +++++++ plugin-mapping.properties | 1 + .../connector-http/connector-http-typeform/pom.xml | 51 +++++ .../seatunnel/typeform/TypeformSource.java | 120 ++++++++++ .../seatunnel/typeform/TypeformSourceFactory.java | 70 ++++++ .../seatunnel/typeform/TypeformSourceReader.java | 126 ++++++++++ .../typeform/config/TypeformSourceOptions.java | 77 +++++++ .../typeform/config/TypeformSourceParameter.java | 115 ++++++++++ .../typeform/TypeformSourceFactoryTest.java | 157 +++++++++++++ .../seatunnel/typeform/TypeformSourceReadTest.java | 253 +++++++++++++++++++++ .../typeform/TypeformSourceReaderTest.java | 208 +++++++++++++++++ seatunnel-connectors-v2/connector-http/pom.xml | 1 + seatunnel-dist/pom.xml | 6 + .../connector-typeform-e2e/pom.xml | 53 +++++ .../e2e/connector/typeform/TypeformIT.java | 170 ++++++++++++++ .../test/resources/typeform_source_to_assert.conf | 78 +++++++ seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml | 1 + 21 files changed, 1682 insertions(+) diff --git a/config/plugin_config b/config/plugin_config index b4dfc44e62..07fd5a7a43 100644 --- a/config/plugin_config +++ b/config/plugin_config @@ -73,6 +73,7 @@ connector-http-base connector-mem0 connector-http-asana connector-http-intercom +connector-http-typeform connector-http-feishu connector-http-gitlab connector-http-github diff --git a/docs/en/connectors/changelog/connector-http-typeform.md b/docs/en/connectors/changelog/connector-http-typeform.md new file mode 100644 index 0000000000..fafa5140f5 --- /dev/null +++ b/docs/en/connectors/changelog/connector-http-typeform.md @@ -0,0 +1,3 @@ +| Change | Commit | Version | +| --- | --- | --- | +| [Feature][connector-http-typeform] Add Typeform source connector | https://github.com/apache/seatunnel | dev | \ No newline at end of file diff --git a/docs/en/connectors/source/Typeform.md b/docs/en/connectors/source/Typeform.md new file mode 100644 index 0000000000..8b596ab210 --- /dev/null +++ b/docs/en/connectors/source/Typeform.md @@ -0,0 +1,108 @@ +import ChangeLog from '../changelog/connector-http-typeform.md'; + +# Http-Typeform + +> Typeform source connector + +## Description + +Reads the responses of one form from the Typeform Responses API +(`GET /forms/{form_id}/responses`). Built on `connector-http-base`. Batch mode only. Version 1 +reads responses only; forms, workspaces, webhooks, OAuth and flattening `answers` into columns are +not supported. + +## Key Features + +- [x] Batch +- [ ] Stream +- [ ] Exactly-once +- [ ] Column projection +- [ ] Parallelism + +## Options + +| Name | Type | Required | Default | Description | +| --- | --- | --- | --- | --- | +| api_key | String | Yes | - | Typeform personal access token with the `responses:read` scope (raw token, no `Bearer ` prefix). | +| form_id | String | Yes | - | ID of the form to read. Letters, digits, `_` and `-` only. | +| base_url | String | No | `https://api.typeform.com` | Accounts on Typeform's EU data center use `https://api.eu.typeform.com` or `https://api.typeform.eu`. | +| page_size | Int | No | `100` | Responses per page, 1 to 1000 (Typeform's maximum). | +| since | String | No | - | Only responses since this time, inclusive. Unix seconds or ISO 8601 UTC, for example `2026-01-01T00:00:00`. | +| until | String | No | - | Only responses until this time, inclusive. Same format as `since`. | +| response_type | String | No | - | Comma-separated, for example `partial,completed`. Typeform returns completed responses when this is not set. | +| 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 +Sent only in the `Authorization: Bearer <token>` header, never in query parameters. The connector +never logs headers or the config, and persists no state, so the token does not reach checkpoints. +Whether the engine masks `api_key` when it logs the job config is outside this connector. + +### since / until / response_type +Passed to Typeform as is. For completed responses Typeform filters `since` and `until` on +`submitted_at`; for other response types it uses a different timestamp (see Typeform's API +reference). + +### retry +Requests are spaced at least 500 ms apart, because Typeform allows 2 requests per second per +account. HTTP 429 and 5xx are retried with a backoff that doubles from +`retry_backoff_multiplier_ms` up to `retry_backoff_max_ms`. `Retry-After` is not honored. + +## Output columns + +| Column | Type | Notes | +| --- | --- | --- | +| response_id | STRING | Typeform response id | +| landing_id | STRING | | +| token | STRING | Used as the paging cursor | +| landed_at | STRING | ISO 8601 timestamp | +| submitted_at | STRING | ISO 8601 timestamp; null for partial responses | +| answers | STRING | The `answers` array as a JSON string, unchanged. NULL when the response has no answers (empty array or missing). | +| hidden | STRING | The `hidden` fields object as a JSON string, unchanged. `{}` when the form has no hidden fields; NULL when missing. | + +Answer shapes differ by question type, so `answers` is kept as raw JSON. Use a JSON transform or +the sink's JSON functions to extract the fields you need. + +## Example + +```hocon +env { + parallelism = 1 + job.mode = BATCH +} + +source { + Typeform { + api_key = "<PERSONAL_ACCESS_TOKEN>" + form_id = "<FORM_ID>" + since = "2026-01-01T00:00:00" + plugin_output = "typeform_responses" + } +} + +sink { + Console { + plugin_input = "typeform_responses" + } +} +``` + +## Notes + +Each run is a bounded snapshot of the form's responses, optionally narrowed by `since` and +`until`. Pages are read newest first: the `token` of the last response on a page is sent as the +`before` cursor, and an empty page ends the read. Responses submitted during the read arrive at the +newest end and do not shift the pages being read. Sort order is not configurable, because the +`before` cursor depends on Typeform's default order. + +No incremental state is stored. After a failure, a restarted job re-reads from the first page and +may emit rows that were already sent. Every row includes `response_id`, which a downstream sink can +use to deduplicate or upsert, but the connector provides no delivery guarantee. + +Typeform notes that responses from roughly the last 30 minutes may not be returned yet. + +The whole form is read in one uninterrupted pass, with a short pause between requests, so on a very +large form checkpoint triggering waits until the read finishes. + +<ChangeLog /> diff --git a/docs/zh/connectors/changelog/connector-http-typeform.md b/docs/zh/connectors/changelog/connector-http-typeform.md new file mode 100644 index 0000000000..fafa5140f5 --- /dev/null +++ b/docs/zh/connectors/changelog/connector-http-typeform.md @@ -0,0 +1,3 @@ +| Change | Commit | Version | +| --- | --- | --- | +| [Feature][connector-http-typeform] Add Typeform source connector | https://github.com/apache/seatunnel | dev | \ No newline at end of file diff --git a/docs/zh/connectors/source/Typeform.md b/docs/zh/connectors/source/Typeform.md new file mode 100644 index 0000000000..0d060b5d65 --- /dev/null +++ b/docs/zh/connectors/source/Typeform.md @@ -0,0 +1,80 @@ +import ChangeLog from '../changelog/connector-http-typeform.md'; + +# Http-Typeform + +> Typeform 源连接器 + +## 描述 + +通过 Typeform Responses API（`GET /forms/{form_id}/responses`）读取一个表单的回复。基于 +`connector-http-base` 构建，仅支持批处理模式。第一版只读取回复，不支持表单、工作区、Webhook、OAuth， +也不会把 `answers` 展开成多列。 + +## 主要特性 + +- [x] 批处理 +- [ ] 流处理 +- [ ] 精确一次 +- [ ] 列投影 +- [ ] 并行度 + +## 选项 + +| 名称 | 类型 | 必填 | 默认值 | 描述 | +| --- | --- | --- | --- | --- | +| api_key | String | 是 | - | 具有 `responses:read` 权限的 Typeform 个人访问令牌（原始令牌，不带 `Bearer ` 前缀）。 | +| form_id | String | 是 | - | 要读取的表单 ID，只能包含字母、数字、`_` 和 `-`。 | +| base_url | String | 否 | `https://api.typeform.com` | 使用 Typeform 欧盟数据中心的账号请使用 `https://api.eu.typeform.com` 或 `https://api.typeform.eu`。 | +| page_size | Int | 否 | `100` | 每页回复数量，范围 1 到 1000。 | +| since | String | 否 | - | 只读取该时间（含）之后的回复。Unix 秒或 ISO 8601 UTC，例如 `2026-01-01T00:00:00`。 | +| until | String | 否 | - | 只读取该时间（含）之前的回复，格式同 `since`。 | +| response_type | String | 否 | - | 逗号分隔，例如 `partial,completed`。未设置时 Typeform 只返回已完成的回复。 | +| retry | Int | 否 | `3` | HTTP 429 和 5xx 的最大重试次数。 | +| retry_backoff_multiplier_ms | Int | 否 | `100` | 退避倍数（毫秒）。 | +| retry_backoff_max_ms | Int | 否 | `10000` | 最大退避时间（毫秒）。 | + +令牌只放在 `Authorization: Bearer <token>` 请求头中，不会出现在查询参数、日志或作业状态里。 +Typeform 每个账号每秒允许 2 个请求，因此连接器会让请求间隔至少 500 毫秒。 + +## 输出列 + +`response_id`、`landing_id`、`token`、`landed_at`、`submitted_at`（均为 STRING），以及 +`answers`、`hidden`（原样输出的 JSON 字符串）。回复没有答案时（空数组或缺失），`answers` 为 NULL； +表单没有隐藏字段时 `hidden` 为 `{}`。 + +## 示例 + +```hocon +env { + parallelism = 1 + job.mode = BATCH +} + +source { + Typeform { + api_key = "<PERSONAL_ACCESS_TOKEN>" + form_id = "<FORM_ID>" + since = "2026-01-01T00:00:00" + plugin_output = "typeform_responses" + } +} + +sink { + Console { + plugin_input = "typeform_responses" + } +} +``` + +## 注意事项 + +每次运行都是表单回复的有界快照，可用 `since` 和 `until` 缩小范围。分页从最新的回复开始：把每页最后一条回复的 +`token` 作为 `before` 游标发送，遇到空页即结束。读取期间新提交的回复出现在最新的一端，不会影响正在读取的分页。 +排序方式不可配置，因为 `before` 游标依赖 Typeform 的默认排序。 + +连接器不保存增量状态。作业失败后重启会从第一页重新读取，可能重复发送已发送的行。每行都包含 `response_id`， +下游可据此去重或 upsert，但连接器不提供投递保证。Typeform 提示最近约 30 分钟内的回复可能尚未返回。 + +整个表单在一次不中断的读取中完成，请求之间有短暂间隔，因此表单很大时，触发 checkpoint 需要等到读取结束。 + +<ChangeLog /> diff --git a/plugin-mapping.properties b/plugin-mapping.properties index f47a5e41a0..5df09428b5 100644 --- a/plugin-mapping.properties +++ b/plugin-mapping.properties @@ -209,6 +209,7 @@ seatunnel.source.Splunk = connector-http-splunk seatunnel.source.WebSocket = connector-websocket seatunnel.source.Asana = connector-http-asana seatunnel.source.Intercom = connector-http-intercom +seatunnel.source.Typeform = connector-http-typeform # For custom transforms, make sure to use the seatunnel.transform.[PluginIdentifier]=[JarPerfix] naming convention. For example: # seatunnel.transform.Sql = seatunnel-transforms-v2 diff --git a/seatunnel-connectors-v2/connector-http/connector-http-typeform/pom.xml b/seatunnel-connectors-v2/connector-http/connector-http-typeform/pom.xml new file mode 100644 index 0000000000..2b36ba5339 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-typeform/pom.xml @@ -0,0 +1,51 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http</artifactId> + <version>${revision}</version> + </parent> + + <artifactId>connector-http-typeform</artifactId> + <name>SeaTunnel : Connectors V2 : Http : Typeform</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-typeform/src/main/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSource.java b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/main/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSource.java new file mode 100644 index 0000000000..bbc899efe9 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/main/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSource.java @@ -0,0 +1,120 @@ +/* + * 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.typeform; + +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.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.connectors.seatunnel.typeform.config.TypeformSourceParameter; +import org.apache.seatunnel.format.json.JsonDeserializationSchema; + +import lombok.extern.slf4j.Slf4j; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * Batch source that reads one form's responses from the Typeform Responses API. Pages are read + * newest first by passing the token of the last response on a page as the {@code before} cursor. + * {@code answers} and {@code hidden} are emitted as raw JSON strings. + */ +@Slf4j +public class TypeformSource extends HttpSource { + static final String[] COLUMNS = { + "response_id", "landing_id", "token", "landed_at", "submitted_at", "answers", "hidden" + }; + + private final TypeformSourceParameter typeformSourceParameter = new TypeformSourceParameter(); + + public TypeformSource(ReadonlyConfig pluginConfig) { + super(pluginConfig); + typeformSourceParameter.buildWithConfig(pluginConfig); + + TableSchema.Builder schema = TableSchema.builder(); + Map<String, String> fields = new LinkedHashMap<>(); + for (String column : COLUMNS) { + fields.put(column, "$.items[*]." + column); + schema.column(PhysicalColumn.of(column, BasicType.STRING_TYPE, 0, true, null, null)); + } + this.catalogTable = + CatalogTable.of( + TableIdentifier.of("Typeform", 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("before"); + // The token of the last response on the page; an empty page yields no value and ends the + // read. + info.setPageCursorResponseField("$.items[-1:].token"); + 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 TypeformSourceReader( + this.typeformSourceParameter, + readerContext, + this.deserializationSchema, + jsonField, + contentField, + pageInfo); + } + + @Override + public String getPluginName() { + return "Typeform"; + } + + @Override + public Boundedness getBoundedness() { + if (JobMode.STREAMING.equals(jobContext.getJobMode())) { + throw new HttpConnectorException( + CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT, + "Typeform source only supports batch mode."); + } + return Boundedness.BOUNDED; + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/main/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSourceFactory.java b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/main/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSourceFactory.java new file mode 100644 index 0000000000..647471965a --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/main/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSourceFactory.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.typeform; + +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.http.source.HttpSourceFactory; +import org.apache.seatunnel.connectors.seatunnel.typeform.config.TypeformSourceOptions; + +import com.google.auto.service.AutoService; + +import java.io.Serializable; + +/** + * Factory that registers the Typeform source under the identifier "Typeform" and declares its + * options. + */ +@AutoService(Factory.class) +public class TypeformSourceFactory extends HttpSourceFactory { + @Override + public String factoryIdentifier() { + return "Typeform"; + } + + @Override + public <T, SplitT extends SourceSplit, StateT extends Serializable> + TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) { + return () -> (SeaTunnelSource<T, SplitT, StateT>) new TypeformSource(context.getOptions()); + } + + @Override + public OptionRule optionRule() { + return OptionRule.builder() + .required(TypeformSourceOptions.API_KEY, TypeformSourceOptions.FORM_ID) + .optional( + TypeformSourceOptions.BASE_URL, + TypeformSourceOptions.PAGE_SIZE, + TypeformSourceOptions.SINCE, + TypeformSourceOptions.UNTIL, + TypeformSourceOptions.RESPONSE_TYPE, + TypeformSourceOptions.RETRY, + TypeformSourceOptions.RETRY_BACKOFF_MULTIPLIER_MS, + TypeformSourceOptions.RETRY_BACKOFF_MAX_MS) + .build(); + } + + @Override + public Class<? extends SeaTunnelSource> getSourceClass() { + return TypeformSource.class; + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/main/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSourceReader.java b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/main/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSourceReader.java new file mode 100644 index 0000000000..80336c0eb6 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/main/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSourceReader.java @@ -0,0 +1,126 @@ +/* + * 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.typeform; + +import org.apache.seatunnel.api.serialization.DeserializationSchema; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +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.HttpConnectorErrorCode; +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 keep requests at least 500 ms apart (Typeform allows + * two requests per second per account) and to retry HTTP 429 and 5xx with exponential backoff + * before handing the response to the base reader. + */ +@Slf4j +public class TypeformSourceReader extends HttpSourceReader { + + static final long DEFAULT_MIN_REQUEST_INTERVAL_MILLIS = 500; + + private long minRequestIntervalMillis = DEFAULT_MIN_REQUEST_INTERVAL_MILLIS; + private long lastRequestNanos = -1; + + public TypeformSourceReader( + HttpParameter httpParameter, + SingleSplitReaderContext readerContext, + DeserializationSchema<SeaTunnelRow> deserializationSchema, + JsonField jsonField, + String contentField, + PageInfo pageInfo) { + super( + httpParameter, + readerContext, + deserializationSchema, + jsonField, + contentField, + pageInfo); + } + + /** Lets tests shorten the pacing interval. */ + void setMinRequestIntervalMillis(long minRequestIntervalMillis) { + this.minRequestIntervalMillis = minRequestIntervalMillis; + } + + @Override + protected HttpResponse executeRequest() throws Exception { + int maxRetries = httpParameter.getRetry(); + long waitMillis = httpParameter.getRetryBackoffMultiplierMillis(); + long maxWaitMillis = httpParameter.getRetryBackoffMaxMillis(); + int attempt = 1; + while (true) { + waitForRateLimit(); + 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( + HttpConnectorErrorCode.REQUEST_FAILED, + String.format( + "Typeform request failed with HTTP %d after %d attempt(s) (retry=%d)", + code, attempt, maxRetries)); + } + log.warn( + "Typeform 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 Typeform HTTP %d (attempt %d)", + code, attempt), + e); + } + waitMillis = Math.min(waitMillis * 2, maxWaitMillis); + attempt++; + } + } + + private void waitForRateLimit() { + if (lastRequestNanos >= 0) { + long elapsedMillis = (System.nanoTime() - lastRequestNanos) / 1_000_000L; + long remaining = minRequestIntervalMillis - elapsedMillis; + if (remaining > 0) { + try { + Thread.sleep(remaining); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException( + "Interrupted while pacing requests to the Typeform API", e); + } + } + } + lastRequestNanos = System.nanoTime(); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/main/java/org/apache/seatunnel/connectors/seatunnel/typeform/config/TypeformSourceOptions.java b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/main/java/org/apache/seatunnel/connectors/seatunnel/typeform/config/TypeformSourceOptions.java new file mode 100644 index 0000000000..c170be85c0 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/main/java/org/apache/seatunnel/connectors/seatunnel/typeform/config/TypeformSourceOptions.java @@ -0,0 +1,77 @@ +/* + * 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.typeform.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 Typeform source, on top of the shared HTTP options. */ +public class TypeformSourceOptions extends HttpCommonOptions { + public static final Option<String> API_KEY = + Options.key("api_key") + .stringType() + .noDefaultValue() + .withDescription( + "Typeform personal access token with the responses:read scope (raw" + + " token, no Bearer prefix)"); + + public static final Option<String> FORM_ID = + Options.key("form_id") + .stringType() + .noDefaultValue() + .withDescription("ID of the form whose responses are read"); + + public static final Option<String> BASE_URL = + Options.key("base_url") + .stringType() + .defaultValue("https://api.typeform.com";) + .withDescription( + "Typeform API base URL. Accounts on Typeform's EU data center use" + + " https://api.eu.typeform.com or https://api.typeform.eu.";); + + public static final Option<Integer> PAGE_SIZE = + Options.key("page_size") + .intType() + .defaultValue(100) + .withDescription("Responses requested per page (1 to 1000)"); + + public static final Option<String> SINCE = + Options.key("since") + .stringType() + .noDefaultValue() + .withDescription( + "Only read responses since this time, inclusive. Unix seconds or ISO" + + " 8601 UTC, passed to Typeform as is."); + + public static final Option<String> UNTIL = + Options.key("until") + .stringType() + .noDefaultValue() + .withDescription( + "Only read responses until this time, inclusive. Unix seconds or ISO" + + " 8601 UTC, passed to Typeform as is."); + + public static final Option<String> RESPONSE_TYPE = + Options.key("response_type") + .stringType() + .noDefaultValue() + .withDescription( + "Comma-separated response types, for example partial,completed." + + " Typeform returns completed responses when this is not set."); +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/main/java/org/apache/seatunnel/connectors/seatunnel/typeform/config/TypeformSourceParameter.java b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/main/java/org/apache/seatunnel/connectors/seatunnel/typeform/config/TypeformSourceParameter.java new file mode 100644 index 0000000000..b9f3531bc1 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/main/java/org/apache/seatunnel/connectors/seatunnel/typeform/config/TypeformSourceParameter.java @@ -0,0 +1,115 @@ +/* + * 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.typeform.config; + +import org.apache.seatunnel.api.configuration.Option; +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; +import java.util.Optional; +import java.util.regex.Pattern; + +/** + * Builds the HTTP request parameters for the Typeform responses endpoint: the responses URL for one + * form, the Bearer Authorization header, the page size, the optional filters and the retry + * defaults. + */ +public class TypeformSourceParameter extends HttpParameter { + + private static final int MAX_PAGE_SIZE = 1000; + private static final Pattern FORM_ID_PATTERN = Pattern.compile("[A-Za-z0-9_-]+"); + + private static String requireNonBlank(String value, String optionKey) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException( + String.format("Typeform option '%s' must not be blank.", optionKey)); + } + return value.trim(); + } + + /** + * Builds the request for {@code GET {base_url}/forms/{form_id}/responses}. Validates that + * api_key, form_id and base_url are not blank, that form_id contains only letters, digits, '_' + * and '-', that page_size is within 1..1000 and that the retry backoff values are not negative. + * Passes since, until and response_type to Typeform only when they are set. Retry defaults to 3 + * only when the user did not set {@code retry}. Warns when the bearer token would be sent over + * a non-HTTPS base_url. + */ + public void buildWithConfig(ReadonlyConfig config) { + super.buildWithConfig(config); + String apiKey = requireNonBlank(config.get(TypeformSourceOptions.API_KEY), "api_key"); + String formId = requireNonBlank(config.get(TypeformSourceOptions.FORM_ID), "form_id"); + if (!FORM_ID_PATTERN.matcher(formId).matches()) { + throw new IllegalArgumentException( + "Typeform option 'form_id' may only contain letters, digits, '_' and '-'."); + } + int pageSize = config.get(TypeformSourceOptions.PAGE_SIZE); + if (pageSize < 1 || pageSize > MAX_PAGE_SIZE) { + throw new IllegalArgumentException( + String.format( + "Typeform option 'page_size' must be between 1 and %d.", + MAX_PAGE_SIZE)); + } + + String base = requireNonBlank(config.get(TypeformSourceOptions.BASE_URL), "base_url"); + if (base.endsWith("/")) { + base = base.substring(0, base.length() - 1); + } + setUrl(base + "/forms/" + formId + "/responses"); + setMethod(HttpRequestMethod.GET); + + if (headers == null) { + headers = new HashMap<>(); + } + headers.put("Authorization", "Bearer " + apiKey); + headers.put("Accept", "application/json"); + // The base class runs this check before the URL and Authorization header exist, so run it + // again now that both are set. + validateCredentialScheme(); + + params = new HashMap<>(); + params.put("page_size", String.valueOf(pageSize)); + putIfSet(config, TypeformSourceOptions.SINCE); + putIfSet(config, TypeformSourceOptions.UNTIL); + putIfSet(config, TypeformSourceOptions.RESPONSE_TYPE); + setKeepPageParamAsHttpParam(true); + setJsonFiledMissedReturnNull(true); + + if (!config.getOptional(TypeformSourceOptions.RETRY).isPresent()) { + setRetry(3); + } + setRetryBackoffMultiplierMillis( + config.get(TypeformSourceOptions.RETRY_BACKOFF_MULTIPLIER_MS)); + setRetryBackoffMaxMillis(config.get(TypeformSourceOptions.RETRY_BACKOFF_MAX_MS)); + + if (getRetryBackoffMultiplierMillis() < 0 || getRetryBackoffMaxMillis() < 0) { + throw new IllegalArgumentException( + "Typeform retry backoff values must not be negative."); + } + } + + /** Sends an optional filter to Typeform under its own option name, only when it is set. */ + private void putIfSet(ReadonlyConfig config, Option<String> option) { + Optional<String> value = config.getOptional(option); + if (value.isPresent() && !value.get().trim().isEmpty()) { + params.put(option.key(), value.get().trim()); + } + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/test/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSourceFactoryTest.java b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/test/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSourceFactoryTest.java new file mode 100644 index 0000000000..db59e05bbd --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/test/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSourceFactoryTest.java @@ -0,0 +1,157 @@ +/* + * 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.typeform; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.connectors.seatunnel.http.config.HttpRequestMethod; +import org.apache.seatunnel.connectors.seatunnel.typeform.config.TypeformSourceParameter; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.Map; + +public class TypeformSourceFactoryTest { + + private static Map<String, Object> baseConfig() { + Map<String, Object> m = new HashMap<>(); + m.put("api_key", "tfp_test-typeform-token-123"); + m.put("form_id", "lT9Z2j"); + return m; + } + + private static TypeformSourceParameter build(Map<String, Object> m) { + TypeformSourceParameter parameter = new TypeformSourceParameter(); + parameter.buildWithConfig(ReadonlyConfig.fromMap(m)); + return parameter; + } + + private static void assertRejected(Map<String, Object> m, String optionName) { + Exception e = Assertions.assertThrows(IllegalArgumentException.class, () -> build(m)); + Assertions.assertTrue(e.getMessage().contains(optionName), e.getMessage()); + Assertions.assertFalse(e.getMessage().contains("tfp_test-typeform-token-123")); + } + + @Test + public void testFactoryIdentifier() { + Assertions.assertEquals("Typeform", new TypeformSourceFactory().factoryIdentifier()); + Assertions.assertEquals( + "Typeform", + new TypeformSource(ReadonlyConfig.fromMap(baseConfig())).getPluginName()); + } + + @Test + public void testParameterBuildingAndHeaders() { + TypeformSourceParameter parameter = build(baseConfig()); + + Assertions.assertEquals( + "Bearer tfp_test-typeform-token-123", parameter.getHeaders().get("Authorization")); + Assertions.assertEquals(HttpRequestMethod.GET, parameter.getMethod()); + Assertions.assertEquals( + "https://api.typeform.com/forms/lT9Z2j/responses";, parameter.getUrl()); + Assertions.assertEquals("100", parameter.getParams().get("page_size")); + Assertions.assertEquals(1, parameter.getParams().size(), "no filters unless set"); + Assertions.assertFalse(parameter.getParams().containsValue("tfp_test-typeform-token-123")); + Assertions.assertEquals(3, parameter.getRetry()); + } + + @Test + public void testEuBaseUrlPageSizeAndFilters() { + Map<String, Object> m = baseConfig(); + m.put("base_url", "https://api.eu.typeform.com/";); + m.put("page_size", 1000); + m.put("since", "2026-01-01T00:00:00"); + m.put("until", "1767225600"); + m.put("response_type", "partial,completed"); + TypeformSourceParameter parameter = build(m); + + Assertions.assertEquals( + "https://api.eu.typeform.com/forms/lT9Z2j/responses";, parameter.getUrl()); + Assertions.assertEquals("1000", parameter.getParams().get("page_size")); + Assertions.assertEquals("2026-01-01T00:00:00", parameter.getParams().get("since")); + Assertions.assertEquals("1767225600", parameter.getParams().get("until")); + Assertions.assertEquals("partial,completed", parameter.getParams().get("response_type")); + } + + @Test + public void testBlankFilterIsNotSent() { + Map<String, Object> m = baseConfig(); + m.put("since", " "); + Assertions.assertFalse(build(m).getParams().containsKey("since")); + } + + @Test + public void testBlankApiKeyRejected() { + Map<String, Object> m = baseConfig(); + m.put("api_key", " "); + assertRejected(m, "api_key"); + } + + @Test + public void testBlankFormIdRejected() { + Map<String, Object> m = baseConfig(); + m.put("form_id", ""); + assertRejected(m, "form_id"); + } + + @Test + public void testFormIdWithPathCharactersRejected() { + for (String bad : new String[] {"../me", "abc/def", "abc?x=1", "a b"}) { + Map<String, Object> m = baseConfig(); + m.put("form_id", bad); + assertRejected(m, "form_id"); + } + } + + @Test + public void testPageSizeOutOfRangeRejected() { + for (int bad : new int[] {0, 1001}) { + Map<String, Object> m = baseConfig(); + m.put("page_size", bad); + assertRejected(m, "page_size"); + } + } + + @Test + public void testNegativeBackoffRejected() { + Map<String, Object> m = baseConfig(); + m.put("retry_backoff_multiplier_ms", -1); + Assertions.assertThrows(IllegalArgumentException.class, () -> build(m)); + } + + @Test + public void testCustomBackoffHonoredWithoutExplicitRetry() { + Map<String, Object> m = baseConfig(); + m.put("retry_backoff_multiplier_ms", 2000); + m.put("retry_backoff_max_ms", 3000); + TypeformSourceParameter parameter = build(m); + + Assertions.assertEquals(3, parameter.getRetry()); + Assertions.assertEquals(2000, parameter.getRetryBackoffMultiplierMillis()); + Assertions.assertEquals(3000, parameter.getRetryBackoffMaxMillis()); + } + + @Test + public void testDefaultBackoffWhenNothingConfigured() { + TypeformSourceParameter parameter = build(baseConfig()); + + Assertions.assertEquals(100, parameter.getRetryBackoffMultiplierMillis()); + Assertions.assertEquals(10000, parameter.getRetryBackoffMaxMillis()); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/test/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSourceReadTest.java b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/test/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSourceReadTest.java new file mode 100644 index 0000000000..25d1c9550e --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/test/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSourceReadTest.java @@ -0,0 +1,253 @@ +/* + * 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.typeform; + +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.ObjectMapper; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.source.Collector; +import org.apache.seatunnel.api.source.SourceReader; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext; + +import org.junit.jupiter.api.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.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +/** + * Reads Typeform-shaped pages through {@link TypeformSource} and the base HTTP reader, and checks + * the cursor, the filters and how {@code answers} and {@code hidden} are emitted. + */ +public class TypeformSourceReadTest { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + private static final int ANSWERS = 5; + private static final int HIDDEN = 6; + + // Answer shapes from Typeform's Retrieve Responses reference, plus escaping, unicode, a + // decimal and an integer above 2^53. + private static final String R1_ANSWERS = + "[{\"field\":{\"id\":\"SMEUb7VJz92Q\",\"ref\":\"my_text\",\"type\":\"short_text\"},\"type\":\"text\",\"text\":\"Lian \\\"quoted\\\"\\nnew line, caf\\u00e9 \\u65e5\\u672c\"}," + + "{\"field\":{\"id\":\"RUqkXSeXBXSd\",\"ref\":\"my_yes_no\",\"type\":\"yes_no\"},\"type\":\"boolean\",\"boolean\":false}," + + "{\"field\":{\"id\":\"JwWggjAKtOkA\",\"ref\":\"my_email\",\"type\":\"email\"},\"type\":\"email\",\"email\":\"[email protected]\"}," + + "{\"field\":{\"id\":\"Q7M2XAwY04dW\",\"ref\":\"my_number\",\"type\":\"number\"},\"type\":\"number\",\"number\":9007199254740993}," + + "{\"field\":{\"id\":\"dec1\",\"ref\":\"my_decimal\",\"type\":\"number\"},\"type\":\"number\",\"number\":3.14}," + + "{\"field\":{\"id\":\"KoJxDM3c6x8h\",\"ref\":\"my_date\",\"type\":\"date\"},\"type\":\"date\",\"date\":\"2012-03-20T00:00:00Z\"}," + + "{\"field\":{\"id\":\"PNe8ZKBK8C2Q\",\"ref\":\"my_multi\",\"type\":\"multiple_choice\"},\"type\":\"choices\",\"choices\":{\"labels\":[\"New York\",\"Tokyo\"]}}," + + "{\"field\":{\"id\":\"ceIXxpbP3t2q\",\"ref\":\"my_choice\",\"type\":\"multiple_choice\"},\"type\":\"choice\",\"choice\":{\"label\":\"Other\",\"other\":\"Kathmandu\"}}," + + "{\"field\":{\"id\":\"pn48RmPazVdM\",\"ref\":\"my_file\",\"type\":\"file_upload\"},\"type\":\"file_url\",\"file_url\":\"https://api.typeform.com/forms/lT9Z2j/responses/r1/fields/pn48/files/a.pdf\"},"; + + "{\"field\":{\"id\":\"pay1\",\"ref\":\"my_payment\",\"type\":\"payment\"},\"type\":\"payment\",\"payment\":{\"amount\":\"4.20\",\"last4\":\"4242\",\"name\":\"Lian\",\"success\":true}}]"; + private static final String R1_HIDDEN = + "{\"utm_source\":\"newsletter\",\"z_last\":\"z\",\"a_first\":\"a\",\"empty\":\"\"}"; + private static final String R5_ANSWERS = + "[{\"field\":{\"id\":\"m1\",\"type\":\"matrix\"},\"type\":\"future_type\",\"future_type\":{\"rows\":[[1,2],[3,null]],\"meta\":{}}}]"; + private static final String R5_HIDDEN = "{\"nested_like\":\"{\\\"not\\\":\\\"parsed\\\"}\"}"; + + /** A response item; {@code token} differs from {@code response_id} on purpose. */ + private static String item(String id, String answersJson, String hiddenJson) { + StringBuilder sb = new StringBuilder(); + sb.append("{\"landing_id\":\"l-") + .append(id) + .append("\",\"token\":\"tok-") + .append(id) + .append("\",\"response_id\":\"") + .append(id) + .append("\",\"landed_at\":\"2026-09-14T22:33:59Z\",") + .append("\"submitted_at\":\"2026-09-14T22:38:22Z\""); + if (answersJson != null) { + sb.append(",\"answers\":").append(answersJson); + } + if (hiddenJson != null) { + sb.append(",\"hidden\":").append(hiddenJson); + } + return sb.append("}").toString(); + } + + private static String page(String... items) { + return "{\"total_items\":5,\"page_count\":3,\"items\":[" + String.join(",", items) + "]}"; + } + + private static List<SeaTunnelRow> read(MockWebServer server, Map<String, Object> extra) + throws Exception { + Map<String, Object> m = new HashMap<>(); + m.put("api_key", "tfp_test-token"); + m.put("form_id", "lT9Z2j"); + m.put("page_size", 2); + m.put("base_url", server.url("/").toString()); + m.putAll(extra); + TypeformSource source = new TypeformSource(ReadonlyConfig.fromMap(m)); + + SourceReader.Context ctx = Mockito.mock(SourceReader.Context.class); + TypeformSourceReader reader = + (TypeformSourceReader) source.createReader(new SingleSplitReaderContext(ctx)); + reader.setMinRequestIntervalMillis(0); + + List<SeaTunnelRow> rows = new ArrayList<>(); + Collector<SeaTunnelRow> collector = + new Collector<SeaTunnelRow>() { + @Override + public void collect(SeaTunnelRow record) { + rows.add(record); + } + + @Override + public Object getCheckpointLock() { + return this; + } + }; + reader.open(); + try { + reader.pollNext(collector); + } finally { + reader.close(); + } + return rows; + } + + private static Map<String, SeaTunnelRow> byResponseId(List<SeaTunnelRow> rows) { + Map<String, SeaTunnelRow> map = new LinkedHashMap<>(); + for (SeaTunnelRow row : rows) { + map.put((String) row.getField(0), row); + } + return map; + } + + @Test + public void testPagingWithBeforeTokenAndJsonStrings() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.enqueue( + new MockResponse() + .setBody( + page( + item("r1", R1_ANSWERS, R1_HIDDEN), + item("r2", "[]", "{}")))); + server.enqueue( + new MockResponse() + .setBody(page(item("r3", "null", "null"), item("r4", null, null)))); + server.enqueue(new MockResponse().setBody(page(item("r5", R5_ANSWERS, R5_HIDDEN)))); + server.enqueue(new MockResponse().setBody(page())); + server.start(); + + List<SeaTunnelRow> rows = read(server, new HashMap<>()); + + // Paging: no cursor first, then the token (not the response_id) of the last item. + Assertions.assertEquals(4, server.getRequestCount()); + RecordedRequest first = server.takeRequest(5, TimeUnit.SECONDS); + Assertions.assertEquals("/forms/lT9Z2j/responses", first.getRequestUrl().encodedPath()); + Assertions.assertNull(first.getRequestUrl().queryParameter("before")); + Assertions.assertEquals("2", first.getRequestUrl().queryParameter("page_size")); + Assertions.assertEquals("Bearer tfp_test-token", first.getHeader("Authorization")); + Assertions.assertFalse(first.getRequestUrl().toString().contains("tfp_test-token")); + Assertions.assertEquals( + "tok-r2", + server.takeRequest(5, TimeUnit.SECONDS) + .getRequestUrl() + .queryParameter("before")); + Assertions.assertEquals( + "tok-r4", + server.takeRequest(5, TimeUnit.SECONDS) + .getRequestUrl() + .queryParameter("before")); + Assertions.assertEquals( + "tok-r5", + server.takeRequest(5, TimeUnit.SECONDS) + .getRequestUrl() + .queryParameter("before")); + + // Rows: five responses, the empty last page ends the read. + Assertions.assertEquals(5, rows.size()); + Map<String, SeaTunnelRow> byId = byResponseId(rows); + SeaTunnelRow r1 = byId.get("r1"); + Assertions.assertEquals("l-r1", r1.getField(1)); + Assertions.assertEquals("tok-r1", r1.getField(2)); + Assertions.assertEquals("2026-09-14T22:33:59Z", r1.getField(3)); + Assertions.assertEquals("2026-09-14T22:38:22Z", r1.getField(4)); + + // answers and hidden are the source JSON, byte for byte: key order, escaping, + // unicode, the decimal and the integer above 2^53 all survive. + Assertions.assertEquals( + MAPPER.writeValueAsString(MAPPER.readTree(R1_ANSWERS)), r1.getField(ANSWERS)); + Assertions.assertEquals(R1_HIDDEN, r1.getField(HIDDEN)); + + // Empty answers array becomes NULL (base reader behaviour); empty hidden stays "{}". + Assertions.assertNull(byId.get("r2").getField(ANSWERS)); + Assertions.assertEquals("{}", byId.get("r2").getField(HIDDEN)); + // Explicit nulls and missing keys become NULL. + Assertions.assertNull(byId.get("r3").getField(ANSWERS)); + Assertions.assertNull(byId.get("r3").getField(HIDDEN)); + Assertions.assertNull(byId.get("r4").getField(ANSWERS)); + Assertions.assertNull(byId.get("r4").getField(HIDDEN)); + // An unknown answer type passes through, and a JSON-looking string stays a string. + Assertions.assertEquals( + MAPPER.writeValueAsString(MAPPER.readTree(R5_ANSWERS)), + byId.get("r5").getField(ANSWERS)); + Assertions.assertEquals(R5_HIDDEN, byId.get("r5").getField(HIDDEN)); + } + } + + @Test + public void testFiltersAreSentOnEveryPage() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.enqueue(new MockResponse().setBody(page(item("r1", "[]", "{}")))); + server.enqueue(new MockResponse().setBody(page())); + server.start(); + + Map<String, Object> filters = new HashMap<>(); + filters.put("since", "2026-01-01T00:00:00"); + filters.put("until", "2026-12-31T23:59:59"); + filters.put("response_type", "partial,completed"); + List<SeaTunnelRow> rows = read(server, filters); + + Assertions.assertEquals(1, rows.size()); + Assertions.assertEquals(2, server.getRequestCount()); + for (int i = 0; i < 2; i++) { + RecordedRequest request = server.takeRequest(5, TimeUnit.SECONDS); + Assertions.assertEquals( + "2026-01-01T00:00:00", request.getRequestUrl().queryParameter("since")); + Assertions.assertEquals( + "2026-12-31T23:59:59", request.getRequestUrl().queryParameter("until")); + Assertions.assertEquals( + "partial,completed", + request.getRequestUrl().queryParameter("response_type")); + } + } + } + + @Test + public void testEmptyFormReadsNothing() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.enqueue(new MockResponse().setBody(page())); + server.start(); + + Assertions.assertTrue(read(server, new HashMap<>()).isEmpty()); + Assertions.assertEquals(1, server.getRequestCount()); + } + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/test/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSourceReaderTest.java b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/test/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSourceReaderTest.java new file mode 100644 index 0000000000..c830618780 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-typeform/src/test/java/org/apache/seatunnel/connectors/seatunnel/typeform/TypeformSourceReaderTest.java @@ -0,0 +1,208 @@ +/* + * 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.typeform; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.connectors.seatunnel.typeform.config.TypeformSourceParameter; + +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; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +public class TypeformSourceReaderTest { + + private TypeformSourceReader newReader(MockWebServer server, int retry) { + return newReader(server, retry, 10, 50, 0); + } + + private TypeformSourceReader newReader( + MockWebServer server, + int retry, + int backoffMillis, + int maxBackoffMillis, + long minRequestIntervalMillis) { + Map<String, Object> configMap = new HashMap<>(); + configMap.put("base_url", server.url("/").toString()); + configMap.put("api_key", "test-token"); + configMap.put("form_id", "abc123"); + configMap.put("retry", retry); + configMap.put("retry_backoff_multiplier_ms", backoffMillis); + configMap.put("retry_backoff_max_ms", maxBackoffMillis); + TypeformSourceParameter parameter = new TypeformSourceParameter(); + parameter.buildWithConfig(ReadonlyConfig.fromMap(configMap)); + TypeformSourceReader reader = + new TypeformSourceReader(parameter, null, null, null, null, null); + reader.setMinRequestIntervalMillis(minRequestIntervalMillis); + return reader; + } + + @Test + public void testRetryOn429And5xx() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.enqueue(new MockResponse().setResponseCode(429)); + server.enqueue(new MockResponse().setResponseCode(503)); + server.enqueue(new MockResponse().setResponseCode(200).setBody("{\"items\":[]}")); + server.start(); + TypeformSourceReader 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("/forms/abc123/responses?")); + Assertions.assertTrue(first.getPath().contains("page_size=100")); + Assertions.assertFalse(first.getPath().contains("test-token")); + } finally { + reader.close(); + } + } + } + + @Test + public void testNoRetryOn401() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.enqueue( + new MockResponse() + .setResponseCode(401) + .setBody("{\"code\":\"AUTHENTICATION_FAILED\"}")); + server.start(); + TypeformSourceReader 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(); + TypeformSourceReader reader = newReader(server, 3); + reader.open(); + try { + RuntimeException ex = + Assertions.assertThrows( + RuntimeException.class, () -> reader.executeRequest()); + Assertions.assertTrue(ex.getMessage().contains("HTTP 500")); + Assertions.assertFalse(ex.getMessage().contains("test-token")); + Assertions.assertEquals(4, server.getRequestCount()); + } finally { + reader.close(); + } + } + } + + @Test + public void testRequestsArePaced() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.enqueue(new MockResponse().setResponseCode(429)); + server.enqueue(new MockResponse().setResponseCode(200).setBody("{\"items\":[]}")); + server.start(); + // A 1 ms backoff, so the gap between the two requests comes from the pacing. + TypeformSourceReader reader = newReader(server, 3, 1, 50, 300); + reader.open(); + try { + long start = System.nanoTime(); + Assertions.assertEquals(200, reader.executeRequest().getCode()); + long elapsedMillis = (System.nanoTime() - start) / 1_000_000L; + Assertions.assertEquals(2, server.getRequestCount()); + Assertions.assertTrue( + elapsedMillis >= 300, + "Second request must wait for the pacing interval, took " + elapsedMillis); + } finally { + reader.close(); + } + } + } + + @Test + public void testDefaultPacingMatchesTypeformRateLimit() { + Assertions.assertEquals(500, TypeformSourceReader.DEFAULT_MIN_REQUEST_INTERVAL_MILLIS); + } + + @Test + public void testInterruptedBackoffRestoresFlag() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.enqueue(new MockResponse().setResponseCode(429)); + server.start(); + // A long backoff keeps the worker in Thread.sleep while it is interrupted. + TypeformSourceReader reader = newReader(server, 3, 20_000, 30_000, 0); + reader.open(); + + AtomicReference<Exception> thrown = new AtomicReference<>(); + AtomicBoolean flagRestored = new AtomicBoolean(false); + Thread worker = + new Thread( + () -> { + try { + reader.executeRequest(); + } catch (Exception e) { + thrown.set(e); + flagRestored.set(Thread.currentThread().isInterrupted()); + } + }); + try { + worker.start(); + Assertions.assertNotNull( + server.takeRequest(10, TimeUnit.SECONDS), "429 request not received"); + // Wait until the worker has read the 429 and is sleeping in the backoff. + long deadline = System.currentTimeMillis() + 10_000; + while (worker.getState() != Thread.State.TIMED_WAITING + && System.currentTimeMillis() < deadline) { + Thread.sleep(5); + } + worker.interrupt(); + worker.join(10_000); + + Assertions.assertFalse(worker.isAlive(), "Worker must stop after interrupt"); + Exception e = thrown.get(); + Assertions.assertNotNull(e, "Expected exception due to thread interruption"); + String msg = String.valueOf(e.getMessage()); + Assertions.assertTrue( + msg.contains("Interrupted while backing off"), + "Exception must come from the backoff sleep, got: " + msg); + Assertions.assertTrue(flagRestored.get(), "Interrupt flag must be restored"); + Assertions.assertFalse( + msg.contains("test-token"), "Message must not contain token"); + Assertions.assertFalse( + msg.contains("Authorization"), "Message must not contain Authorization"); + Assertions.assertEquals(1, server.getRequestCount()); + } finally { + worker.interrupt(); + reader.close(); + } + } + } +} diff --git a/seatunnel-connectors-v2/connector-http/pom.xml b/seatunnel-connectors-v2/connector-http/pom.xml index 25c61ca085..264bd42087 100644 --- a/seatunnel-connectors-v2/connector-http/pom.xml +++ b/seatunnel-connectors-v2/connector-http/pom.xml @@ -52,6 +52,7 @@ <module>connector-http-splunk</module> <module>connector-http-asana</module> <module>connector-http-intercom</module> + <module>connector-http-typeform</module> <module>connector-mem0</module> </modules> diff --git a/seatunnel-dist/pom.xml b/seatunnel-dist/pom.xml index c13f2757fa..c3762bf50e 100644 --- a/seatunnel-dist/pom.xml +++ b/seatunnel-dist/pom.xml @@ -260,6 +260,12 @@ <version>${project.version}</version> <scope>provided</scope> </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-typeform</artifactId> + <version>${project.version}</version> + <scope>provided</scope> + </dependency> <dependency> <groupId>org.apache.seatunnel</groupId> <artifactId>connector-kafka</artifactId> diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-typeform-e2e/pom.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-typeform-e2e/pom.xml new file mode 100644 index 0000000000..4afe6badcb --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-typeform-e2e/pom.xml @@ -0,0 +1,53 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.seatunnel</groupId> + <artifactId>seatunnel-connector-v2-e2e</artifactId> + <version>${revision}</version> + </parent> + <artifactId>connector-typeform-e2e</artifactId> + <name>SeaTunnel : E2E : Connector V2 : Typeform</name> + + <dependencies> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-typeform</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-typeform-e2e/src/test/java/org/apache/seatunnel/e2e/connector/typeform/TypeformIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-typeform-e2e/src/test/java/org/apache/seatunnel/e2e/connector/typeform/TypeformIT.java new file mode 100644 index 0000000000..bfae1c4169 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-typeform-e2e/src/test/java/org/apache/seatunnel/e2e/connector/typeform/TypeformIT.java @@ -0,0 +1,170 @@ +/* + * 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.typeform; + +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.io.UnsupportedEncodingException; +import java.net.InetSocketAddress; +import java.net.URI; +import java.net.URLDecoder; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.concurrent.atomic.AtomicInteger; + +public class TypeformIT extends TestSuiteBase implements TestResource { + + private static final String TOKEN = "placeholder-typeform-token-12345"; + private static final String PAGE_2_CURSOR = "tok-r2"; + private static final String PAGE_3_CURSOR = "tok-r3"; + + private static HttpServer httpServer; + private static int serverPort; + private static final AtomicInteger requestCount = new AtomicInteger(0); + + private static String item(String id, String answers) { + return "{\"landing_id\":\"l-" + + id + + "\",\"token\":\"tok-" + + id + + "\",\"response_id\":\"" + + id + + "\",\"landed_at\":\"2026-09-14T22:33:59Z\",\"submitted_at\":\"2026-09-14T22:38:22Z\"," + + "\"answers\":" + + answers + + ",\"hidden\":{\"utm_source\":\"newsletter\"}}"; + } + + private static String page(String... items) { + return "{\"total_items\":3,\"page_count\":2,\"items\":[" + String.join(",", items) + "]}"; + } + + @BeforeAll + @Override + public void startUp() throws IOException { + httpServer = HttpServer.create(new InetSocketAddress(0), 0); + serverPort = httpServer.getAddress().getPort(); + Testcontainers.exposeHostPorts(serverPort); + + httpServer.createContext( + "/forms/abc123/responses", + exchange -> { + int currentReq = requestCount.incrementAndGet(); + String cursor = queryParam(exchange.getRequestURI(), "before"); + String auth = exchange.getRequestHeaders().getFirst("Authorization"); + String responseBody; + int statusCode; + + // The first request gets a 429 to exercise the retry. After that the page is + // chosen by the before cursor, so the job only reads page 2 if the connector + // sends back the token of the last response on page 1. + if (!("Bearer " + TOKEN).equals(auth)) { + statusCode = 401; + responseBody = "{\"code\":\"AUTHENTICATION_FAILED\"}"; + } else if (currentReq == 1) { + statusCode = 429; + responseBody = "{\"code\":\"RATE_LIMITED\"}"; + } else if (cursor == null) { + statusCode = 200; + responseBody = + page( + item( + "r1", + "[{\"field\":{\"id\":\"f1\",\"type\":\"short_text\"},\"type\":\"text\",\"text\":\"Lian\"}]"), + item("r2", "[]")); + } else if (PAGE_2_CURSOR.equals(cursor)) { + statusCode = 200; + responseBody = + page( + item( + "r3", + "[{\"field\":{\"id\":\"f2\",\"type\":\"yes_no\"},\"type\":\"boolean\",\"boolean\":true}]")); + } else if (PAGE_3_CURSOR.equals(cursor)) { + statusCode = 200; + responseBody = page(); + } else { + statusCode = 400; + responseBody = "{\"code\":\"VALIDATION_ERROR\"}"; + } + + 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(); + } + + private static String queryParam(URI uri, String name) { + String query = uri.getRawQuery(); + if (query == null) { + return null; + } + for (String pair : query.split("&")) { + int eq = pair.indexOf('='); + if (eq > 0 && pair.substring(0, eq).equals(name)) { + try { + return URLDecoder.decode(pair.substring(eq + 1), "UTF-8"); + } catch (UnsupportedEncodingException e) { + throw new IllegalStateException(e); + } + } + } + return null; + } + + @AfterAll + @Override + public void tearDown() { + if (httpServer != null) { + httpServer.stop(0); + } + } + + @BeforeEach + public void resetCounter() { + requestCount.set(0); + } + + @TestTemplate + public void testTypeformSourceE2E(TestContainer container) + throws IOException, InterruptedException { + Container.ExecResult execResult = + container.executeJob( + "/typeform_source_to_assert.conf", + Collections.singletonList("typeform.test.port=" + serverPort)); + Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr()); + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-typeform-e2e/src/test/resources/typeform_source_to_assert.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-typeform-e2e/src/test/resources/typeform_source_to_assert.conf new file mode 100644 index 0000000000..3b7ce32500 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-typeform-e2e/src/test/resources/typeform_source_to_assert.conf @@ -0,0 +1,78 @@ +# +# 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 { + Typeform { + base_url = "http://host.testcontainers.internal:${typeform.test.port}"; + api_key = "placeholder-typeform-token-12345" + form_id = "abc123" + page_size = 2 + 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 = "response_id" + field_type = "string" + field_value = [ + { + rule_type = "NOT_NULL" + } + ] + }, + { + field_name = "token" + field_type = "string" + field_value = [ + { + rule_type = "NOT_NULL" + } + ] + }, + { + field_name = "hidden" + 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 af592af155..7368584279 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml @@ -53,6 +53,7 @@ <module>connector-neo4j-e2e</module> <module>connector-asana-e2e</module> <module>connector-intercom-e2e</module> + <module>connector-typeform-e2e</module> <module>connector-http-e2e</module> <module>connector-paypal-e2e</module> <module>connector-tiktok-ads-e2e</module>
