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-12245-7a427d40de48dddba1188a51c6baec7a1ad60c26 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 7d3649ade23ee077d608453d3e9ed54da4479d82 Author: Goutam Adwant <[email protected]> AuthorDate: Sat Sep 12 09:20:52 2026 +0000 [Feature][Connector-V2] Add Salesforce sink (#12245) Signed-off-by: Goutam Adwant <[email protected]> --- .github/workflows/labeler/label-scope-conf.yml | 5 +- .../connectors/changelog/connector-salesforce.md | 5 + docs/en/connectors/sink/Salesforce.md | 157 +++++++ .../connectors/changelog/connector-salesforce.md | 5 + docs/zh/connectors/sink/Salesforce.md | 138 ++++++ plugin-mapping.properties | 1 + .../salesforce/client/SalesforceClient.java | 223 ++++++++- .../salesforce/config/SalesforceSinkConfig.java | 107 +++++ .../salesforce/config/SalesforceSinkOptions.java | 53 +++ .../exception/SalesforceConnectorErrorCode.java | 3 +- .../salesforce/sink/SalesforceRowSerializer.java | 206 +++++++++ .../seatunnel/salesforce/sink/SalesforceSink.java | 57 +++ .../salesforce/sink/SalesforceSinkFactory.java | 91 ++++ .../salesforce/sink/SalesforceSinkWriter.java | 174 +++++++ .../sink/SalesforceRowSerializerTest.java | 251 ++++++++++ .../salesforce/sink/SalesforceSinkTest.java | 505 +++++++++++++++++++++ .../core/starter/utils/ConfigShadeUtils.java | 2 +- .../core/starter/utils/ConfigBuilderTest.java | 24 + .../connector-http-e2e/pom.xml | 6 + .../e2e/connector/http/SalesforceSinkIT.java | 132 ++++++ .../src/test/resources/fake_to_salesforce.conf | 47 ++ 21 files changed, 2186 insertions(+), 6 deletions(-) diff --git a/.github/workflows/labeler/label-scope-conf.yml b/.github/workflows/labeler/label-scope-conf.yml index fcc5cbdc9e..7dea290bc8 100644 --- a/.github/workflows/labeler/label-scope-conf.yml +++ b/.github/workflows/labeler/label-scope-conf.yml @@ -102,7 +102,10 @@ cassandra: salesforce: - all: - changed-files: - - any-glob-to-any-file: seatunnel-connectors-v2/connector-salesforce/** + - any-glob-to-any-file: + - seatunnel-connectors-v2/connector-salesforce/** + - seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/java/org/apache/seatunnel/e2e/connector/http/SalesforceSinkIT.java + - seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/resources/fake_to_salesforce.conf - all-globs-to-all-files: '!seatunnel-connectors-v2/connector-!(salesforce)/**' deeplake: - all: diff --git a/docs/en/connectors/changelog/connector-salesforce.md b/docs/en/connectors/changelog/connector-salesforce.md new file mode 100644 index 0000000000..04fce7f381 --- /dev/null +++ b/docs/en/connectors/changelog/connector-salesforce.md @@ -0,0 +1,5 @@ +# Changelog + +| Change | Commit | Version | +| --- | --- | --- | +| Add external-ID REST collection upsert sink | - | Next | diff --git a/docs/en/connectors/sink/Salesforce.md b/docs/en/connectors/sink/Salesforce.md new file mode 100644 index 0000000000..deb2d9e588 --- /dev/null +++ b/docs/en/connectors/sink/Salesforce.md @@ -0,0 +1,157 @@ +import ChangeLog from '../changelog/connector-salesforce.md'; + +# Salesforce + +> Salesforce REST sObject Collections upsert sink + +## Description + +Writes one input table to one existing Salesforce object using its external ID field. +The connector extends the existing Salesforce module without changing source options or data-reading behavior. +It does not create objects, fields, external IDs, or connected apps. + +## Support Those Engines + +> SeaTunnel Zeta<br/> +> Flink<br/> +> Spark<br/> + +## Key Features + +- [x] batch +- [x] stream +- [ ] exactly-once +- [ ] cdc +- [ ] support multiple table write + +Streaming support is limited to `INSERT` and `UPDATE_AFTER` rows. This is not a CDC sink: +`UPDATE_BEFORE` and `DELETE` fail the job. See [Delivery Semantics](#delivery-semantics) +before connecting a changelog source. + +## Options + +| name | type | required | default value | +| --- | --- | --- | --- | +| client_id | string | yes | - | +| client_secret | string | yes | - | +| username | string | yes | - | +| password | string | yes | - | +| security_token | string | no | empty string | +| instance_url | string | yes | - | +| api_version | string | no | v59.0 | +| object_name | string | yes | - | +| external_id_field | string | yes | - | +| batch_size | int | no | 200 | +| batch_max_bytes | int | no | 1048576 | +| request_timeout_ms | int | no | 60000 | +| max_retries | int | no | 3 | +| retry_interval_ms | long | no | 1000 | +| common-options | | no | - | + +### Authentication and Permissions + +Uses the existing source connector's OAuth username-password flow. Your org must permit that flow; +the integration user needs API access, object create/update access, and write access to all mapped fields. +The security token is appended to the password, as in the source connector. + +Use an HTTPS login or org origin for `instance_url`, for example +`https://login.salesforce.com` or your sandbox login origin. Do not include a trailing slash, +path, embedded credentials, query, or fragment. HTTP is accepted for explicitly configured +trusted test endpoints but transmits credentials in plaintext; do not use it for production. +The authenticated instance URL is used for data requests. Redirects and HTTPS-to-HTTP +authentication downgrades are rejected in the sink path. + +`client_secret` and `security_token` are automatically masked in parsed-configuration logs. +This is log masking, not configuration encryption. Keep credentials out of checked-in job files. + +### Object and Field Mapping + +`object_name` is the Salesforce API object name, such as `Account`. +Input field names must match writable Salesforce API fields. Relationships, nested objects, +arrays, and automatic schema creation are not supported. + +`external_id_field` must exist in both Salesforce and the input schema. Configure it as an +external ID and preferably Unique in Salesforce. Each row must provide a non-null, non-blank +STRING, integer, or DECIMAL key. Salesforce `Id` is not accepted as the external ID or an input field. +The connector does not verify remote field metadata before sending; Salesforce validation or +permission errors fail the job. Null values in other columns explicitly clear those fields. + +BOOLEAN and numeric values remain native JSON values. DECIMAL is serialized without conversion +to double; the target field's configured precision still applies. Non-finite floats are rejected. +DATE uses ISO dates, TIME uses UTC time with milliseconds, TIMESTAMP without a zone is interpreted +as UTC, and TIMESTAMP_TZ preserves its explicit offset. Sub-millisecond temporal values are +rejected instead of truncated. BYTES uses Base64. Salesforce determines whether each value +is compatible with the configured target field. + +### Batching and Failures + +`batch_size` is 1-200 records per request. `batch_max_bytes` is the UTF-8 JSON request bound, +including the envelope, from 128 to 8388608 bytes. This is a client-side bound, not a promise +about Salesforce's request limits. A single oversized record fails before sending. + +Batches use `allOrNone = true`. Every record result must report success; malformed responses, +missing results, or any record error fail the task and prevent checkpoint completion. +Error diagnostics include result indices and Salesforce status codes, not record values or +raw response bodies. Per-record failures are not retried locally, including lock/validation errors. + +`max_retries` is 0-10 additional attempts after an I/O failure or HTTP 429, 500, 502, 503, or 504. +One HTTP 401 may refresh authentication within the same retry budget. Other HTTP errors fail +immediately. Authentication failures themselves are not retried automatically. This avoids +repeated login attempts with rejected credentials. Quota errors returned as HTTP 403 also fail +immediately. Retry-After is honored up to 60 seconds; +larger or invalid values fail the task rather than retrying earlier than the service permits. +`retry_interval_ms` is 0-60000. `request_timeout_ms` is a positive connection/pool/socket timeout, +not a whole-job deadline. Choose parallelism and retry settings within your org's API quota. +Flushing is synchronous, so size the checkpoint timeout to allow the complete flush, including +request timeouts, retry delays, and any authentication refresh. A single request timeout is not +the total flush budget, especially during sustained throttling or service failures. + +The writer flushes on count/byte bounds, checkpoint preparation, and normal close. It also +registers the engine flush callback. Zeta can enable periodic flushing through +`env.sink.flush.interval` (milliseconds; 0 disables it by default). Engines without that callback +rely on count/checkpoint/end of input. Streaming jobs should enable periodic checkpoints to +bound low-volume delivery latency. + +## Delivery Semantics + +Only `INSERT` and `UPDATE_AFTER` are accepted; `DELETE` and `UPDATE_BEFORE` fail rather than +being ignored. The sink does not reconcile deletes or external-ID changes. Silently discarding +before-images could hide an unsupported changelog pipeline and leave old Salesforce records +behind when a key changes. + +Delivery is at-least-once with a replay-capable source and checkpointing. An uncertain HTTP +outcome or task recovery can repeat an upsert and re-run Salesforce triggers/flows. This is not +an exactly-once or distributed-transaction sink; earlier successful requests cannot be rolled back. + +Repeated external IDs within one writer are flushed in input order in separate requests. +For ordered updates to the same key, use parallelism 1 or ensure all updates for that key are +routed in order to one writer. There is no global ordering across writers or recovery attempts. +Delete, insert-only, Bulk API ingestion, and multi-object routing are outside this connector slice. + +## Task Example + +Create `External_Id__c` on Account as a unique external ID and supply the following input columns: + +```hocon +sink { + Salesforce { + instance_url = "https://login.salesforce.com"; + client_id = "${SALESFORCE_CLIENT_ID}" + client_secret = "${SALESFORCE_CLIENT_SECRET}" + username = "${SALESFORCE_USERNAME}" + password = "${SALESFORCE_PASSWORD}" + security_token = "${SALESFORCE_SECURITY_TOKEN}" + object_name = "Account" + external_id_field = "External_Id__c" + batch_size = 200 + } +} +``` + +Use your deployment's substitution mechanism to supply these placeholders. +The input schema should include `External_Id__c = string` and `Name = string`. +See [Sink Common Options](../common-options/sink-common-options.md) for shared parameters. + +## Changelog + +<ChangeLog /> diff --git a/docs/zh/connectors/changelog/connector-salesforce.md b/docs/zh/connectors/changelog/connector-salesforce.md new file mode 100644 index 0000000000..d7789cd617 --- /dev/null +++ b/docs/zh/connectors/changelog/connector-salesforce.md @@ -0,0 +1,5 @@ +# 更新日志 + +| 变更 | 提交 | 版本 | +| --- | --- | --- | +| 添加基于外部 ID 的 REST collection upsert Sink | - | Next | diff --git a/docs/zh/connectors/sink/Salesforce.md b/docs/zh/connectors/sink/Salesforce.md new file mode 100644 index 0000000000..915fcdcd04 --- /dev/null +++ b/docs/zh/connectors/sink/Salesforce.md @@ -0,0 +1,138 @@ +import ChangeLog from '../changelog/connector-salesforce.md'; + +# Salesforce + +> 基于 Salesforce REST sObject Collections 的 upsert Sink + +## 描述 + +将一个输入表写入一个已存在的 Salesforce 对象，根据外部 ID 更新或插入记录。 +复用现有 Salesforce 模块，不改变 Source 配置和读取行为；不会创建对象、字段、外部 ID 或应用。 + +## 支持的引擎 + +> SeaTunnel Zeta<br/> +> Flink<br/> +> Spark<br/> + +## 功能 + +- [x] 批处理 +- [x] 流处理 +- [ ] exactly-once +- [ ] 完整 CDC +- [ ] 多表写入 + +流处理仅支持 `INSERT` 和 `UPDATE_AFTER`。本连接器不是 CDC Sink： +`UPDATE_BEFORE` 和 `DELETE` 会导致作业失败。连接变更日志 Source 前请阅读下方的投递语义。 + +## 配置 + +| 名称 | 类型 | 必填 | 默认值 | +| --- | --- | --- | --- | +| client_id | string | 是 | - | +| client_secret | string | 是 | - | +| username | string | 是 | - | +| password | string | 是 | - | +| security_token | string | 否 | 空字符串 | +| instance_url | string | 是 | - | +| api_version | string | 否 | v59.0 | +| object_name | string | 是 | - | +| external_id_field | string | 是 | - | +| batch_size | int | 否 | 200 | +| batch_max_bytes | int | 否 | 1048576 | +| request_timeout_ms | int | 否 | 60000 | +| max_retries | int | 否 | 3 | +| retry_interval_ms | long | 否 | 1000 | +| common-options | | 否 | - | + +### 认证 + +使用与 Source 相同的 OAuth 用户名密码流程，组织必须允许该认证方式。 +集成用户需要 API、对象创建/更新以及所有目标字段的写权限。security_token 追加在密码后。 +生产环境的 instance_url 应为 HTTPS 登录或组织地址，例如 [Salesforce 登录地址](https://login.salesforce.com)， +不包含末尾斜杠、路径、内嵌凭证、查询参数或片段。HTTP 仅适用于显式配置的可信测试端点， +会明文传输凭证，不应在生产环境使用。数据请求使用认证响应中的实例地址。 +Sink 不跟随重定向，也不接受认证地址从 HTTPS 降级到 HTTP。 + +client_secret 和 security_token 自动在解析后的配置日志中掩码；这不是配置加密。 +不要将真实凭证提交到配置文件。 + +### 字段映射 + +object_name 为 Salesforce API 对象名，例如 Account。输入列名应与可写 API 字段匹配。 +不支持嵌套对象、数组、关系对象或自动创建表结构。 +external_id_field 必须存在于输入结构和 Salesforce 中，并在 Salesforce 中配置为外部 ID， +建议同时启用 Unique。每行键必须为非空、非空白的 STRING、整数或 DECIMAL。 +不接受 Salesforce Id 作为外部 ID 或输入列。 + +不预先查询远端字段元数据，权限、校验或字段不兼容由 Salesforce 返回并导致作业失败。 +其他字段的 null 会明确清空目标值。布尔和数值使用 JSON 原生类型，DECIMAL 不经过 double 转换； +远端字段的精度限制仍适用。拒绝非有限浮点数。 +DATE 使用 ISO 日期，TIME 使用 UTC 毫秒格式，无时区 TIMESTAMP 按 UTC 解释， +TIMESTAMP_TZ 保留显式偏移量。拒绝亚毫秒时间值而不是静默截断；BYTES 使用 Base64。 + +### 批次和错误 + +batch_size 范围为 1-200。batch_max_bytes 限制包含封装在内的 UTF-8 JSON 请求大小， +范围为 128-8388608 字节，这是客户端限制而不是 Salesforce 限制的声明。 +单条记录超限时在发送前失败。 + +固定使用 allOrNone=true，逐条检查响应。任意失败、缺失结果或非法响应均导致任务失败， +阻止检查点完成。错误只包含结果索引和状态码，不包含原始响应或字段值。 +单条记录错误（包括锁冲突、校验错误）不会在客户端局部重试。 + +max_retries 为额外尝试次数（0-10），仅重试 I/O 错误或 HTTP 429、500、502、503、504。 +在同一预算内允许一次 HTTP 401 重新认证。包括配额限制 HTTP 403 在内的其他错误立即失败。 +认证请求自身失败时不自动重试，避免使用被拒绝的凭证反复登录。 +遵守不超过 60 秒的 Retry-After；无效或更大的值直接失败，不提前重试。 +retry_interval_ms 范围 0-60000。request_timeout_ms 必须为正，是连接、连接池和 socket 超时， +不是整个任务的截止时间。并行度和重试次数需要符合组织 API 配额。 +刷新是同步执行的，检查点超时需要覆盖完整刷新过程，包括请求超时、重试等待和可能发生的重新认证。 +单次请求超时并不等于总刷新预算，尤其是在持续限流或服务故障期间。 + +达到批次数量/字节限制、检查点准备或正常关闭时刷新。 +同时注册引擎刷新回调。Zeta 可通过 env.sink.flush.interval 启用周期刷新（毫秒，默认 0 表示禁用）。 +未实现此回调的引擎依赖数量限制、检查点和输入结束刷新。 +流作业应启用周期性检查点以限制低流量下的延迟。 + +## 投递语义 + +仅接受 `INSERT` 和 `UPDATE_AFTER`；`DELETE` 和 `UPDATE_BEFORE` 会失败而不是被忽略。 +本连接器不处理删除或外部 ID 变更的旧记录清理。静默丢弃更新前镜像可能掩盖不受支持的变更日志输入， +并在键变更后将旧记录留在 Salesforce 中。 + +配合可回放 Source 和检查点提供至少一次语义。不确定的 HTTP 结果和恢复可能再次 upsert， +重复执行触发器或 Flow，因此不保证 exactly-once，也不能回滚之前成功的请求。 + +单个 writer 内重复外部 ID 按输入顺序分批发送。同键有序更新应使用并行度 1， +或确保同键更新有序路由到同一 writer；不同 writer 和恢复尝试之间不保证全局顺序。 +删除、仅插入模式、Bulk API 和多对象路由不属于本次范围。 + +## 示例 + +先在 Account 中创建启用 Unique 的外部 ID 字段 External_Id__c， +输入包含 External_Id__c=string 和 Name=string： + +```hocon +sink { + Salesforce { + instance_url = "https://login.salesforce.com"; + client_id = "${SALESFORCE_CLIENT_ID}" + client_secret = "${SALESFORCE_CLIENT_SECRET}" + username = "${SALESFORCE_USERNAME}" + password = "${SALESFORCE_PASSWORD}" + security_token = "${SALESFORCE_SECURITY_TOKEN}" + object_name = "Account" + external_id_field = "External_Id__c" + batch_size = 200 + } +} +``` + +请使用部署环境的变量替换机制提供示例占位符。 +通用配置请参阅 [Sink Common Options](../common-options/sink-common-options.md)。 + +## 更新日志 + +<ChangeLog /> diff --git a/plugin-mapping.properties b/plugin-mapping.properties index 48e5d11c3d..08fd3b0494 100644 --- a/plugin-mapping.properties +++ b/plugin-mapping.properties @@ -93,6 +93,7 @@ seatunnel.source.AzureCosmosDB = connector-azurecosmosdb seatunnel.source.Cassandra = connector-cassandra seatunnel.sink.Cassandra = connector-cassandra seatunnel.source.Salesforce = connector-salesforce +seatunnel.sink.Salesforce = connector-salesforce seatunnel.sink.StarRocks = connector-starrocks seatunnel.source.MyHours = connector-http-myhours seatunnel.sink.InfluxDB = connector-influxdb diff --git a/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/client/SalesforceClient.java b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/client/SalesforceClient.java index 03aa7c5509..f7d8183829 100644 --- a/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/client/SalesforceClient.java +++ b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/client/SalesforceClient.java @@ -30,6 +30,7 @@ import org.apache.seatunnel.api.table.type.LocalTimeType; import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType; import org.apache.seatunnel.api.table.type.SeaTunnelDataType; import org.apache.seatunnel.connectors.seatunnel.salesforce.config.SalesforceParameters; +import org.apache.seatunnel.connectors.seatunnel.salesforce.config.SalesforceSinkConfig; import org.apache.seatunnel.connectors.seatunnel.salesforce.exception.SalesforceConnectorErrorCode; import org.apache.seatunnel.connectors.seatunnel.salesforce.exception.SalesforceConnectorException; @@ -43,22 +44,28 @@ import org.apache.http.client.config.RequestConfig; import org.apache.http.client.entity.UrlEncodedFormEntity; import org.apache.http.client.methods.CloseableHttpResponse; import org.apache.http.client.methods.HttpGet; +import org.apache.http.client.methods.HttpPatch; import org.apache.http.client.methods.HttpPost; +import org.apache.http.client.utils.DateUtils; import org.apache.http.entity.ContentType; import org.apache.http.entity.StringEntity; import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClientBuilder; import org.apache.http.impl.client.HttpClients; import org.apache.http.message.BasicNameValuePair; import org.apache.http.util.EntityUtils; import lombok.extern.slf4j.Slf4j; +import java.io.ByteArrayOutputStream; import java.io.Closeable; import java.io.IOException; +import java.io.InputStream; import java.io.StringReader; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Collections; +import java.util.Date; import java.util.List; import java.util.function.Consumer; @@ -75,17 +82,33 @@ public class SalesforceClient implements Closeable { private final SalesforceParameters params; private final CloseableHttpClient httpClient; private final ObjectMapper objectMapper = new ObjectMapper(); + private final boolean sinkMode; private String accessToken; private String authorizedInstanceUrl; public SalesforceClient(SalesforceParameters params) { + this(params, false); + } + + /** Create a write client with explicit retries and no credential-forwarding redirects. */ + public static SalesforceClient forSink(SalesforceParameters params) { + return new SalesforceClient(params, true); + } + + private SalesforceClient(SalesforceParameters params, boolean sinkMode) { this.params = params; + this.sinkMode = sinkMode; RequestConfig requestConfig = RequestConfig.custom() .setConnectTimeout(params.getRequestTimeoutMs()) .setSocketTimeout(params.getRequestTimeoutMs()) + .setConnectionRequestTimeout(sinkMode ? params.getRequestTimeoutMs() : -1) .build(); - this.httpClient = HttpClients.custom().setDefaultRequestConfig(requestConfig).build(); + HttpClientBuilder builder = HttpClients.custom().setDefaultRequestConfig(requestConfig); + if (sinkMode) { + builder.disableAutomaticRetries().disableRedirectHandling(); + } + this.httpClient = builder.build(); } public void authenticate() { @@ -105,13 +128,32 @@ public class SalesforceClient implements Closeable { post.setEntity(new UrlEncodedFormEntity(form, StandardCharsets.UTF_8)); try (CloseableHttpResponse response = httpClient.execute(post)) { int status = response.getStatusLine().getStatusCode(); - String body = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8); + String body = + sinkMode + ? readSinkBody(response) + : EntityUtils.toString( + response.getEntity(), StandardCharsets.UTF_8); if (status != 200) { throw new SalesforceConnectorException( SalesforceConnectorErrorCode.AUTH_FAILED, - "HTTP " + status + ": " + body); + "HTTP " + status + (sinkMode ? "" : ": " + body)); } JsonNode json = objectMapper.readTree(body); + if (sinkMode) { + if (json == null + || !json.path("access_token").isTextual() + || json.path("access_token").asText().isEmpty() + || !json.path("instance_url").isTextual()) { + throw new IOException("Invalid authentication response"); + } + SalesforceSinkConfig.validateInstanceUrl(json.path("instance_url").asText()); + if (params.getInstanceUrl().regionMatches(true, 0, "https:", 0, 6) + && !json.path("instance_url") + .asText() + .regionMatches(true, 0, "https:", 0, 6)) { + throw new IOException("Authentication returned an insecure instance"); + } + } this.accessToken = json.get("access_token").asText(); this.authorizedInstanceUrl = json.get("instance_url").asText(); log.info("Authenticated with Salesforce instance {}", authorizedInstanceUrl); @@ -119,10 +161,185 @@ public class SalesforceClient implements Closeable { } catch (SalesforceConnectorException e) { throw e; } catch (Exception e) { + if (sinkMode) { + throw new SalesforceConnectorException( + SalesforceConnectorErrorCode.AUTH_FAILED, + "Authentication failed (" + e.getClass().getSimpleName() + ")"); + } throw new SalesforceConnectorException(SalesforceConnectorErrorCode.AUTH_FAILED, e); } } + /** + * Upsert one bounded collection with allOrNone=true. Request retries can repeat Salesforce-side + * effects; this is not a checkpoint transaction or exactly-once protocol. + */ + public void upsert( + String objectName, + String externalIdField, + List<ObjectNode> records, + int maxRetries, + long retryIntervalMs) { + if (!sinkMode || accessToken == null) { + throw writeFailure("An authenticated sink client is required"); + } + SalesforceSinkConfig.requireIdentifier(objectName, "object_name"); + SalesforceSinkConfig.requireIdentifier(externalIdField, "external_id_field"); + if (records.isEmpty() + || records.size() > 200 + || maxRetries < 0 + || maxRetries > 10 + || retryIntervalMs < 0 + || retryIntervalMs > 60000) { + throw writeFailure("Invalid upsert batch or retry bounds"); + } + ObjectNode payload = objectMapper.createObjectNode(); + payload.put("allOrNone", true); + records.forEach(payload.putArray("records")::add); + String requestBody = payload.toString(); + boolean refreshed = false; + for (int attempt = 0; attempt <= maxRetries; attempt++) { + if (Thread.currentThread().isInterrupted()) { + throw writeFailure("Interrupted before collection upsert"); + } + long delay = retryIntervalMs; + HttpPatch patch = + new HttpPatch( + authorizedInstanceUrl + + "/services/data/" + + params.getApiVersion() + + "/composite/sobjects/" + + objectName + + "/" + + externalIdField); + patch.setHeader(HttpHeaders.AUTHORIZATION, "Bearer " + accessToken); + patch.setEntity(new StringEntity(requestBody, ContentType.APPLICATION_JSON)); + boolean refresh = false; + try (CloseableHttpResponse response = httpClient.execute(patch)) { + int status = response.getStatusLine().getStatusCode(); + if (status == 200) { + validateUpsertResults(readSinkBody(response), records.size()); + return; + } + if (status == 401 && !refreshed && attempt < maxRetries) { + refresh = true; + } else if ((status == 429 + || status == 500 + || status == 502 + || status == 503 + || status == 504) + && attempt < maxRetries) { + delay = retryDelay(response.getFirstHeader("Retry-After"), retryIntervalMs); + } else { + throw writeFailure( + "Collection upsert returned HTTP " + + status + + " after " + + (attempt + 1) + + " attempt(s)"); + } + } catch (IOException e) { + if (Thread.currentThread().isInterrupted() || attempt == maxRetries) { + throw writeFailure( + "Collection upsert I/O failed (" + + e.getClass().getSimpleName() + + ") after " + + (attempt + 1) + + " attempt(s)"); + } + } + if (refresh) { + authenticate(); + refreshed = true; + } else { + try { + Thread.sleep(delay); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw writeFailure("Interrupted while waiting to retry collection upsert"); + } + } + } + throw writeFailure("Collection upsert exhausted retries"); + } + + private void validateUpsertResults(String body, int count) { + JsonNode results; + try { + results = objectMapper.readTree(body); + } catch (IOException e) { + throw writeFailure("Invalid collection upsert JSON response"); + } + if (results == null || !results.isArray() || results.size() != count) { + throw writeFailure("Collection upsert response count does not match the request"); + } + List<String> failures = new ArrayList<>(); + for (int i = 0; i < count; i++) { + JsonNode result = results.get(i); + if (!result.path("success").isBoolean() || !result.path("errors").isArray()) { + throw writeFailure("Malformed collection result at index " + i); + } + if (!result.path("success").asBoolean() || result.path("errors").size() != 0) { + List<String> codes = new ArrayList<>(); + for (JsonNode error : result.path("errors")) { + String code = error.path("statusCode").asText(); + codes.add(code.matches("[A-Z_]{1,80}") ? code : "UNKNOWN"); + } + failures.add("index " + i + ": " + codes); + } else if (!result.path("id").isTextual() || result.path("id").asText().isEmpty()) { + throw writeFailure("Missing record ID in collection result at index " + i); + } + } + if (!failures.isEmpty()) { + // API messages can contain field values; report indices and codes, not response bodies. + throw writeFailure("Collection upsert failed: " + failures); + } + } + + private long retryDelay(Header retryAfter, long configuredDelay) { + if (retryAfter == null) { + return configuredDelay; + } + long delay; + try { + delay = Math.multiplyExact(Long.parseLong(retryAfter.getValue().trim()), 1000L); + } catch (NumberFormatException e) { + Date date = DateUtils.parseDate(retryAfter.getValue()); + if (date == null) { + throw writeFailure("Invalid Retry-After response header"); + } + delay = Math.max(0, date.getTime() - System.currentTimeMillis()); + } catch (ArithmeticException e) { + throw writeFailure("Retry-After exceeds the bounded retry budget"); + } + if (delay < 0 || delay > 60000) { + throw writeFailure("Retry-After exceeds the bounded retry budget"); + } + return Math.max(delay, configuredDelay); + } + + private String readSinkBody(CloseableHttpResponse response) throws IOException { + if (response.getEntity() == null) { + throw writeFailure("Missing Salesforce response body"); + } + try (InputStream input = response.getEntity().getContent(); + ByteArrayOutputStream output = new ByteArrayOutputStream()) { + byte[] buffer = new byte[4096]; + int read; + while ((read = input.read(buffer)) != -1) { + if (output.size() + read > 1024 * 1024) { + throw writeFailure("Salesforce response exceeds the 1 MiB client limit"); + } + output.write(buffer, 0, read); + } + return new String(output.toByteArray(), StandardCharsets.UTF_8); + } + } + + private SalesforceConnectorException writeFailure(String message) { + return new SalesforceConnectorException(SalesforceConnectorErrorCode.WRITE_FAILED, message); + } + public CatalogTable describeObject(String database, String objectName) { String url = authorizedInstanceUrl diff --git a/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/config/SalesforceSinkConfig.java b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/config/SalesforceSinkConfig.java new file mode 100644 index 0000000000..372ae44b7d --- /dev/null +++ b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/config/SalesforceSinkConfig.java @@ -0,0 +1,107 @@ +/* + * 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.salesforce.config; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; + +import lombok.Getter; + +import java.io.Serializable; +import java.net.URI; + +@Getter +public final class SalesforceSinkConfig implements Serializable { + private static final long serialVersionUID = 1L; + public static final String IDENTIFIER_PATTERN = "[A-Za-z][A-Za-z0-9_]*"; + + private final SalesforceParameters parameters; + private final String objectName; + private final String externalIdField; + private final int batchSize; + private final int batchMaxBytes; + private final int maxRetries; + private final long retryIntervalMs; + + public SalesforceSinkConfig(ReadonlyConfig config) { + parameters = new SalesforceParameters(); + parameters.buildWithConfig(config); + requireText(parameters.getClientId(), "client_id"); + requireText(parameters.getClientSecret(), "client_secret"); + requireText(parameters.getUsername(), "username"); + requireText(parameters.getPassword(), "password"); + validateInstanceUrl(parameters.getInstanceUrl()); + if (!parameters.getApiVersion().matches("v[0-9]+\\.[0-9]+")) { + throw new IllegalArgumentException("api_version must have the form v59.0"); + } + if (parameters.getRequestTimeoutMs() <= 0) { + throw new IllegalArgumentException("request_timeout_ms must be positive"); + } + objectName = config.get(SalesforceSourceOptions.OBJECT_NAME); + externalIdField = config.get(SalesforceSinkOptions.EXTERNAL_ID_FIELD); + requireIdentifier(objectName, "object_name"); + requireIdentifier(externalIdField, "external_id_field"); + if ("Id".equalsIgnoreCase(externalIdField)) { + throw new IllegalArgumentException( + "external_id_field must be an external ID, not the Salesforce Id"); + } + batchSize = config.get(SalesforceSinkOptions.BATCH_SIZE); + batchMaxBytes = config.get(SalesforceSinkOptions.BATCH_MAX_BYTES); + maxRetries = config.get(SalesforceSinkOptions.MAX_RETRIES); + retryIntervalMs = config.get(SalesforceSinkOptions.RETRY_INTERVAL_MS); + requireRange(batchSize, 1, 200, "batch_size"); + requireRange(batchMaxBytes, 128, 8 * 1024 * 1024, "batch_max_bytes"); + requireRange(maxRetries, 0, 10, "max_retries"); + requireRange(retryIntervalMs, 0, 60000, "retry_interval_ms"); + } + + public static void requireIdentifier(String value, String option) { + if (value == null || !value.matches(IDENTIFIER_PATTERN)) { + throw new IllegalArgumentException(option + " must be a Salesforce API identifier"); + } + } + + public static void validateInstanceUrl(String value) { + try { + URI uri = URI.create(value); + if ((!"http".equalsIgnoreCase(uri.getScheme()) + && !"https".equalsIgnoreCase(uri.getScheme())) + || uri.getHost() == null + || uri.getUserInfo() != null + || uri.getRawQuery() != null + || uri.getRawFragment() != null + || (uri.getPath() != null && !uri.getPath().isEmpty())) { + throw new IllegalArgumentException(); + } + } catch (RuntimeException invalid) { + throw new IllegalArgumentException( + "instance_url must be an HTTP(S) origin without credentials, path, query or fragment"); + } + } + + private static void requireText(String value, String option) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException(option + " must not be blank"); + } + } + + private static void requireRange(long value, long min, long max, String option) { + if (value < min || value > max) { + throw new IllegalArgumentException(option + " must be between " + min + " and " + max); + } + } +} diff --git a/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/config/SalesforceSinkOptions.java b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/config/SalesforceSinkOptions.java new file mode 100644 index 0000000000..7e9fb8708d --- /dev/null +++ b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/config/SalesforceSinkOptions.java @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.salesforce.config; + +import org.apache.seatunnel.api.configuration.Option; +import org.apache.seatunnel.api.configuration.Options; + +public final class SalesforceSinkOptions { + private SalesforceSinkOptions() {} + + public static final Option<String> EXTERNAL_ID_FIELD = + Options.key("external_id_field") + .stringType() + .noDefaultValue() + .withDescription("Salesforce external ID field present in every input row."); + public static final Option<Integer> BATCH_SIZE = + Options.key("batch_size") + .intType() + .defaultValue(200) + .withDescription("Maximum records per REST collection upsert, from 1 to 200."); + public static final Option<Integer> BATCH_MAX_BYTES = + Options.key("batch_max_bytes") + .intType() + .defaultValue(1024 * 1024) + .withDescription( + "Maximum serialized request size in bytes, including the JSON envelope. This is a client memory bound, not a Salesforce API limit."); + public static final Option<Integer> MAX_RETRIES = + Options.key("max_retries") + .intType() + .defaultValue(3) + .withDescription( + "Maximum retries per upsert request, from 0 to 10. Record-level failures are not retried."); + public static final Option<Long> RETRY_INTERVAL_MS = + Options.key("retry_interval_ms") + .longType() + .defaultValue(1000L) + .withDescription("Delay between retries, from 0 to 60000 milliseconds."); +} diff --git a/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/exception/SalesforceConnectorErrorCode.java b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/exception/SalesforceConnectorErrorCode.java index 0ee618b4c0..1717d5f077 100644 --- a/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/exception/SalesforceConnectorErrorCode.java +++ b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/exception/SalesforceConnectorErrorCode.java @@ -26,7 +26,8 @@ public enum SalesforceConnectorErrorCode implements SeaTunnelErrorCode { BULK_JOB_FAILED("SALESFORCE-04", "Salesforce Bulk API query job failed or was aborted"), BULK_RESULTS_FAILED("SALESFORCE-05", "Failed to retrieve Salesforce Bulk API job results"), INVALID_TABLE_PATH("SALESFORCE-06", "Invalid table_path; expected format: database.ObjectName"), - DUPLICATE_OBJECT("SALESFORCE-07", "Duplicate object found in tables_configs"); + DUPLICATE_OBJECT("SALESFORCE-07", "Duplicate object found in tables_configs"), + WRITE_FAILED("SALESFORCE-08", "Failed to upsert Salesforce records"); private final String code; private final String description; diff --git a/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceRowSerializer.java b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceRowSerializer.java new file mode 100644 index 0000000000..1101afbe89 --- /dev/null +++ b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceRowSerializer.java @@ -0,0 +1,206 @@ +/* + * 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.salesforce.sink; + +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode; + +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.api.table.type.SqlType; +import org.apache.seatunnel.connectors.seatunnel.salesforce.config.SalesforceSinkConfig; + +import java.math.BigDecimal; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import java.time.format.DateTimeFormatter; +import java.util.HashSet; +import java.util.Locale; +import java.util.Set; + +final class SalesforceRowSerializer { + private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final DateTimeFormatter TIME_FORMAT = + DateTimeFormatter.ofPattern("HH:mm:ss.SSS'Z'"); + private final SeaTunnelRowType rowType; + private final SalesforceSinkConfig config; + + SalesforceRowSerializer(SeaTunnelRowType rowType, SalesforceSinkConfig config) { + validateSchema(rowType, config); + this.rowType = rowType; + this.config = config; + } + + static void validateSchema(SeaTunnelRowType rowType, SalesforceSinkConfig config) { + boolean externalIdFound = false; + Set<String> names = new HashSet<>(); + for (int i = 0; i < rowType.getTotalFields(); i++) { + String name = rowType.getFieldName(i); + SalesforceSinkConfig.requireIdentifier(name, "Input field"); + if (!names.add(name.toLowerCase(Locale.ROOT)) + || "attributes".equalsIgnoreCase(name) + || "Id".equalsIgnoreCase(name)) { + throw new IllegalArgumentException( + "Input schema contains a duplicate or reserved Salesforce field: " + name); + } + SqlType type = rowType.getFieldType(i).getSqlType(); + switch (type) { + case STRING: + case BOOLEAN: + case TINYINT: + case SMALLINT: + case INT: + case BIGINT: + case FLOAT: + case DOUBLE: + case DECIMAL: + case DATE: + case TIME: + case TIMESTAMP: + case TIMESTAMP_TZ: + case BYTES: + break; + default: + throw new IllegalArgumentException( + "Unsupported Salesforce field type: " + name + " (" + type + ")"); + } + if (name.equals(config.getExternalIdField())) { + externalIdFound = true; + if (type != SqlType.STRING + && type != SqlType.TINYINT + && type != SqlType.SMALLINT + && type != SqlType.INT + && type != SqlType.BIGINT + && type != SqlType.DECIMAL) { + throw new IllegalArgumentException( + "external_id_field must be STRING, an integer type or DECIMAL"); + } + } + } + if (!externalIdFound) { + throw new IllegalArgumentException( + "external_id_field must be present in the input schema"); + } + } + + ObjectNode serialize(SeaTunnelRow row) { + if (row.getFields().length != rowType.getTotalFields()) { + throw new IllegalArgumentException( + "Input row does not match the Salesforce sink schema"); + } + ObjectNode record = MAPPER.createObjectNode(); + record.putObject("attributes").put("type", config.getObjectName()); + for (int i = 0; i < rowType.getTotalFields(); i++) { + String name = rowType.getFieldName(i); + Object value = row.getField(i); + if (value == null) { + record.putNull(name); + continue; + } + switch (rowType.getFieldType(i).getSqlType()) { + case STRING: + record.put(name, (String) value); + break; + case BOOLEAN: + record.put(name, (Boolean) value); + break; + case TINYINT: + case SMALLINT: + case INT: + case BIGINT: + record.put(name, ((Number) value).longValue()); + break; + case FLOAT: + float floatValue = ((Number) value).floatValue(); + if (!Float.isFinite(floatValue)) { + throw invalidNumber(name); + } + record.put(name, floatValue); + break; + case DOUBLE: + double doubleValue = ((Number) value).doubleValue(); + if (!Double.isFinite(doubleValue)) { + throw invalidNumber(name); + } + record.put(name, doubleValue); + break; + case DECIMAL: + record.put(name, (BigDecimal) value); + break; + case DATE: + record.put(name, ((LocalDate) value).toString()); + break; + case TIME: + LocalTime time = (LocalTime) value; + requireMilliseconds(time.getNano(), name); + record.put(name, time.format(TIME_FORMAT)); + break; + case TIMESTAMP: + requireMilliseconds(((LocalDateTime) value).getNano(), name); + record.put( + name, + ((LocalDateTime) value) + .atOffset(ZoneOffset.UTC) + .format(DateTimeFormatter.ISO_OFFSET_DATE_TIME)); + break; + case TIMESTAMP_TZ: + requireMilliseconds(((OffsetDateTime) value).getNano(), name); + record.put( + name, + ((OffsetDateTime) value) + .format(DateTimeFormatter.ISO_OFFSET_DATE_TIME)); + break; + case BYTES: + record.put(name, ((byte[]) value).clone()); + break; + default: + throw new IllegalArgumentException("Unsupported Salesforce field: " + name); + } + } + if (record.path(config.getExternalIdField()).isNull() + || record.path(config.getExternalIdField()).asText().trim().isEmpty()) { + throw new IllegalArgumentException( + "external_id_field must be non-null and non-blank in every row"); + } + return record; + } + + String externalIdKey(ObjectNode record) { + if (record.path(config.getExternalIdField()).isNumber()) { + return record.path(config.getExternalIdField()) + .decimalValue() + .stripTrailingZeros() + .toPlainString(); + } + return record.path(config.getExternalIdField()).asText().toLowerCase(Locale.ROOT); + } + + private IllegalArgumentException invalidNumber(String name) { + return new IllegalArgumentException("Non-finite number in Salesforce field " + name); + } + + private static void requireMilliseconds(int nanos, String name) { + if (nanos % 1_000_000 != 0) { + throw new IllegalArgumentException( + "Salesforce temporal fields support millisecond precision: " + name); + } + } +} diff --git a/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceSink.java b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceSink.java new file mode 100644 index 0000000000..836ad0d3ca --- /dev/null +++ b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceSink.java @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.connectors.seatunnel.salesforce.sink; + +import org.apache.seatunnel.api.sink.SinkWriter; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSimpleSink; +import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter; +import org.apache.seatunnel.connectors.seatunnel.salesforce.config.SalesforceSinkConfig; + +import java.io.IOException; +import java.util.Optional; + +public class SalesforceSink extends AbstractSimpleSink<SeaTunnelRow, Void> { + private final SalesforceSinkConfig config; + private final CatalogTable catalogTable; + + public SalesforceSink(SalesforceSinkConfig config, CatalogTable catalogTable) { + this.config = config; + this.catalogTable = catalogTable; + } + + @Override + public String getPluginName() { + return "Salesforce"; + } + + @Override + public AbstractSinkWriter<SeaTunnelRow, Void> createWriter(SinkWriter.Context context) + throws IOException { + SalesforceSinkWriter writer = + new SalesforceSinkWriter(catalogTable.getSeaTunnelRowType(), config); + context.registerFlushAction(writer::flush); + return writer; + } + + @Override + public Optional<CatalogTable> getWriteCatalogTable() { + return Optional.of(catalogTable); + } +} diff --git a/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceSinkFactory.java b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceSinkFactory.java new file mode 100644 index 0000000000..31b9d7338c --- /dev/null +++ b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceSinkFactory.java @@ -0,0 +1,91 @@ +/* + * 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.salesforce.sink; + +import org.apache.seatunnel.api.configuration.util.Conditions; +import org.apache.seatunnel.api.configuration.util.OptionRule; +import org.apache.seatunnel.api.table.connector.TableSink; +import org.apache.seatunnel.api.table.factory.Factory; +import org.apache.seatunnel.api.table.factory.TableSinkFactory; +import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext; +import org.apache.seatunnel.connectors.seatunnel.salesforce.config.SalesforceSinkConfig; +import org.apache.seatunnel.connectors.seatunnel.salesforce.config.SalesforceSinkOptions; +import org.apache.seatunnel.connectors.seatunnel.salesforce.config.SalesforceSourceOptions; + +import com.google.auto.service.AutoService; + +@AutoService(Factory.class) +public class SalesforceSinkFactory implements TableSinkFactory { + @Override + public String factoryIdentifier() { + return "Salesforce"; + } + + @Override + public OptionRule optionRule() { + return OptionRule.builder() + .required( + SalesforceSourceOptions.CLIENT_ID, + SalesforceSourceOptions.CLIENT_SECRET, + SalesforceSourceOptions.USERNAME, + SalesforceSourceOptions.PASSWORD, + SalesforceSourceOptions.INSTANCE_URL) + .required( + SalesforceSourceOptions.OBJECT_NAME, + Conditions.matches( + SalesforceSourceOptions.OBJECT_NAME, + SalesforceSinkConfig.IDENTIFIER_PATTERN)) + .required( + SalesforceSinkOptions.EXTERNAL_ID_FIELD, + Conditions.matches( + SalesforceSinkOptions.EXTERNAL_ID_FIELD, + SalesforceSinkConfig.IDENTIFIER_PATTERN)) + .optional( + SalesforceSourceOptions.SECURITY_TOKEN, SalesforceSourceOptions.API_VERSION) + .optional( + SalesforceSourceOptions.REQUEST_TIMEOUT_MS, + Conditions.greaterThan(SalesforceSourceOptions.REQUEST_TIMEOUT_MS, 0)) + .optional( + SalesforceSinkOptions.BATCH_SIZE, + Conditions.greaterThan(SalesforceSinkOptions.BATCH_SIZE, 0), + Conditions.lessThan(SalesforceSinkOptions.BATCH_SIZE, 201)) + .optional( + SalesforceSinkOptions.BATCH_MAX_BYTES, + Conditions.greaterThan(SalesforceSinkOptions.BATCH_MAX_BYTES, 127), + Conditions.lessThan( + SalesforceSinkOptions.BATCH_MAX_BYTES, 8 * 1024 * 1024 + 1)) + .optional( + SalesforceSinkOptions.MAX_RETRIES, + Conditions.greaterThan(SalesforceSinkOptions.MAX_RETRIES, -1), + Conditions.lessThan(SalesforceSinkOptions.MAX_RETRIES, 11)) + .optional( + SalesforceSinkOptions.RETRY_INTERVAL_MS, + Conditions.greaterThan(SalesforceSinkOptions.RETRY_INTERVAL_MS, -1L), + Conditions.lessThan(SalesforceSinkOptions.RETRY_INTERVAL_MS, 60001L)) + .build(); + } + + @Override + public TableSink createSink(TableSinkFactoryContext context) { + SalesforceSinkConfig config = new SalesforceSinkConfig(context.getOptions()); + // Fail during planning; the serializer also validates direct writer construction. + SalesforceRowSerializer.validateSchema( + context.getCatalogTable().getSeaTunnelRowType(), config); + return () -> new SalesforceSink(config, context.getCatalogTable()); + } +} diff --git a/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceSinkWriter.java b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceSinkWriter.java new file mode 100644 index 0000000000..c81e239bd0 --- /dev/null +++ b/seatunnel-connectors-v2/connector-salesforce/src/main/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceSinkWriter.java @@ -0,0 +1,174 @@ +/* + * 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.salesforce.sink; + +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode; + +import org.apache.seatunnel.api.table.type.RowKind; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter; +import org.apache.seatunnel.connectors.seatunnel.salesforce.client.SalesforceClient; +import org.apache.seatunnel.connectors.seatunnel.salesforce.config.SalesforceSinkConfig; +import org.apache.seatunnel.connectors.seatunnel.salesforce.exception.SalesforceConnectorErrorCode; +import org.apache.seatunnel.connectors.seatunnel.salesforce.exception.SalesforceConnectorException; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Optional; +import java.util.Set; + +/** + * Buffers detached rows and verifies every upsert result before allowing a checkpoint to advance. + */ +public final class SalesforceSinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void> { + private static final ObjectMapper MAPPER = new ObjectMapper(); + // Exact UTF-8 size of {"allOrNone":true,"records":[]} before adding records and commas. + private static final int ENVELOPE_BYTES = 31; + private final SalesforceSinkConfig config; + private final SalesforceRowSerializer serializer; + private final SalesforceClient client; + private final List<ObjectNode> batch = new ArrayList<>(); + private final Set<String> externalIds = new HashSet<>(); + private int batchBytes = ENVELOPE_BYTES; + private RuntimeException failure; + private boolean closed; + + public SalesforceSinkWriter(SeaTunnelRowType rowType, SalesforceSinkConfig config) + throws IOException { + this.config = config; + this.serializer = new SalesforceRowSerializer(rowType, config); + this.client = SalesforceClient.forSink(config.getParameters()); + try { + client.authenticate(); + } catch (RuntimeException authenticationFailure) { + try { + client.close(); + } catch (IOException closeFailure) { + authenticationFailure.addSuppressed(closeFailure); + } + throw authenticationFailure; + } + } + + @Override + public synchronized void write(SeaTunnelRow row) throws IOException { + checkOpen(); + try { + if (row.getRowKind() != RowKind.INSERT && row.getRowKind() != RowKind.UPDATE_AFTER) { + throw new IllegalArgumentException( + "Salesforce sink supports INSERT and UPDATE_AFTER only"); + } + ObjectNode record = serializer.serialize(row); + int bytes = MAPPER.writeValueAsBytes(record).length; + if ((long) ENVELOPE_BYTES + bytes > config.getBatchMaxBytes()) { + throw new IllegalArgumentException("A Salesforce record exceeds batch_max_bytes"); + } + String key = serializer.externalIdKey(record); + if (!batch.isEmpty() + && (externalIds.contains(key) + || (long) batchBytes + 1 + bytes > config.getBatchMaxBytes())) { + flush(); + } + batchBytes += bytes + (batch.isEmpty() ? 0 : 1); + batch.add(record); + externalIds.add(key); + if (batch.size() >= config.getBatchSize()) { + flush(); + } + } catch (RuntimeException e) { + failure = e; + throw e; + } catch (IOException e) { + failure = + new SalesforceConnectorException( + SalesforceConnectorErrorCode.WRITE_FAILED, + "Could not serialize a Salesforce record"); + throw failure; + } + } + + /** A checkpoint can advance only after all buffered records have succeeded remotely. */ + @Override + public synchronized Optional<Void> prepareCommit() { + flush(); + return Optional.empty(); + } + + /** Flush on the writer/engine thread; no connector-owned timer or executor is created. */ + public synchronized void flush() { + checkOpen(); + if (batch.isEmpty()) { + return; + } + try { + client.upsert( + config.getObjectName(), + config.getExternalIdField(), + batch, + config.getMaxRetries(), + config.getRetryIntervalMs()); + batch.clear(); + externalIds.clear(); + batchBytes = ENVELOPE_BYTES; + } catch (RuntimeException e) { + failure = e; + throw e; + } + } + + @Override + public synchronized void close() throws IOException { + if (closed) { + return; + } + RuntimeException flushFailure = null; + try { + // A failed batch must not be silently retried during abort/cleanup. + if (failure == null) { + flush(); + } + } catch (RuntimeException e) { + flushFailure = e; + throw e; + } finally { + closed = true; + try { + client.close(); + } catch (IOException closeFailure) { + if (flushFailure != null) { + flushFailure.addSuppressed(closeFailure); + } else { + throw closeFailure; + } + } + } + } + + private void checkOpen() { + if (failure != null) { + throw failure; + } + if (closed) { + throw new IllegalStateException("Salesforce sink writer is closed"); + } + } +} diff --git a/seatunnel-connectors-v2/connector-salesforce/src/test/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceRowSerializerTest.java b/seatunnel-connectors-v2/connector-salesforce/src/test/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceRowSerializerTest.java new file mode 100644 index 0000000000..7c0e0075e4 --- /dev/null +++ b/seatunnel-connectors-v2/connector-salesforce/src/test/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceRowSerializerTest.java @@ -0,0 +1,251 @@ +/* + * 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.salesforce.sink; + +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.DecimalType; +import org.apache.seatunnel.api.table.type.LocalTimeType; +import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.connectors.seatunnel.salesforce.config.SalesforceSinkConfig; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.math.BigDecimal; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.time.OffsetDateTime; +import java.util.HashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SalesforceRowSerializerTest { + private Map<String, Object> options() { + Map<String, Object> values = new HashMap<>(); + values.put("client_id", "client"); + values.put("client_secret", "secret"); + values.put("username", "user"); + values.put("password", "password"); + values.put("instance_url", "https://example.my.salesforce.com";); + values.put("object_name", "Account"); + values.put("external_id_field", "External_Id__c"); + return values; + } + + private SalesforceSinkConfig config() { + return new SalesforceSinkConfig(ReadonlyConfig.fromMap(options())); + } + + @Test + void serializesNativeValuesAndNullsWithoutStringifyingNumbers() { + String[] names = { + "External_Id__c", + "Flag", + "Count", + "Amount", + "Day", + "Time", + "Timestamp", + "ZonedTimestamp", + "Body", + "Empty" + }; + SeaTunnelDataType<?>[] types = { + BasicType.STRING_TYPE, + BasicType.BOOLEAN_TYPE, + BasicType.LONG_TYPE, + new DecimalType(38, 10), + LocalTimeType.LOCAL_DATE_TYPE, + LocalTimeType.LOCAL_TIME_TYPE, + LocalTimeType.LOCAL_DATE_TIME_TYPE, + LocalTimeType.OFFSET_DATE_TIME_TYPE, + PrimitiveByteArrayType.INSTANCE, + BasicType.STRING_TYPE + }; + SalesforceRowSerializer serializer = + new SalesforceRowSerializer(new SeaTunnelRowType(names, types), config()); + BigDecimal amount = new BigDecimal("12345678901234567890.1234567890"); + ObjectNode result = + serializer.serialize( + new SeaTunnelRow( + new Object[] { + "key", + true, + 9007199254740993L, + amount, + LocalDate.of(2026, 9, 8), + LocalTime.of(12, 3, 4, 123000000), + LocalDateTime.of(2026, 9, 8, 12, 3, 4), + OffsetDateTime.parse("2026-09-08T12:03:04+05:30"), + new byte[] {0, 1, 2}, + null + })); + assertTrue(result.path("Flag").isBoolean()); + assertEquals(9007199254740993L, result.path("Count").longValue()); + assertEquals(amount, result.path("Amount").decimalValue()); + assertEquals("2026-09-08", result.path("Day").asText()); + assertEquals("12:03:04.123Z", result.path("Time").asText()); + assertEquals("2026-09-08T12:03:04Z", result.path("Timestamp").asText()); + assertEquals("2026-09-08T12:03:04+05:30", result.path("ZonedTimestamp").asText()); + assertEquals("AAEC", result.path("Body").asText()); + assertTrue(result.path("Empty").isNull()); + } + + @Test + void rejectsPrecisionLossAndNonFiniteNumbers() { + SalesforceRowSerializer time = serializer(LocalTimeType.LOCAL_TIME_TYPE); + assertThrows( + IllegalArgumentException.class, + () -> + time.serialize( + new SeaTunnelRow(new Object[] {"key", LocalTime.of(12, 0, 0, 1)}))); + SalesforceRowSerializer timestamp = serializer(LocalTimeType.LOCAL_DATE_TIME_TYPE); + assertThrows( + IllegalArgumentException.class, + () -> + timestamp.serialize( + new SeaTunnelRow( + new Object[] { + "key", LocalDateTime.of(2026, 9, 8, 12, 0, 0, 1) + }))); + SalesforceRowSerializer number = serializer(BasicType.DOUBLE_TYPE); + for (double value : + new double[] {Double.NaN, Double.POSITIVE_INFINITY, Double.NEGATIVE_INFINITY}) { + assertThrows( + IllegalArgumentException.class, + () -> number.serialize(new SeaTunnelRow(new Object[] {"key", value}))); + } + } + + @Test + void detachesMutableBinaryFieldsFromReusedInputRows() { + SalesforceRowSerializer serializer = serializer(PrimitiveByteArrayType.INSTANCE); + byte[] bytes = {0, 1, 2}; + ObjectNode result = serializer.serialize(new SeaTunnelRow(new Object[] {"key", bytes})); + bytes[0] = 42; + assertEquals("AAEC", result.path("Value").asText()); + } + + private SalesforceRowSerializer serializer(SeaTunnelDataType<?> type) { + return new SalesforceRowSerializer( + new SeaTunnelRowType( + new String[] {"External_Id__c", "Value"}, + new SeaTunnelDataType[] {BasicType.STRING_TYPE, type}), + config()); + } + + @ParameterizedTest + @ValueSource(strings = {"Id", "attributes", "external_id__c", "bad/field"}) + void rejectsReservedDuplicateAndInvalidSchemaFields(String field) { + assertThrows( + IllegalArgumentException.class, + () -> + new SalesforceRowSerializer( + new SeaTunnelRowType( + new String[] {"External_Id__c", field}, + new SeaTunnelDataType[] { + BasicType.STRING_TYPE, BasicType.STRING_TYPE + }), + config())); + } + + @Test + void requiresExternalIdAndRejectsNestedTypes() { + assertThrows( + IllegalArgumentException.class, + () -> + new SalesforceRowSerializer( + new SeaTunnelRowType( + new String[] {"Name"}, + new SeaTunnelDataType[] {BasicType.STRING_TYPE}), + config())); + assertThrows( + IllegalArgumentException.class, + () -> + serializer( + new SeaTunnelRowType( + new String[] {"nested"}, + new SeaTunnelDataType[] {BasicType.STRING_TYPE}))); + assertThrows( + IllegalArgumentException.class, + () -> + new SalesforceRowSerializer( + new SeaTunnelRowType( + new String[] {"External_Id__c"}, + new SeaTunnelDataType[] {BasicType.BOOLEAN_TYPE}), + config())); + } + + @ParameterizedTest + @ValueSource( + strings = { + "ftp://host";, + "https://user:secret@host";, + "https://host/path";, + "https://host?token=secret";, + "https://host#secret";, + "invalid" + }) + void rejectsUnsafeOrNonOriginUrls(String url) { + Map<String, Object> values = options(); + values.put("instance_url", url); + IllegalArgumentException error = + assertThrows( + IllegalArgumentException.class, + () -> new SalesforceSinkConfig(ReadonlyConfig.fromMap(values))); + assertTrue(!error.getMessage().contains("secret")); + } + + @Test + void validatesBoundsBeforeOpeningConnections() { + Object[][] invalid = { + {"batch_size", 0}, + {"batch_size", 201}, + {"batch_max_bytes", 127}, + {"batch_max_bytes", 8388609}, + {"max_retries", -1}, + {"max_retries", 11}, + {"retry_interval_ms", -1L}, + {"retry_interval_ms", 60001L}, + {"request_timeout_ms", 0}, + {"object_name", "../Account"}, + {"external_id_field", "Id"}, + {"api_version", "v59/0"}, + {"client_secret", " "} + }; + for (Object[] setting : invalid) { + Map<String, Object> values = options(); + values.put((String) setting[0], setting[1]); + assertThrows( + IllegalArgumentException.class, + () -> new SalesforceSinkConfig(ReadonlyConfig.fromMap(values)), + setting[0].toString()); + } + } +} diff --git a/seatunnel-connectors-v2/connector-salesforce/src/test/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceSinkTest.java b/seatunnel-connectors-v2/connector-salesforce/src/test/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceSinkTest.java new file mode 100644 index 0000000000..f957711801 --- /dev/null +++ b/seatunnel-connectors-v2/connector-salesforce/src/test/java/org/apache/seatunnel/connectors/seatunnel/salesforce/sink/SalesforceSinkTest.java @@ -0,0 +1,505 @@ +/* + * 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.salesforce.sink; + +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode; +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ArrayNode; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.configuration.util.ConfigValidator; +import org.apache.seatunnel.api.table.factory.FactoryUtil; +import org.apache.seatunnel.api.table.factory.TableSinkFactory; +import org.apache.seatunnel.api.table.factory.TableSourceFactory; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.RowKind; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.connectors.seatunnel.salesforce.config.SalesforceSinkConfig; +import org.apache.seatunnel.connectors.seatunnel.salesforce.exception.SalesforceConnectorException; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.ValueSource; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiFunction; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SalesforceSinkTest { + private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final SeaTunnelRowType ROW_TYPE = + new SeaTunnelRowType( + new String[] {"External_Id__c", "Name"}, + new SeaTunnelDataType[] {BasicType.STRING_TYPE, BasicType.STRING_TYPE}); + private HttpServer server; + private String url; + private Map<String, Object> options; + private final List<SalesforceSinkWriter> writers = new ArrayList<>(); + private final List<JsonNode> requests = Collections.synchronizedList(new ArrayList<>()); + private final List<Integer> sizes = Collections.synchronizedList(new ArrayList<>()); + private final List<String> authorization = Collections.synchronizedList(new ArrayList<>()); + private final AtomicInteger tokens = new AtomicInteger(); + private volatile BiFunction<Integer, JsonNode, Reply> reply; + private volatile String retryAfter; + + @BeforeEach + void start() throws IOException { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + url = "http://127.0.0.1:"; + server.getAddress().getPort(); + options = new HashMap<>(); + options.put("client_id", "client"); + options.put("client_secret", "private-secret"); + options.put("username", "[email protected]"); + options.put("password", "private-password"); + options.put("security_token", "private-token"); + options.put("instance_url", url); + options.put("object_name", "Account"); + options.put("external_id_field", "External_Id__c"); + options.put("max_retries", 0); + options.put("retry_interval_ms", 0L); + options.put("request_timeout_ms", 1000); + server.createContext( + "/services/oauth2/token", + exchange -> { + int token = tokens.incrementAndGet(); + respond( + exchange, + 200, + "{\"access_token\":\"token-" + + token + + "\",\"instance_url\":\"" + + url + + "\"}"); + }); + reply = (index, body) -> new Reply(200, success(body.path("records").size())); + server.createContext( + "/services/data/v59.0/composite/sobjects/Account/External_Id__c", + exchange -> { + if (!"PATCH".equals(exchange.getRequestMethod())) { + respond(exchange, 405, "[]"); + return; + } + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + byte[] bytes = new byte[4096]; + int read; + while ((read = exchange.getRequestBody().read(bytes)) != -1) { + buffer.write(bytes, 0, read); + } + JsonNode request = MAPPER.readTree(buffer.toByteArray()); + requests.add(request); + sizes.add(buffer.size()); + authorization.add(exchange.getRequestHeaders().getFirst("Authorization")); + Reply result = reply.apply(requests.size(), request); + if (retryAfter != null) { + exchange.getResponseHeaders().add("Retry-After", retryAfter); + } + if (result.status == 307) { + exchange.getResponseHeaders().add("Location", url + "/trap"); + } + respond(exchange, result.status, result.body); + }); + server.start(); + } + + @AfterEach + void stop() throws IOException { + try { + for (SalesforceSinkWriter writer : writers) { + writer.close(); + } + } finally { + server.stop(0); + } + } + + @Test + void sourceAndSinkFactoriesRemainDiscoverable() { + assertNotNull( + FactoryUtil.discoverFactory( + getClass().getClassLoader(), TableSourceFactory.class, "Salesforce")); + assertNotNull( + FactoryUtil.discoverFactory( + getClass().getClassLoader(), TableSinkFactory.class, "Salesforce")); + ConfigValidator.of(ReadonlyConfig.fromMap(options)) + .validate(new SalesforceSinkFactory().optionRule()); + } + + @ParameterizedTest + @ValueSource( + strings = { + "{}", + "{\"access_token\":\"secret\"}", + "invalid-json", + "{\"access_token\":\"secret\",\"instance_url\":\"https://user:secret@host\"}"; + }) + void rejectsMalformedAuthenticationWithoutLeakingCredentials(String body) { + server.removeContext("/services/oauth2/token"); + server.createContext("/services/oauth2/token", exchange -> respond(exchange, 200, body)); + SalesforceConnectorException failure = + assertThrows(SalesforceConnectorException.class, this::writer); + assertFalse(failure.toString().contains("private-secret")); + assertFalse(failure.toString().contains("user:secret")); + assertTrue(requests.isEmpty()); + } + + @Test + void socketTimeoutHasBoundedAttemptsAndDoesNotCompleteCheckpoint() throws Exception { + options.put("request_timeout_ms", 1000); + options.put("max_retries", 1); + CountDownLatch release = new CountDownLatch(1); + reply = + (index, body) -> { + try { + release.await(60, TimeUnit.SECONDS); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + } + return new Reply(200, success(body.path("records").size())); + }; + SalesforceSinkWriter writer = writer(); + writer.write(row("1", "first")); + try { + SalesforceConnectorException failure = + assertThrows(SalesforceConnectorException.class, writer::prepareCommit); + assertTrue(failure.getMessage().contains("after 2 attempt(s)")); + assertThrows(SalesforceConnectorException.class, writer::prepareCommit); + } finally { + release.countDown(); + } + } + + @Test + void flushesByCountAndCheckpointWithDetachedRows() throws IOException { + options.put("batch_size", 2); + SalesforceSinkWriter writer = writer(); + SeaTunnelRow first = row("1", "first"); + writer.write(first); + first.setField(1, "mutated-after-write"); + writer.write(row("2", "second")); + assertEquals(1, requests.size()); + assertTrue(requests.get(0).path("allOrNone").asBoolean()); + assertEquals("first", requests.get(0).path("records").get(0).path("Name").asText()); + assertEquals( + "Account", + requests.get(0).path("records").get(0).path("attributes").path("type").asText()); + writer.write(row("3", null)); + writer.prepareCommit(11); + assertEquals(2, requests.size()); + assertTrue(requests.get(1).path("records").get(0).path("Name").isNull()); + writer.prepareCommit(12); + assertEquals(2, requests.size()); + } + + @Test + void flushesRepeatedKeysInOrder() throws IOException { + SalesforceSinkWriter writer = writer(); + writer.write(row("customer", "before")); + SeaTunnelRow update = row("CUSTOMER", "after"); + update.setRowKind(RowKind.UPDATE_AFTER); + writer.write(update); + writer.prepareCommit(); + assertEquals(2, requests.size()); + assertEquals("before", requests.get(0).path("records").get(0).path("Name").asText()); + assertEquals("after", requests.get(1).path("records").get(0).path("Name").asText()); + } + + @Test + void boundsActualUtf8RequestBytes() throws IOException { + options.put("batch_max_bytes", 180); + SalesforceSinkWriter writer = writer(); + writer.write(row("1", "雪雪雪雪雪")); + writer.write(row("2", "雪雪雪雪雪")); + writer.prepareCommit(); + assertEquals(2, requests.size()); + assertTrue(sizes.stream().allMatch(size -> size <= 180)); + } + + @Test + void rejectsOversizedRecordWithoutSendingOrReplayingOnClose() throws IOException { + options.put("batch_max_bytes", 128); + SalesforceSinkWriter writer = writer(); + assertThrows( + IllegalArgumentException.class, + () -> writer.write(row("1", String.join("", Collections.nCopies(200, "x"))))); + assertThrows(IllegalArgumentException.class, writer::prepareCommit); + writer.close(); + writer.close(); + assertTrue(requests.isEmpty()); + } + + @ParameterizedTest + @EnumSource( + value = RowKind.class, + names = {"DELETE", "UPDATE_BEFORE"}) + void rejectsUnsupportedChangelogKinds(RowKind kind) throws IOException { + SalesforceSinkWriter writer = writer(); + SeaTunnelRow row = row("1", "name"); + row.setRowKind(kind); + assertThrows(IllegalArgumentException.class, () -> writer.write(row)); + assertTrue(requests.isEmpty()); + } + + @Test + void rejectsMissingExternalIdBeforeSending() throws IOException { + SalesforceSinkWriter writer = writer(); + assertThrows(IllegalArgumentException.class, () -> writer.write(row(null, "name"))); + assertTrue(requests.isEmpty()); + } + + @Test + void partialResultFailsCheckpointWithoutRetryAndWithoutEchoingData() throws IOException { + options.put("max_retries", 3); + reply = + (index, body) -> + new Reply( + 200, + "[{\"id\":\"001\",\"success\":true,\"errors\":[]}," + + "{\"id\":null,\"success\":false,\"errors\":[{\"statusCode\":\"INVALID_FIELD\",\"message\":\"private-secret row value\"}]}]"); + SalesforceSinkWriter writer = writer(); + writer.write(row("1", "one")); + writer.write(row("2", "two")); + SalesforceConnectorException failure = + assertThrows(SalesforceConnectorException.class, writer::prepareCommit); + assertTrue(failure.getMessage().contains("index 1")); + assertTrue(failure.getMessage().contains("INVALID_FIELD")); + assertFalse(failure.toString().contains("private-secret")); + assertThrows(SalesforceConnectorException.class, writer::prepareCommit); + writer.close(); + assertEquals(1, requests.size()); + } + + @ParameterizedTest + @ValueSource( + strings = { + "not-json", + "[]", + "{}", + "[{\"success\":true,\"errors\":[]}]", + "[{\"id\":\"001\",\"success\":true}]", + "[{\"success\":\"true\",\"errors\":[]}]" + }) + void malformedSuccessResponseDoesNotAcknowledgeOrRetry(String body) throws IOException { + reply = (index, request) -> new Reply(200, body); + options.put("max_retries", 3); + SalesforceSinkWriter writer = writer(); + writer.write(row("1", "name")); + assertThrows(SalesforceConnectorException.class, writer::prepareCommit); + assertEquals(1, requests.size()); + } + + @ParameterizedTest + @ValueSource(ints = {429, 500, 502, 503, 504}) + void retriesTransientHttpStatusWithIdenticalPayload(int status) throws IOException { + options.put("max_retries", 1); + retryAfter = "0"; + reply = + (index, body) -> + index == 1 + ? new Reply(status, "[]") + : new Reply(200, success(body.path("records").size())); + SalesforceSinkWriter writer = writer(); + writer.write(row("1", "name")); + writer.prepareCommit(); + assertEquals(2, requests.size()); + assertEquals(requests.get(0), requests.get(1)); + } + + @Test + void retryExhaustionIsBounded() throws IOException { + options.put("max_retries", 2); + reply = (index, body) -> new Reply(503, "[]"); + SalesforceSinkWriter writer = writer(); + writer.write(row("1", "name")); + assertThrows(SalesforceConnectorException.class, writer::prepareCommit); + assertEquals(3, requests.size()); + } + + @Test + void excessiveRetryAfterFailsWithoutHammeringEndpoint() throws IOException { + options.put("max_retries", 3); + retryAfter = "3600"; + reply = (index, body) -> new Reply(429, "[]"); + SalesforceSinkWriter writer = writer(); + writer.write(row("1", "name")); + assertThrows(SalesforceConnectorException.class, writer::prepareCommit); + assertEquals(1, requests.size()); + } + + @Test + void expiredSessionRefreshesOnce() throws IOException { + options.put("max_retries", 3); + reply = (index, body) -> index == 1 ? new Reply(401, "[]") : new Reply(200, success(1)); + SalesforceSinkWriter writer = writer(); + writer.write(row("1", "name")); + writer.prepareCommit(); + assertEquals(2, tokens.get()); + assertEquals("Bearer token-1", authorization.get(0)); + assertEquals("Bearer token-2", authorization.get(1)); + } + + @Test + void repeatedUnauthorizedResponseDoesNotRefreshForever() throws IOException { + options.put("max_retries", 3); + reply = (index, body) -> new Reply(401, "[]"); + SalesforceSinkWriter writer = writer(); + writer.write(row("1", "name")); + assertThrows(SalesforceConnectorException.class, writer::prepareCommit); + assertEquals(2, tokens.get()); + assertEquals(2, requests.size()); + } + + @ParameterizedTest + @ValueSource(ints = {400, 403, 404, 307}) + void permanentErrorsAndRedirectsAreNotRetried(int status) throws IOException { + AtomicInteger trap = new AtomicInteger(); + server.createContext( + "/trap", + exchange -> { + trap.incrementAndGet(); + respond(exchange, 200, success(1)); + }); + options.put("max_retries", 3); + reply = (index, body) -> new Reply(status, "private-secret"); + SalesforceSinkWriter writer = writer(); + writer.write(row("1", "name")); + SalesforceConnectorException failure = + assertThrows(SalesforceConnectorException.class, writer::prepareCommit); + assertFalse(failure.toString().contains("private-secret")); + assertEquals(1, requests.size()); + assertEquals(0, trap.get()); + } + + @Test + void interruptionStopsBackoffAndPreservesInterrupt() throws Exception { + options.put("max_retries", 3); + options.put("retry_interval_ms", 60000L); + CountDownLatch requested = new CountDownLatch(1); + reply = + (index, body) -> { + requested.countDown(); + return new Reply(503, "[]"); + }; + SalesforceSinkWriter writer = writer(); + writer.write(row("1", "name")); + AtomicReference<Throwable> error = new AtomicReference<>(); + AtomicReference<Boolean> interrupted = new AtomicReference<>(false); + Thread thread = + new Thread( + () -> { + try { + writer.prepareCommit(); + } catch (Throwable e) { + error.set(e); + interrupted.set(Thread.currentThread().isInterrupted()); + } + }); + thread.start(); + try { + assertTrue(requested.await(5, TimeUnit.SECONDS)); + thread.interrupt(); + thread.join(5000); + assertFalse(thread.isAlive()); + assertInstanceOf(SalesforceConnectorException.class, error.get()); + assertEquals(true, interrupted.get()); + assertEquals(1, requests.size()); + } finally { + thread.interrupt(); + thread.join(5000); + } + } + + @Test + void closesWithFinalFlushAndIsIdempotent() throws IOException { + SalesforceSinkWriter writer = writer(); + writer.write(row("1", "name")); + writer.close(); + writer.close(); + assertEquals(1, requests.size()); + assertThrows(IllegalStateException.class, () -> writer.write(row("2", "name"))); + } + + private SalesforceSinkWriter writer() throws IOException { + SalesforceSinkWriter writer = + new SalesforceSinkWriter( + ROW_TYPE, new SalesforceSinkConfig(ReadonlyConfig.fromMap(options))); + writers.add(writer); + return writer; + } + + private static SeaTunnelRow row(String key, String name) { + return new SeaTunnelRow(new Object[] {key, name}); + } + + private static String success(int count) { + ArrayNode results = MAPPER.createArrayNode(); + for (int i = 0; i < count; i++) { + results.addObject() + .put("id", "001" + i) + .put("success", true) + .put("created", false) + .putArray("errors"); + } + return results.toString(); + } + + private static void respond(HttpExchange exchange, int status, String body) throws IOException { + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(status, bytes.length); + try (OutputStream out = exchange.getResponseBody()) { + out.write(bytes); + } + } + + private static final class Reply { + final int status; + final String body; + + Reply(int status, String body) { + this.status = status; + this.body = body; + } + } +} diff --git a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigShadeUtils.java b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigShadeUtils.java index 706ffc8237..ae73de60f2 100644 --- a/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigShadeUtils.java +++ b/seatunnel-core/seatunnel-core-starter/src/main/java/org/apache/seatunnel/core/starter/utils/ConfigShadeUtils.java @@ -55,7 +55,7 @@ public final class ConfigShadeUtils { new String[] {"password", "username", "auth", "token", "access_key", "secret_key"}; private static final String[] DEFAULT_LOG_MASK_ONLY_KEYWORDS = - new String[] {"sasl.jaas.config", "community"}; + new String[] {"sasl.jaas.config", "community", "client_secret", "security_token"}; private static final Map<String, ConfigShade> CONFIG_SHADES = new HashMap<>(); diff --git a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigBuilderTest.java b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigBuilderTest.java index fe1b71f7a9..415ff58562 100644 --- a/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigBuilderTest.java +++ b/seatunnel-core/seatunnel-core-starter/src/test/java/org/apache/seatunnel/core/starter/utils/ConfigBuilderTest.java @@ -120,6 +120,30 @@ public class ConfigBuilderTest { Assertions.assertEquals("127.0.0.1", desensitizedSink.get("host")); } + @Test + public void testConfigDesensitizationMasksSalesforceSecretsWithoutAddingEncryptionKeys() { + Map<String, Object> connector = new LinkedHashMap<>(); + connector.put("client_secret", "private-secret"); + connector.put("security_token", "private-token"); + connector.put("object_name", "Account"); + Map<String, Object> config = new LinkedHashMap<>(); + config.put("source", Arrays.asList(connector)); + config.put("sink", Arrays.asList(connector)); + Map<String, Object> masked = + ConfigBuilder.configDesensitization( + config, ConfigShadeUtils.getLogDesensitizationOptions(null)); + for (String role : Arrays.asList("source", "sink")) { + Map<?, ?> entry = (Map<?, ?>) ((List<?>) masked.get(role)).get(0); + Assertions.assertEquals("******", entry.get("client_secret")); + Assertions.assertEquals("******", entry.get("security_token")); + Assertions.assertEquals("Account", entry.get("object_name")); + } + Assertions.assertFalse( + ConfigShadeUtils.getSensitiveOptions(null).contains("client_secret")); + Assertions.assertFalse( + ConfigShadeUtils.getSensitiveOptions(null).contains("security_token")); + } + @Test public void testConfigDesensitizationMasksS3CredentialOptions() { Map<String, Object> accessKeyConfig = new LinkedHashMap<>(); diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/pom.xml b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/pom.xml index 4c5f122512..702f6de9db 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/pom.xml +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/pom.xml @@ -26,6 +26,12 @@ <name>SeaTunnel : E2E : Connector V2 : Http</name> <dependencies> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-salesforce</artifactId> + <version>${project.version}</version> + <scope>test</scope> + </dependency> <dependency> <groupId>org.apache.seatunnel</groupId> <artifactId>connector-assert</artifactId> diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/java/org/apache/seatunnel/e2e/connector/http/SalesforceSinkIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/java/org/apache/seatunnel/e2e/connector/http/SalesforceSinkIT.java new file mode 100644 index 0000000000..24349090a2 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/java/org/apache/seatunnel/e2e/connector/http/SalesforceSinkIT.java @@ -0,0 +1,132 @@ +/* + * 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.http; + +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode; +import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.ObjectMapper; + +import org.apache.seatunnel.e2e.common.TestResource; +import org.apache.seatunnel.e2e.common.TestSuiteBase; +import org.apache.seatunnel.e2e.common.container.TestContainer; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.TestTemplate; +import org.mockserver.client.MockServerClient; +import org.mockserver.model.HttpRequest; +import org.testcontainers.containers.Container; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.utility.DockerImageName; + +import static org.mockserver.model.HttpRequest.request; +import static org.mockserver.model.HttpResponse.response; + +class SalesforceSinkIT extends TestSuiteBase implements TestResource { + private static final String PATH = + "/services/data/v59.0/composite/sobjects/Account/External_Id__c"; + private GenericContainer<?> server; + private MockServerClient mock; + + @BeforeAll + @Override + public void startUp() { + server = + new GenericContainer<>(DockerImageName.parse("mockserver/mockserver:5.14.0")) + .withNetwork(NETWORK) + .withNetworkAliases("salesforce-mock") + .withExposedPorts(1080) + .waitingFor(Wait.forHttp("/").forStatusCode(404)); + server.start(); + mock = new MockServerClient(server.getHost(), server.getMappedPort(1080)); + } + + @AfterAll + @Override + public void tearDown() { + if (mock != null) { + mock.close(); + } + if (server != null) { + server.close(); + } + } + + private void responses(String result) { + mock.reset(); + mock.when(request().withMethod("POST").withPath("/services/oauth2/token")) + .respond( + response() + .withStatusCode(200) + .withHeader("Content-Type", "application/json") + .withBody( + "{\"access_token\":\"test-token\",\"instance_url\":\"http://salesforce-mock:1080\"}";)); + mock.when( + request() + .withMethod("PATCH") + .withPath(PATH) + .withHeader("Authorization", "Bearer test-token")) + .respond( + response() + .withStatusCode(200) + .withHeader("Content-Type", "application/json") + .withBody(result)); + } + + @TestTemplate + void upsertsOneObjectAndFlushesAtEndOfInput(TestContainer container) throws Exception { + responses( + "[{\"id\":\"001A\",\"success\":true,\"created\":true,\"errors\":[]}," + + "{\"id\":\"001B\",\"success\":true,\"created\":true,\"errors\":[]}]"); + Container.ExecResult result = container.executeJob("/fake_to_salesforce.conf"); + Assertions.assertEquals(0, result.getExitCode(), result.getStderr()); + HttpRequest[] requests = + mock.retrieveRecordedRequests(request().withMethod("PATCH").withPath(PATH)); + Assertions.assertTrue(requests.length > 0); + ObjectMapper mapper = new ObjectMapper(); + for (HttpRequest sent : requests) { + JsonNode body = mapper.readTree(sent.getBodyAsString()); + Assertions.assertTrue(body.path("allOrNone").asBoolean()); + Assertions.assertEquals(2, body.path("records").size()); + Assertions.assertEquals( + "first", body.path("records").get(0).path("External_Id__c").asText()); + Assertions.assertEquals( + "second", body.path("records").get(1).path("External_Id__c").asText()); + Assertions.assertEquals( + "Account", + body.path("records").get(0).path("attributes").path("type").asText()); + } + } + + @TestTemplate + void rejectsRecordErrorsInsteadOfCompletingTheJob(TestContainer container) throws Exception { + responses( + "[{\"id\":null,\"success\":false,\"errors\":[{\"statusCode\":\"REQUIRED_FIELD_MISSING\",\"message\":\"mock validation failure\"}]}," + + "{\"id\":null,\"success\":false,\"errors\":[{\"statusCode\":\"ALL_OR_NONE_OPERATION_ROLLED_BACK\",\"message\":\"rollback\"}]}]"); + Container.ExecResult result = container.executeJob("/fake_to_salesforce.conf"); + Assertions.assertNotEquals(0, result.getExitCode(), "A rejected upsert must fail the job"); + String logs = result.getStdout() + result.getStderr() + container.getServerLogs(); + Assertions.assertTrue( + logs.contains("REQUIRED_FIELD_MISSING"), + "The failure must expose the Salesforce record status code"); + Assertions.assertTrue( + mock.retrieveRecordedRequests(request().withMethod("PATCH").withPath(PATH)).length + > 0); + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/resources/fake_to_salesforce.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/resources/fake_to_salesforce.conf new file mode 100644 index 0000000000..7c895c0778 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-http-e2e/src/test/resources/fake_to_salesforce.conf @@ -0,0 +1,47 @@ +# 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 { + FakeSource { + schema { + fields { + External_Id__c = string + Name = string + } + } + rows = [ + { kind = INSERT, fields = ["first", "Alice"] }, + { kind = INSERT, fields = ["second", "Bob"] } + ] + } +} +sink { + Salesforce { + instance_url = "http://salesforce-mock:1080"; + client_id = "test-client" + client_secret = "test-secret" + username = "[email protected]" + password = "test-password" + object_name = "Account" + external_id_field = "External_Id__c" + batch_size = 200 + max_retries = 0 + } +}
