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-12831-59775c6cb02ed4dbb0415dd15c468b3cafcad9ca in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 4eb1dc591277fc85bf78ee29daa1fc6c7c5e077c Author: Rishavbhattarai <[email protected]> AuthorDate: Sat Oct 17 07:59:39 2026 +0000 [Feature][Connector-V2] Add Intercom source connector (#12831) Co-authored-by: Claude Opus 5.5 <[email protected]> --- config/plugin_config | 1 + .../changelog/connector-http-intercom.md | 3 + docs/en/connectors/source/Intercom.md | 88 +++++++++ .../changelog/connector-http-intercom.md | 3 + docs/zh/connectors/source/Intercom.md | 65 +++++++ plugin-mapping.properties | 1 + .../connector-http/connector-http-intercom/pom.xml | 51 ++++++ .../seatunnel/intercom/IntercomSource.java | 125 +++++++++++++ .../seatunnel/intercom/IntercomSourceFactory.java | 68 +++++++ .../seatunnel/intercom/IntercomSourceReader.java | 97 ++++++++++ .../intercom/config/IntercomSourceOptions.java | 52 ++++++ .../intercom/config/IntercomSourceParameter.java | 95 ++++++++++ .../intercom/IntercomSourceFactoryTest.java | 196 +++++++++++++++++++++ .../intercom/IntercomSourceReaderTest.java | 196 +++++++++++++++++++++ seatunnel-connectors-v2/connector-http/pom.xml | 1 + seatunnel-dist/pom.xml | 6 + .../connector-intercom-e2e/pom.xml | 53 ++++++ .../e2e/connector/intercom/IntercomIT.java | 144 +++++++++++++++ .../test/resources/intercom_source_to_assert.conf | 65 +++++++ seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml | 1 + 20 files changed, 1311 insertions(+) diff --git a/config/plugin_config b/config/plugin_config index 7600b6afc3..b4dfc44e62 100644 --- a/config/plugin_config +++ b/config/plugin_config @@ -72,6 +72,7 @@ connector-hive connector-http-base connector-mem0 connector-http-asana +connector-http-intercom connector-http-feishu connector-http-gitlab connector-http-github diff --git a/docs/en/connectors/changelog/connector-http-intercom.md b/docs/en/connectors/changelog/connector-http-intercom.md new file mode 100644 index 0000000000..111f2b8e2e --- /dev/null +++ b/docs/en/connectors/changelog/connector-http-intercom.md @@ -0,0 +1,3 @@ +| Change | Commit | Version | +| --- | --- | --- | +| [Feature][connector-http-intercom] Add Intercom source connector | https://github.com/apache/seatunnel | dev | \ No newline at end of file diff --git a/docs/en/connectors/source/Intercom.md b/docs/en/connectors/source/Intercom.md new file mode 100644 index 0000000000..fcdd33f52f --- /dev/null +++ b/docs/en/connectors/source/Intercom.md @@ -0,0 +1,88 @@ +import ChangeLog from '../changelog/connector-http-intercom.md'; + +# Http-Intercom + +> Intercom source connector + +## Description + +Reads contacts from the Intercom REST API (`GET /contacts`). Built on +`connector-http-base`. Batch mode only. Version 1 reads contacts only; conversations, tickets, +companies, search, OAuth and custom attributes are not supported. + +## Key Features + +- [x] Batch +- [ ] Stream +- [ ] Exactly-once +- [ ] Column projection +- [ ] Parallelism + +## Options + +| Name | Type | Required | Default | Description | +| --- | --- | --- | --- | --- | +| api_key | String | Yes | - | Intercom access token (raw token, no `Bearer ` prefix). | +| base_url | String | No | `https://api.intercom.io` | Use `https://api.eu.intercom.io` or `https://api.au.intercom.io` for EU or AU hosted workspaces. | +| api_version | String | No | `2.17` | Sent as the `Intercom-Version` header, so the response shape does not change with the workspace default. | +| per_page | Int | No | `150` | Contacts per page, 1 to 150 (Intercom's maximum). | +| 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. 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. + +### retry +Max retries on HTTP 429 and 5xx (and, in the base client, IOExceptions). Backoff doubles from +`retry_backoff_multiplier_ms` up to `retry_backoff_max_ms`. `Retry-After` is not honored. +Intercom's default limit is 10,000 calls per minute per app (25,000 per workspace), enforced in +10-second windows. + +## Output columns + +| Column | Type | Notes | +| --- | --- | --- | +| id | STRING | Intercom contact id | +| role | STRING | `user` or `lead` | +| email | STRING | | +| name | STRING | Can be null | +| phone | STRING | Can be null | +| external_id | STRING | Can be null | +| created_at | BIGINT | Unix timestamp in seconds | +| updated_at | BIGINT | Unix timestamp in seconds | + +## Example + +```hocon +env { + execution.parallelism = 1 + job.mode = BATCH +} + +source { + Intercom { + api_key = "<ACCESS_TOKEN>" + plugin_output = "intercom_contacts" + } +} + +sink { + Console { + plugin_input = "intercom_contacts" + } +} +``` + +## Notes + +Each run is a bounded snapshot of all contacts, read from the first page by following the +`pages.next.starting_after` cursor. The list endpoint has no "updated since" filter, so there is +no incremental mode. After a failure, a restarted job re-reads from the beginning and may emit rows +that were already sent; contacts created or merged during a run can also appear or disappear +between pages. Every row includes the contact `id`, which a downstream sink can use to deduplicate +or upsert, but the connector provides no delivery guarantee. + +<ChangeLog /> diff --git a/docs/zh/connectors/changelog/connector-http-intercom.md b/docs/zh/connectors/changelog/connector-http-intercom.md new file mode 100644 index 0000000000..377a9b9d5c --- /dev/null +++ b/docs/zh/connectors/changelog/connector-http-intercom.md @@ -0,0 +1,3 @@ +| Change | Commit | Version | +|------------------------------------------------------------| --- | --- | +| [Feature][connector-http-intercom] Add Intercom source connector | https://github.com/apache/seatunnel | dev | \ No newline at end of file diff --git a/docs/zh/connectors/source/Intercom.md b/docs/zh/connectors/source/Intercom.md new file mode 100644 index 0000000000..8bbd122aac --- /dev/null +++ b/docs/zh/connectors/source/Intercom.md @@ -0,0 +1,65 @@ +import ChangeLog from '../changelog/connector-http-intercom.md'; + +# Http-Intercom + +> Intercom 源连接器 + +## 描述 + +通过 Intercom REST API（`GET /contacts`）读取联系人。基于 `connector-http-base` 构建，仅支持批处理模式。 +第一版只读取联系人，不支持会话、工单、公司、搜索、OAuth 和自定义属性。 + +## 主要特性 + +- [x] 批处理 +- [ ] 流处理 +- [ ] 精确一次 +- [ ] 列投影 +- [ ] 并行度 + +## 选项 + +| 名称 | 类型 | 必填 | 默认值 | 描述 | +| --- | --- | --- | --- | --- | +| api_key | String | 是 | - | Intercom 访问令牌（原始令牌，不带 `Bearer ` 前缀）。 | +| base_url | String | 否 | `https://api.intercom.io` | EU 或 AU 区域的工作区请使用 `https://api.eu.intercom.io` 或 `https://api.au.intercom.io`。 | +| api_version | String | 否 | `2.17` | 作为 `Intercom-Version` 请求头发送，避免工作区默认版本变化导致响应结构改变。 | +| per_page | Int | 否 | `150` | 每页联系人数量，范围 1 到 150。 | +| retry | Int | 否 | `3` | HTTP 429 和 5xx 的最大重试次数。 | +| retry_backoff_multiplier_ms | Int | 否 | `100` | 退避倍数（毫秒）。 | +| retry_backoff_max_ms | Int | 否 | `10000` | 最大退避时间（毫秒）。 | + +## 输出列 + +`id`、`role`、`email`、`name`、`phone`、`external_id`（均为 STRING，后三者可为 null）， +以及 `created_at`、`updated_at`（BIGINT，Unix 秒级时间戳）。 + +## 示例 + +```hocon +env { + execution.parallelism = 1 + job.mode = BATCH +} + +source { + Intercom { + api_key = "<ACCESS_TOKEN>" + plugin_output = "intercom_contacts" + } +} + +sink { + Console { + plugin_input = "intercom_contacts" + } +} +``` + +## 注意事项 + +每次运行都是联系人的有界快照，从第一页开始沿 `pages.next.starting_after` 游标读取。列表接口没有“按更新时间过滤”，因此没有增量模式。 +作业失败后重启会从头读取，可能重复发送已发送的行，读取期间新建或合并的联系人也可能在分页之间出现或消失。 +每行都包含联系人 `id`，下游可据此去重或 upsert，但连接器不提供投递保证。 + +<ChangeLog /> diff --git a/plugin-mapping.properties b/plugin-mapping.properties index cf0aa84a45..f47a5e41a0 100644 --- a/plugin-mapping.properties +++ b/plugin-mapping.properties @@ -208,6 +208,7 @@ seatunnel.source.Linear = connector-http-linear seatunnel.source.Splunk = connector-http-splunk seatunnel.source.WebSocket = connector-websocket seatunnel.source.Asana = connector-http-asana +seatunnel.source.Intercom = connector-http-intercom # 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-intercom/pom.xml b/seatunnel-connectors-v2/connector-http/connector-http-intercom/pom.xml new file mode 100644 index 0000000000..968df7627c --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-intercom/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-intercom</artifactId> + <name>SeaTunnel : Connectors V2 : Http : Intercom</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-intercom/src/main/java/org/apache/seatunnel/connectors/seatunnel/intercom/IntercomSource.java b/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/main/java/org/apache/seatunnel/connectors/seatunnel/intercom/IntercomSource.java new file mode 100644 index 0000000000..de01805525 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/main/java/org/apache/seatunnel/connectors/seatunnel/intercom/IntercomSource.java @@ -0,0 +1,125 @@ +/* + * 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.intercom; + +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.SeaTunnelDataType; +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.intercom.config.IntercomSourceParameter; +import org.apache.seatunnel.format.json.JsonDeserializationSchema; + +import lombok.extern.slf4j.Slf4j; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * Intercom contacts source. Reuses {@link HttpSource}, but after super(pluginConfig) replaces the + * schema, JSONPath fields and paging so the eight fixed contact columns and the cursor pagination + * ("pages.next.starting_after") are used instead of user-supplied config. + */ +@Slf4j +public class IntercomSource extends HttpSource { + private final IntercomSourceParameter intercomSourceParameter = new IntercomSourceParameter(); + + public IntercomSource(ReadonlyConfig pluginConfig) { + super(pluginConfig); + intercomSourceParameter.buildWithConfig(pluginConfig); + + TableSchema.Builder schema = TableSchema.builder(); + Map<String, String> fields = new LinkedHashMap<>(); + + Map<String, SeaTunnelDataType<?>> types = new LinkedHashMap<>(); + types.put("id", BasicType.STRING_TYPE); + types.put("role", BasicType.STRING_TYPE); + types.put("email", BasicType.STRING_TYPE); + types.put("name", BasicType.STRING_TYPE); + types.put("phone", BasicType.STRING_TYPE); + types.put("external_id", BasicType.STRING_TYPE); + types.put("created_at", BasicType.LONG_TYPE); + types.put("updated_at", BasicType.LONG_TYPE); + types.keySet().forEach(c -> fields.put(c, "$.data[*]." + c)); + + types.forEach( + (name, type) -> schema.column(PhysicalColumn.of(name, type, 0, true, null, null))); + this.catalogTable = + CatalogTable.of( + TableIdentifier.of("Intercom", 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("starting_after"); + info.setPageCursorResponseField("$.pages.next.starting_after"); + 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 IntercomSourceReader( + this.intercomSourceParameter, + readerContext, + this.deserializationSchema, + jsonField, + contentField, + pageInfo); + } + + @Override + public String getPluginName() { + return "Intercom"; + } + + @Override + public Boundedness getBoundedness() { + if (JobMode.STREAMING.equals(jobContext.getJobMode())) { + throw new HttpConnectorException( + CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT, + "Intercom source only supports batch mode."); + } + return Boundedness.BOUNDED; + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/main/java/org/apache/seatunnel/connectors/seatunnel/intercom/IntercomSourceFactory.java b/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/main/java/org/apache/seatunnel/connectors/seatunnel/intercom/IntercomSourceFactory.java new file mode 100644 index 0000000000..66d1adeb9c --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/main/java/org/apache/seatunnel/connectors/seatunnel/intercom/IntercomSourceFactory.java @@ -0,0 +1,68 @@ +/* + * 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.intercom; + +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.intercom.config.IntercomSourceOptions; + +import com.google.auto.service.AutoService; + +import java.io.Serializable; + +/** + * Factory that registers the Intercom source under the identifier "Intercom" and declares its + * options. + */ +@AutoService(Factory.class) +public class IntercomSourceFactory extends HttpSourceFactory { + @Override + public String factoryIdentifier() { + return "Intercom"; + } + + @Override + public <T, SplitT extends SourceSplit, StateT extends Serializable> + TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) { + return () -> (SeaTunnelSource<T, SplitT, StateT>) new IntercomSource(context.getOptions()); + } + + @Override + public OptionRule optionRule() { + return OptionRule.builder() + .required(IntercomSourceOptions.API_KEY) + .optional( + IntercomSourceOptions.BASE_URL, + IntercomSourceOptions.API_VERSION, + IntercomSourceOptions.PER_PAGE, + IntercomSourceOptions.RETRY, + IntercomSourceOptions.RETRY_BACKOFF_MULTIPLIER_MS, + IntercomSourceOptions.RETRY_BACKOFF_MAX_MS) + .build(); + } + + @Override + public Class<? extends SeaTunnelSource> getSourceClass() { + return IntercomSource.class; + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/main/java/org/apache/seatunnel/connectors/seatunnel/intercom/IntercomSourceReader.java b/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/main/java/org/apache/seatunnel/connectors/seatunnel/intercom/IntercomSourceReader.java new file mode 100644 index 0000000000..c31c35335d --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/main/java/org/apache/seatunnel/connectors/seatunnel/intercom/IntercomSourceReader.java @@ -0,0 +1,97 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.intercom; + +import org.apache.seatunnel.api.serialization.DeserializationSchema; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated; +import org.apache.seatunnel.connectors.seatunnel.common.source.SingleSplitReaderContext; +import org.apache.seatunnel.connectors.seatunnel.http.client.HttpResponse; +import org.apache.seatunnel.connectors.seatunnel.http.config.HttpParameter; +import org.apache.seatunnel.connectors.seatunnel.http.config.JsonField; +import org.apache.seatunnel.connectors.seatunnel.http.config.PageInfo; +import org.apache.seatunnel.connectors.seatunnel.http.exception.HttpConnectorException; +import org.apache.seatunnel.connectors.seatunnel.http.source.HttpSourceReader; + +import lombok.extern.slf4j.Slf4j; + +/** + * Reader that re-implements executeRequest to retry HTTP 429 and 5xx with exponential backoff + * before handing the response to the base reader. + */ +@Slf4j +public class IntercomSourceReader extends HttpSourceReader { + + public IntercomSourceReader( + HttpParameter httpParameter, + SingleSplitReaderContext readerContext, + DeserializationSchema<SeaTunnelRow> deserializationSchema, + JsonField jsonField, + String contentField, + PageInfo pageInfo) { + super( + httpParameter, + readerContext, + deserializationSchema, + jsonField, + contentField, + pageInfo); + } + + @Override + protected HttpResponse executeRequest() throws Exception { + int maxRetries = httpParameter.getRetry(); + long waitMillis = httpParameter.getRetryBackoffMultiplierMillis(); + long maxWaitMillis = httpParameter.getRetryBackoffMaxMillis(); + int attempt = 1; + while (true) { + HttpResponse response = super.executeRequest(); + int code = response.getCode(); + boolean retryable = code == 429 || (code >= 500 && code < 600); + if (!retryable) { + return response; + } + if (attempt > maxRetries) { + throw new HttpConnectorException( + CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT, + String.format( + "Intercom request failed with HTTP %d after %d attempt(s) (retry=%d)", + code, attempt, maxRetries)); + } + log.warn( + "Intercom 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 Intercom HTTP %d (attempt %d)", + code, attempt), + e); + } + waitMillis = Math.min(waitMillis * 2, maxWaitMillis); + attempt++; + } + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/main/java/org/apache/seatunnel/connectors/seatunnel/intercom/config/IntercomSourceOptions.java b/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/main/java/org/apache/seatunnel/connectors/seatunnel/intercom/config/IntercomSourceOptions.java new file mode 100644 index 0000000000..2c066e249b --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/main/java/org/apache/seatunnel/connectors/seatunnel/intercom/config/IntercomSourceOptions.java @@ -0,0 +1,52 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.intercom.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 Intercom source, on top of the shared HTTP options. */ +public class IntercomSourceOptions extends HttpCommonOptions { + public static final Option<String> API_KEY = + Options.key("api_key") + .stringType() + .noDefaultValue() + .withDescription("Intercom access token (raw token, no Bearer prefix)"); + + public static final Option<String> BASE_URL = + Options.key("base_url") + .stringType() + .defaultValue("https://api.intercom.io";) + .withDescription( + "Intercom API base URL. Use https://api.eu.intercom.io or" + + " https://api.au.intercom.io for workspaces hosted in those" + + " regions."); + + public static final Option<String> API_VERSION = + Options.key("api_version") + .stringType() + .defaultValue("2.17") + .withDescription("Value sent in the Intercom-Version header"); + + public static final Option<Integer> PER_PAGE = + Options.key("per_page") + .intType() + .defaultValue(150) + .withDescription("Contacts requested per page (1 to 150)"); +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/main/java/org/apache/seatunnel/connectors/seatunnel/intercom/config/IntercomSourceParameter.java b/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/main/java/org/apache/seatunnel/connectors/seatunnel/intercom/config/IntercomSourceParameter.java new file mode 100644 index 0000000000..0c68df23f1 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/main/java/org/apache/seatunnel/connectors/seatunnel/intercom/config/IntercomSourceParameter.java @@ -0,0 +1,95 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.intercom.config; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.connectors.seatunnel.http.config.HttpParameter; +import org.apache.seatunnel.connectors.seatunnel.http.config.HttpRequestMethod; + +import java.util.HashMap; + +/** + * Builds the HTTP request parameters for the Intercom contacts endpoint: the contacts URL, the + * Bearer Authorization and Intercom-Version headers, the page size and retry defaults. Validates + * that api_key and api_version are not blank and that per_page is within 1..150. + */ +public class IntercomSourceParameter extends HttpParameter { + + private static final int MAX_PER_PAGE = 150; + + private static String requireNonBlank(String value, String optionKey) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException( + String.format("Intercom option '%s' must not be blank.", optionKey)); + } + return value.trim(); + } + + /** + * Builds the request for {@code GET {base_url}/contacts}. Validates that api_key, api_version + * and base_url are not blank, that per_page is within 1..150 and that the retry backoff values + * are not negative. 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(IntercomSourceOptions.API_KEY), "api_key"); + String apiVersion = + requireNonBlank(config.get(IntercomSourceOptions.API_VERSION), "api_version"); + int perPage = config.get(IntercomSourceOptions.PER_PAGE); + if (perPage < 1 || perPage > MAX_PER_PAGE) { + throw new IllegalArgumentException( + String.format( + "Intercom option 'per_page' must be between 1 and %d.", MAX_PER_PAGE)); + } + + String base = requireNonBlank(config.get(IntercomSourceOptions.BASE_URL), "base_url"); + if (base.endsWith("/")) { + base = base.substring(0, base.length() - 1); + } + setUrl(base + "/contacts"); + setMethod(HttpRequestMethod.GET); + + if (headers == null) { + headers = new HashMap<>(); + } + headers.put("Authorization", "Bearer " + apiKey); + headers.put("Intercom-Version", apiVersion); + 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("per_page", String.valueOf(perPage)); + setKeepPageParamAsHttpParam(true); + setJsonFiledMissedReturnNull(true); + + if (!config.getOptional(IntercomSourceOptions.RETRY).isPresent()) { + setRetry(3); + } + setRetryBackoffMultiplierMillis( + config.get(IntercomSourceOptions.RETRY_BACKOFF_MULTIPLIER_MS)); + setRetryBackoffMaxMillis(config.get(IntercomSourceOptions.RETRY_BACKOFF_MAX_MS)); + + if (getRetryBackoffMultiplierMillis() < 0 || getRetryBackoffMaxMillis() < 0) { + throw new IllegalArgumentException( + "Intercom retry backoff values must not be negative."); + } + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/test/java/org/apache/seatunnel/connectors/seatunnel/intercom/IntercomSourceFactoryTest.java b/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/test/java/org/apache/seatunnel/connectors/seatunnel/intercom/IntercomSourceFactoryTest.java new file mode 100644 index 0000000000..c82cb2dd42 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/test/java/org/apache/seatunnel/connectors/seatunnel/intercom/IntercomSourceFactoryTest.java @@ -0,0 +1,196 @@ +/* + * 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.intercom; + +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.apache.seatunnel.connectors.seatunnel.http.config.HttpRequestMethod; +import org.apache.seatunnel.connectors.seatunnel.intercom.config.IntercomSourceParameter; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import okhttp3.mockwebserver.RecordedRequest; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +public class IntercomSourceFactoryTest { + + private static Map<String, Object> baseConfig() { + Map<String, Object> m = new HashMap<>(); + m.put("api_key", "test-intercom-token-123"); + return m; + } + + @Test + public void testFactoryIdentifier() { + Assertions.assertEquals("Intercom", new IntercomSourceFactory().factoryIdentifier()); + Assertions.assertEquals( + "Intercom", + new IntercomSource(ReadonlyConfig.fromMap(baseConfig())).getPluginName()); + } + + @Test + public void testParameterBuildingAndHeaders() { + IntercomSourceParameter parameter = new IntercomSourceParameter(); + parameter.buildWithConfig(ReadonlyConfig.fromMap(baseConfig())); + + Assertions.assertEquals( + "Bearer test-intercom-token-123", parameter.getHeaders().get("Authorization")); + Assertions.assertEquals("2.17", parameter.getHeaders().get("Intercom-Version")); + Assertions.assertEquals(HttpRequestMethod.GET, parameter.getMethod()); + Assertions.assertEquals("https://api.intercom.io/contacts";, parameter.getUrl()); + Assertions.assertEquals("150", parameter.getParams().get("per_page")); + Assertions.assertFalse(parameter.getParams().containsKey("Authorization")); + } + + @Test + public void testRegionBaseUrlVersionAndPerPage() { + Map<String, Object> m = baseConfig(); + m.put("base_url", "https://api.eu.intercom.io/";); + m.put("api_version", "2.14"); + m.put("per_page", 50); + IntercomSourceParameter parameter = new IntercomSourceParameter(); + parameter.buildWithConfig(ReadonlyConfig.fromMap(m)); + + Assertions.assertEquals("https://api.eu.intercom.io/contacts";, parameter.getUrl()); + Assertions.assertEquals("2.14", parameter.getHeaders().get("Intercom-Version")); + Assertions.assertEquals("50", parameter.getParams().get("per_page")); + } + + @Test + public void testBlankApiKeyValidation() { + Map<String, Object> m = baseConfig(); + m.put("api_key", " "); + Exception e = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> + new IntercomSourceParameter() + .buildWithConfig(ReadonlyConfig.fromMap(m))); + Assertions.assertTrue(e.getMessage().contains("api_key")); + Assertions.assertFalse(e.getMessage().contains("Authorization")); + } + + @Test + public void testPerPageOutOfRangeRejected() { + for (int bad : new int[] {0, 151}) { + Map<String, Object> m = baseConfig(); + m.put("per_page", bad); + Exception e = + Assertions.assertThrows( + IllegalArgumentException.class, + () -> + new IntercomSourceParameter() + .buildWithConfig(ReadonlyConfig.fromMap(m))); + Assertions.assertTrue(e.getMessage().contains("per_page")); + Assertions.assertFalse(e.getMessage().contains("test-intercom-token-123")); + } + } + + @Test + public void testNegativeBackoffRejected() { + Map<String, Object> m = baseConfig(); + m.put("retry_backoff_multiplier_ms", -1); + Assertions.assertThrows( + IllegalArgumentException.class, + () -> new IntercomSourceParameter().buildWithConfig(ReadonlyConfig.fromMap(m))); + } + + @Test + public void testSourceReaderPaginationAndNullFields() throws Exception { + try (MockWebServer server = new MockWebServer()) { + String page1 = + "{\"type\":\"list\",\"data\":[" + + "{\"id\":\"c1\",\"role\":\"user\",\"email\":\"[email protected]\",\"name\":\"Alice\",\"phone\":null,\"external_id\":null,\"created_at\":1571672154,\"updated_at\":1571672200}," + + "{\"id\":\"c2\",\"role\":\"lead\",\"email\":\"[email protected]\",\"name\":null,\"phone\":\"+1123456789\",\"external_id\":\"ext-2\",\"created_at\":1571672155,\"updated_at\":1571672201}" + + "],\"pages\":{\"type\":\"pages\",\"next\":{\"page\":2,\"starting_after\":\"ab.c_-==%2B\"}}}"; + String page2 = + "{\"type\":\"list\",\"data\":[" + + "{\"id\":\"c3\",\"role\":\"user\",\"email\":\"[email protected]\",\"name\":\"Carol\",\"created_at\":1571672156,\"updated_at\":1571672202}" + + "],\"pages\":{\"type\":\"pages\"}}"; + server.enqueue(new MockResponse().setBody(page1).setResponseCode(200)); + server.enqueue(new MockResponse().setBody(page2).setResponseCode(200)); + server.start(); + + Map<String, Object> m = baseConfig(); + m.put("base_url", server.url("/").toString()); + IntercomSource source = new IntercomSource(ReadonlyConfig.fromMap(m)); + + SourceReader.Context ctx = Mockito.mock(SourceReader.Context.class); + IntercomSourceReader reader = + (IntercomSourceReader) source.createReader(new SingleSplitReaderContext(ctx)); + + 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(); + } + + RecordedRequest first = server.takeRequest(5, TimeUnit.SECONDS); + Assertions.assertNotNull(first); + Assertions.assertNull(first.getRequestUrl().queryParameter("starting_after")); + Assertions.assertEquals("150", first.getRequestUrl().queryParameter("per_page")); + Assertions.assertEquals( + "Bearer test-intercom-token-123", first.getHeader("Authorization")); + Assertions.assertEquals("2.17", first.getHeader("Intercom-Version")); + + RecordedRequest second = server.takeRequest(5, TimeUnit.SECONDS); + Assertions.assertNotNull(second); + Assertions.assertEquals( + "ab.c_-==%2B", second.getRequestUrl().queryParameter("starting_after")); + Assertions.assertEquals(2, server.getRequestCount()); + + Assertions.assertEquals(3, rows.size()); + Assertions.assertEquals("c1", rows.get(0).getField(0)); + Assertions.assertEquals(1571672154L, rows.get(0).getField(6)); + Assertions.assertEquals(1571672200L, rows.get(0).getField(7)); + Assertions.assertNull(rows.get(0).getField(4)); + Assertions.assertNull(rows.get(0).getField(5)); + Assertions.assertNull(rows.get(1).getField(3)); + Assertions.assertEquals("ext-2", rows.get(1).getField(5)); + Assertions.assertEquals("c3", rows.get(2).getField(0)); + Assertions.assertNull(rows.get(2).getField(4)); + } + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/test/java/org/apache/seatunnel/connectors/seatunnel/intercom/IntercomSourceReaderTest.java b/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/test/java/org/apache/seatunnel/connectors/seatunnel/intercom/IntercomSourceReaderTest.java new file mode 100644 index 0000000000..258bb1d062 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-http-intercom/src/test/java/org/apache/seatunnel/connectors/seatunnel/intercom/IntercomSourceReaderTest.java @@ -0,0 +1,196 @@ +/* + * 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.intercom; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.connectors.seatunnel.intercom.config.IntercomSourceParameter; + +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 IntercomSourceReaderTest { + + private IntercomSourceReader newReader(MockWebServer server, int retry) { + return newReader(server, retry, 10, 50); + } + + private IntercomSourceReader newReader( + MockWebServer server, int retry, int backoffMillis, int maxBackoffMillis) { + Map<String, Object> configMap = new HashMap<>(); + configMap.put("base_url", server.url("/").toString()); + configMap.put("api_key", "test-token"); + configMap.put("retry", retry); + configMap.put("retry_backoff_multiplier_ms", backoffMillis); + configMap.put("retry_backoff_max_ms", maxBackoffMillis); + ReadonlyConfig config = ReadonlyConfig.fromMap(configMap); + IntercomSourceParameter parameter = new IntercomSourceParameter(); + parameter.buildWithConfig(config); + return new IntercomSourceReader(parameter, null, null, null, null, null); + } + + @Test + public void testRetryOn429And5xx() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.enqueue(new MockResponse().setResponseCode(429)); + server.enqueue(new MockResponse().setResponseCode(500)); + server.enqueue(new MockResponse().setResponseCode(200).setBody("{\"data\":[]}")); + server.start(); + IntercomSourceReader 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("/contacts?")); + Assertions.assertEquals("2.17", first.getHeader("Intercom-Version")); + Assertions.assertTrue(first.getPath().contains("per_page=150")); + } finally { + reader.close(); + } + } + } + + @Test + public void testNoRetryOn401() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.enqueue( + new MockResponse().setResponseCode(401).setBody("{\"type\":\"error.list\"}")); + server.start(); + IntercomSourceReader 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(); + IntercomSourceReader reader = newReader(server, 3); + reader.open(); + try { + RuntimeException ex = + Assertions.assertThrows( + RuntimeException.class, () -> reader.executeRequest()); + Assertions.assertTrue(ex.getMessage().contains("HTTP 500")); + Assertions.assertEquals(4, server.getRequestCount()); + } finally { + reader.close(); + } + } + } + + @Test + public void testInterruptedBackoffRestoresFlag() throws Exception { + try (MockWebServer server = new MockWebServer()) { + server.enqueue(new MockResponse().setResponseCode(429)); + server.start(); + // A long backoff keeps the worker in Thread.sleep while it is interrupted. + IntercomSourceReader reader = newReader(server, 3, 20_000, 30_000); + 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(); + } + } + } + + @Test + public void testCustomBackoffHonoredWithoutExplicitRetry() { + HashMap<String, Object> configMap = new HashMap<>(); + configMap.put("api_key", "test-token"); + configMap.put("retry_backoff_multiplier_ms", 2000); + configMap.put("retry_backoff_max_ms", 3000); + IntercomSourceParameter parameter = new IntercomSourceParameter(); + parameter.buildWithConfig(ReadonlyConfig.fromMap(configMap)); + + Assertions.assertEquals(3, parameter.getRetry()); + Assertions.assertEquals(2000, parameter.getRetryBackoffMultiplierMillis()); + Assertions.assertEquals(3000, parameter.getRetryBackoffMaxMillis()); + } + + @Test + public void testDefaultBackoffWhenNothingConfigured() { + HashMap<String, Object> configMap = new HashMap<>(); + configMap.put("api_key", "test-token"); + IntercomSourceParameter parameter = new IntercomSourceParameter(); + parameter.buildWithConfig(ReadonlyConfig.fromMap(configMap)); + + Assertions.assertEquals(3, parameter.getRetry()); + Assertions.assertEquals(100, parameter.getRetryBackoffMultiplierMillis()); + Assertions.assertEquals(10000, parameter.getRetryBackoffMaxMillis()); + } +} diff --git a/seatunnel-connectors-v2/connector-http/pom.xml b/seatunnel-connectors-v2/connector-http/pom.xml index d190af554b..25c61ca085 100644 --- a/seatunnel-connectors-v2/connector-http/pom.xml +++ b/seatunnel-connectors-v2/connector-http/pom.xml @@ -51,6 +51,7 @@ <module>connector-http-linear</module> <module>connector-http-splunk</module> <module>connector-http-asana</module> + <module>connector-http-intercom</module> <module>connector-mem0</module> </modules> diff --git a/seatunnel-dist/pom.xml b/seatunnel-dist/pom.xml index 007279589b..c13f2757fa 100644 --- a/seatunnel-dist/pom.xml +++ b/seatunnel-dist/pom.xml @@ -254,6 +254,12 @@ <version>${project.version}</version> <scope>provided</scope> </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-intercom</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-intercom-e2e/pom.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-intercom-e2e/pom.xml new file mode 100644 index 0000000000..af85ed5889 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-intercom-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-intercom-e2e</artifactId> + <name>SeaTunnel : E2E : Connector V2 : Intercom</name> + + <dependencies> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-intercom</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-intercom-e2e/src/test/java/org/apache/seatunnel/e2e/connector/intercom/IntercomIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-intercom-e2e/src/test/java/org/apache/seatunnel/e2e/connector/intercom/IntercomIT.java new file mode 100644 index 0000000000..4575415a47 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-intercom-e2e/src/test/java/org/apache/seatunnel/e2e/connector/intercom/IntercomIT.java @@ -0,0 +1,144 @@ +/* + * 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.intercom; + +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 IntercomIT extends TestSuiteBase implements TestResource { + + private static HttpServer httpServer; + private static int serverPort; + private static final AtomicInteger requestCount = new AtomicInteger(0); + private static final String PAGE_2_CURSOR = "WzE1NzE2NzIxNTUwMDAsImMyIiwyXQ=="; + + @BeforeAll + @Override + public void startUp() throws IOException { + httpServer = HttpServer.create(new InetSocketAddress(0), 0); + serverPort = httpServer.getAddress().getPort(); + Testcontainers.exposeHostPorts(serverPort); + + httpServer.createContext( + "/contacts", + exchange -> { + int currentReq = requestCount.incrementAndGet(); + String cursor = queryParam(exchange.getRequestURI(), "starting_after"); + String responseBody; + int statusCode; + + // The first request gets a 429 to exercise the retry. After that the page is + // chosen by the starting_after cursor, so the job only sees page 2 if the + // connector sends back the cursor from page 1. + if (currentReq == 1) { + statusCode = 429; + responseBody = "{\"type\": \"error.list\", \"errors\": []}"; + } else if (cursor == null) { + statusCode = 200; + responseBody = + "{\"type\": \"list\", \"data\": [" + + "{\"type\": \"contact\", \"id\": \"c1\", \"role\": \"user\", \"email\": \"[email protected]\", \"name\": \"Alice\", \"phone\": null, \"external_id\": null, \"created_at\": 1571672154, \"updated_at\": 1571672200}," + + "{\"type\": \"contact\", \"id\": \"c2\", \"role\": \"lead\", \"email\": \"[email protected]\", \"name\": null, \"phone\": \"+1123456789\", \"external_id\": \"ext-2\", \"created_at\": 1571672155, \"updated_at\": 1571672201}" + + "], \"pages\": {\"type\": \"pages\", \"next\": {\"page\": 2, \"starting_after\": \"" + + PAGE_2_CURSOR + + "\"}}}"; + } else if (PAGE_2_CURSOR.equals(cursor)) { + statusCode = 200; + responseBody = + "{\"type\": \"list\", \"data\": [" + + "{\"type\": \"contact\", \"id\": \"c3\", \"role\": \"user\", \"email\": \"[email protected]\", \"name\": \"Carol\", \"phone\": null, \"external_id\": null, \"created_at\": 1571672156, \"updated_at\": 1571672202}" + + "], \"pages\": {\"type\": \"pages\"}}"; + } else { + statusCode = 400; + responseBody = + "{\"type\": \"error.list\", \"errors\": [{\"code\": \"parameter_invalid\"}]}"; + } + + 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 testIntercomSourceE2E(TestContainer container) + throws IOException, InterruptedException { + Container.ExecResult execResult = + container.executeJob( + "/intercom_source_to_assert.conf", + Collections.singletonList("intercom.test.port=" + serverPort)); + Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr()); + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-intercom-e2e/src/test/resources/intercom_source_to_assert.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-intercom-e2e/src/test/resources/intercom_source_to_assert.conf new file mode 100644 index 0000000000..b783361469 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-intercom-e2e/src/test/resources/intercom_source_to_assert.conf @@ -0,0 +1,65 @@ +# 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 { + Intercom { + base_url = "http://host.testcontainers.internal:${intercom.test.port}"; + api_key = "placeholder-access-token-12345" + 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 = "id" + field_type = "string" + field_value = [ + { + rule_type = "NOT_NULL" + } + ] + }, + { + field_name = "created_at" + field_type = "bigint" + field_value = [ + { + rule_type = "NOT_NULL" + } + ] + } + ] + } + } +} \ No newline at end of file diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml index cf534b2972..af592af155 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/pom.xml @@ -52,6 +52,7 @@ <module>connector-cassandra-e2e</module> <module>connector-neo4j-e2e</module> <module>connector-asana-e2e</module> + <module>connector-intercom-e2e</module> <module>connector-http-e2e</module> <module>connector-paypal-e2e</module> <module>connector-tiktok-ads-e2e</module>
