This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 75f9636c99 [Feature][Connector-V2] Add Typeform source connector
(#12844)
75f9636c99 is described below
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>
