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-12233-3339132550594fe3128534cf37c86f0cbbe435cc in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 7b88e187059226a64ff09dea3f4fc74de402e901 Author: Yohanes <[email protected]> AuthorDate: Sun Sep 13 13:53:01 2026 +0000 feat(connector): add Mem0 Platform V3 ADD sink (#12233) Signed-off-by: Yohanes <[email protected]> Co-authored-by: Yohanes <[email protected]> --- config/plugin_config | 1 + docs/en/connectors/changelog/connector-mem0.md | 7 + docs/en/connectors/sink/Mem0.md | 33 ++++ docs/zh/connectors/changelog/connector-mem0.md | 7 + docs/zh/connectors/sink/Mem0.md | 30 ++++ plugin-mapping.properties | 1 + .../connector-http/connector-mem0/pom.xml | 35 ++++ .../seatunnel/mem0/config/Mem0SinkOptions.java | 68 ++++++++ .../connectors/seatunnel/mem0/sink/Mem0Sink.java | 94 ++++++++++ .../seatunnel/mem0/sink/Mem0SinkFactory.java | 56 ++++++ .../seatunnel/mem0/sink/Mem0SinkWriter.java | 193 +++++++++++++++++++++ .../seatunnel/mem0/sink/Mem0SinkWriterTest.java | 166 ++++++++++++++++++ seatunnel-connectors-v2/connector-http/pom.xml | 1 + seatunnel-dist/pom.xml | 6 + 14 files changed, 698 insertions(+) diff --git a/config/plugin_config b/config/plugin_config index f8d2f01e99..8e2b4992d2 100644 --- a/config/plugin_config +++ b/config/plugin_config @@ -66,6 +66,7 @@ connector-tiktok-ads connector-graphql connector-hive connector-http-base +connector-mem0 connector-http-feishu connector-http-gitlab connector-http-github diff --git a/docs/en/connectors/changelog/connector-mem0.md b/docs/en/connectors/changelog/connector-mem0.md new file mode 100644 index 0000000000..e931ca8a68 --- /dev/null +++ b/docs/en/connectors/changelog/connector-mem0.md @@ -0,0 +1,7 @@ +<details><summary> Change Log </summary> + +| Change | Commit | Version | +| --- | --- | --- | +| [Feature][Connector-V2] Add Mem0 Platform V3 ADD sink connector | - | Next | + +</details> diff --git a/docs/en/connectors/sink/Mem0.md b/docs/en/connectors/sink/Mem0.md new file mode 100644 index 0000000000..893c59281e --- /dev/null +++ b/docs/en/connectors/sink/Mem0.md @@ -0,0 +1,33 @@ +import ChangeLog from '../changelog/connector-mem0.md'; + +# Mem0 + +The Mem0 sink sends SeaTunnel rows to the hosted Mem0 Platform V3 asynchronous +add API. Phase 1 supports `ADD` only. A row is acknowledged only when the API +returns a non-empty `event_id`; this is accepted at-least-once delivery, not +completion of asynchronous processing. + +## Configuration + +```hocon +sink { + Mem0 { + api_key = "${MEM0_API_KEY}" + messages_field = "messages" + user_id_field = "user_id" + api_base_url = "https://api.mem0.ai"; + } +} +``` + +`messages_field` must resolve to an array. At least one scope field among +`user_id_field`, `agent_id_field`, `app_id_field`, and `run_id_field` must be +configured and non-empty for each row. Optional `metadata_field` must resolve +to a JSON object. + +The connector uses `POST /v3/memories/add/` with `Authorization: Token`, +`Content-Type: application/json`, and `Accept: application/json`. Delete, +self-hosted OSS endpoints, event polling, and a generic Mem0-compatible +protocol are outside Phase 1. + +<ChangeLog /> diff --git a/docs/zh/connectors/changelog/connector-mem0.md b/docs/zh/connectors/changelog/connector-mem0.md new file mode 100644 index 0000000000..e931ca8a68 --- /dev/null +++ b/docs/zh/connectors/changelog/connector-mem0.md @@ -0,0 +1,7 @@ +<details><summary> Change Log </summary> + +| Change | Commit | Version | +| --- | --- | --- | +| [Feature][Connector-V2] Add Mem0 Platform V3 ADD sink connector | - | Next | + +</details> diff --git a/docs/zh/connectors/sink/Mem0.md b/docs/zh/connectors/sink/Mem0.md new file mode 100644 index 0000000000..5709a2ffa1 --- /dev/null +++ b/docs/zh/connectors/sink/Mem0.md @@ -0,0 +1,30 @@ +import ChangeLog from '../changelog/connector-mem0.md'; + +# Mem0 + +Mem0 Sink 将 SeaTunnel 行写入托管版 Mem0 Platform V3 异步添加接口。第一阶段 +只支持 `ADD`。只有接口返回非空 `event_id` 时才确认该行；这表示“已接受的至少一次 +投递”，不代表 Mem0 异步处理已经完成。 + +## 配置 + +```hocon +sink { + Mem0 { + api_key = "${MEM0_API_KEY}" + messages_field = "messages" + user_id_field = "user_id" + api_base_url = "https://api.mem0.ai"; + } +} +``` + +`messages_field` 必须解析为数组。每行至少要有 +`user_id_field`、`agent_id_field`、`app_id_field`、`run_id_field` 中的一个非空范围字段。 +可选的 `metadata_field` 必须解析为 JSON 对象。 + +连接器使用 `POST /v3/memories/add/`，并发送 `Authorization: Token`、 +`Content-Type: application/json` 和 `Accept: application/json`。删除、自托管 OSS +接口、事件轮询以及通用 Mem0 兼容协议不属于第一阶段范围。 + +<ChangeLog /> diff --git a/plugin-mapping.properties b/plugin-mapping.properties index e49dc101c9..538a91f225 100644 --- a/plugin-mapping.properties +++ b/plugin-mapping.properties @@ -171,6 +171,7 @@ seatunnel.source.MQTT = connector-mqtt seatunnel.sink.MQTT = connector-mqtt seatunnel.source.SNMP = connector-snmp seatunnel.sink.SNMP = connector-snmp +seatunnel.sink.Mem0 = connector-mem0 seatunnel.source.Python = connector-python seatunnel.source.Prometheus = connector-prometheus seatunnel.sink.Prometheus = connector-prometheus diff --git a/seatunnel-connectors-v2/connector-http/connector-mem0/pom.xml b/seatunnel-connectors-v2/connector-http/connector-mem0/pom.xml new file mode 100644 index 0000000000..3b9c198209 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-mem0/pom.xml @@ -0,0 +1,35 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http</artifactId> + <version>${revision}</version> + </parent> + <artifactId>connector-mem0</artifactId> + <name>SeaTunnel : Connectors V2 : Http : Mem0</name> + <dependencies> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-http-base</artifactId> + <version>${project.version}</version> + </dependency> + </dependencies> +</project> diff --git a/seatunnel-connectors-v2/connector-http/connector-mem0/src/main/java/org/apache/seatunnel/connectors/seatunnel/mem0/config/Mem0SinkOptions.java b/seatunnel-connectors-v2/connector-http/connector-mem0/src/main/java/org/apache/seatunnel/connectors/seatunnel/mem0/config/Mem0SinkOptions.java new file mode 100644 index 0000000000..ca6d03d0d6 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-mem0/src/main/java/org/apache/seatunnel/connectors/seatunnel/mem0/config/Mem0SinkOptions.java @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.seatunnel.connectors.seatunnel.mem0.config; + +import org.apache.seatunnel.api.configuration.Option; +import org.apache.seatunnel.api.configuration.Options; +import org.apache.seatunnel.connectors.seatunnel.http.config.HttpCommonOptions; + +public class Mem0SinkOptions extends HttpCommonOptions { + public static final String DEFAULT_API_BASE_URL = "https://api.mem0.ai";; + public static final String ADD_PATH = "/v3/memories/add/"; + public static final String AUTHORIZATION = "Authorization"; + public static final String CONTENT_TYPE = "Content-Type"; + public static final String ACCEPT = "Accept"; + public static final String APPLICATION_JSON = "application/json"; + + public static final Option<String> API_BASE_URL = + Options.key("api_base_url") + .stringType() + .defaultValue(DEFAULT_API_BASE_URL) + .withDescription("Mem0 Platform V3 API base URL"); + public static final Option<String> API_KEY = + Options.key("api_key").stringType().noDefaultValue().withDescription("Mem0 API key"); + public static final Option<String> MESSAGES_FIELD = + Options.key("messages_field") + .stringType() + .noDefaultValue() + .withDescription("Input row field containing the messages payload"); + public static final Option<String> USER_ID_FIELD = + Options.key("user_id_field") + .stringType() + .noDefaultValue() + .withDescription("Input row field mapped to Mem0 user_id"); + public static final Option<String> AGENT_ID_FIELD = + Options.key("agent_id_field") + .stringType() + .noDefaultValue() + .withDescription("Optional input row field mapped to Mem0 agent_id"); + public static final Option<String> APP_ID_FIELD = + Options.key("app_id_field") + .stringType() + .noDefaultValue() + .withDescription("Optional input row field mapped to Mem0 app_id"); + public static final Option<String> RUN_ID_FIELD = + Options.key("run_id_field") + .stringType() + .noDefaultValue() + .withDescription("Optional input row field mapped to Mem0 run_id"); + public static final Option<String> METADATA_FIELD = + Options.key("metadata_field") + .stringType() + .noDefaultValue() + .withDescription("Optional input row field mapped to Mem0 metadata"); +} diff --git a/seatunnel-connectors-v2/connector-http/connector-mem0/src/main/java/org/apache/seatunnel/connectors/seatunnel/mem0/sink/Mem0Sink.java b/seatunnel-connectors-v2/connector-http/connector-mem0/src/main/java/org/apache/seatunnel/connectors/seatunnel/mem0/sink/Mem0Sink.java new file mode 100644 index 0000000000..ed5957d96c --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-mem0/src/main/java/org/apache/seatunnel/connectors/seatunnel/mem0/sink/Mem0Sink.java @@ -0,0 +1,94 @@ +/* + * 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.mem0.sink; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +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.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSimpleSink; +import org.apache.seatunnel.connectors.seatunnel.http.config.HttpParameter; +import org.apache.seatunnel.connectors.seatunnel.mem0.config.Mem0SinkOptions; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; + +public class Mem0Sink extends AbstractSimpleSink<SeaTunnelRow, Void> { + private final CatalogTable catalogTable; + private final SeaTunnelRowType rowType; + private final HttpParameter httpParameter; + private final String messagesField; + private final String userIdField; + private final String agentIdField; + private final String appIdField; + private final String runIdField; + private final String metadataField; + + public Mem0Sink(ReadonlyConfig config, CatalogTable catalogTable) { + this.catalogTable = catalogTable; + this.rowType = catalogTable.getSeaTunnelRowType(); + this.messagesField = config.get(Mem0SinkOptions.MESSAGES_FIELD); + this.userIdField = config.get(Mem0SinkOptions.USER_ID_FIELD); + this.agentIdField = config.getOptional(Mem0SinkOptions.AGENT_ID_FIELD).orElse(null); + this.appIdField = config.getOptional(Mem0SinkOptions.APP_ID_FIELD).orElse(null); + this.runIdField = config.getOptional(Mem0SinkOptions.RUN_ID_FIELD).orElse(null); + this.metadataField = config.getOptional(Mem0SinkOptions.METADATA_FIELD).orElse(null); + String baseUrl = config.get(Mem0SinkOptions.API_BASE_URL); + this.httpParameter = new HttpParameter(); + this.httpParameter.setUrl(normalizeBaseUrl(baseUrl) + Mem0SinkOptions.ADD_PATH); + Map<String, String> headers = new HashMap<>(); + headers.put(Mem0SinkOptions.AUTHORIZATION, "Token " + config.get(Mem0SinkOptions.API_KEY)); + headers.put(Mem0SinkOptions.CONTENT_TYPE, Mem0SinkOptions.APPLICATION_JSON); + headers.put(Mem0SinkOptions.ACCEPT, Mem0SinkOptions.APPLICATION_JSON); + this.httpParameter.setHeaders(headers); + this.httpParameter.setRetry(config.getOptional(Mem0SinkOptions.RETRY).orElse(0)); + this.httpParameter.setRetryBackoffMultiplierMillis( + config.get(Mem0SinkOptions.RETRY_BACKOFF_MULTIPLIER_MS)); + this.httpParameter.setRetryBackoffMaxMillis( + config.get(Mem0SinkOptions.RETRY_BACKOFF_MAX_MS)); + } + + private static String normalizeBaseUrl(String value) { + return value.endsWith("/") ? value.substring(0, value.length() - 1) : value; + } + + @Override + public String getPluginName() { + return "Mem0"; + } + + @Override + public Mem0SinkWriter createWriter(SinkWriter.Context context) throws IOException { + return new Mem0SinkWriter( + rowType, + httpParameter, + messagesField, + userIdField, + agentIdField, + appIdField, + runIdField, + metadataField); + } + + @Override + public Optional<CatalogTable> getWriteCatalogTable() { + return Optional.ofNullable(catalogTable); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-mem0/src/main/java/org/apache/seatunnel/connectors/seatunnel/mem0/sink/Mem0SinkFactory.java b/seatunnel-connectors-v2/connector-http/connector-mem0/src/main/java/org/apache/seatunnel/connectors/seatunnel/mem0/sink/Mem0SinkFactory.java new file mode 100644 index 0000000000..af6bb0868f --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-mem0/src/main/java/org/apache/seatunnel/connectors/seatunnel/mem0/sink/Mem0SinkFactory.java @@ -0,0 +1,56 @@ +/* + * 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.mem0.sink; + +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.mem0.config.Mem0SinkOptions; + +import com.google.auto.service.AutoService; + +@AutoService(Factory.class) +public class Mem0SinkFactory implements TableSinkFactory { + @Override + public String factoryIdentifier() { + return "Mem0"; + } + + @Override + public OptionRule optionRule() { + return OptionRule.builder() + .required(Mem0SinkOptions.API_KEY, Mem0SinkOptions.MESSAGES_FIELD) + .optional( + Mem0SinkOptions.API_BASE_URL, + Mem0SinkOptions.USER_ID_FIELD, + Mem0SinkOptions.AGENT_ID_FIELD, + Mem0SinkOptions.APP_ID_FIELD, + Mem0SinkOptions.RUN_ID_FIELD, + Mem0SinkOptions.METADATA_FIELD, + Mem0SinkOptions.RETRY, + Mem0SinkOptions.RETRY_BACKOFF_MULTIPLIER_MS, + Mem0SinkOptions.RETRY_BACKOFF_MAX_MS) + .build(); + } + + @Override + public TableSink createSink(TableSinkFactoryContext context) { + return () -> new Mem0Sink(context.getOptions(), context.getCatalogTable()); + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-mem0/src/main/java/org/apache/seatunnel/connectors/seatunnel/mem0/sink/Mem0SinkWriter.java b/seatunnel-connectors-v2/connector-http/connector-mem0/src/main/java/org/apache/seatunnel/connectors/seatunnel/mem0/sink/Mem0SinkWriter.java new file mode 100644 index 0000000000..5692934889 --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-mem0/src/main/java/org/apache/seatunnel/connectors/seatunnel/mem0/sink/Mem0SinkWriter.java @@ -0,0 +1,193 @@ +/* + * 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.mem0.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.ObjectNode; + +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.http.client.HttpClientProvider; +import org.apache.seatunnel.connectors.seatunnel.http.client.HttpResponse; +import org.apache.seatunnel.connectors.seatunnel.http.config.HttpParameter; +import org.apache.seatunnel.format.json.JsonSerializationSchema; + +import java.io.IOException; +import java.util.Objects; +import java.util.concurrent.TimeUnit; + +/** Writes SeaTunnel rows to the hosted Mem0 Platform V3 asynchronous add API. */ +public class Mem0SinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void> { + private final HttpClientProvider httpClient; + private final String url; + private final JsonSerializationSchema serializationSchema; + private final ObjectMapper objectMapper; + private final String messagesField; + private final String userIdField; + private final String agentIdField; + private final String appIdField; + private final String runIdField; + private final String metadataField; + private final int maxRetries; + private final int retryBackoffMultiplierMillis; + private final int retryBackoffMaxMillis; + + public Mem0SinkWriter( + SeaTunnelRowType rowType, + HttpParameter parameter, + String messagesField, + String userIdField, + String agentIdField, + String appIdField, + String runIdField, + String metadataField) { + this.httpClient = new HttpClientProvider(parameter); + this.url = parameter.getUrl(); + this.serializationSchema = new JsonSerializationSchema(rowType); + this.objectMapper = serializationSchema.getMapper(); + this.messagesField = messagesField; + this.userIdField = userIdField; + this.agentIdField = agentIdField; + this.appIdField = appIdField; + this.runIdField = runIdField; + this.metadataField = metadataField; + this.headers = parameter.getHeaders(); + this.maxRetries = Math.max(0, parameter.getRetry()); + this.retryBackoffMultiplierMillis = + Math.max(0, parameter.getRetryBackoffMultiplierMillis()); + this.retryBackoffMaxMillis = Math.max(0, parameter.getRetryBackoffMaxMillis()); + } + + private final java.util.Map<String, String> headers; + + @Override + public void write(SeaTunnelRow element) throws IOException { + ObjectNode row = parseRow(element); + ObjectNode request = objectMapper.createObjectNode(); + JsonNode messages = required(row, messagesField, "messages"); + if (!messages.isArray()) { + throw new IOException("Mem0 messages field must contain a JSON array"); + } + request.set("messages", messages); + boolean hasScope = false; + hasScope |= copyIfPresent(row, userIdField, "user_id", request); + hasScope |= copyIfPresent(row, agentIdField, "agent_id", request); + hasScope |= copyIfPresent(row, appIdField, "app_id", request); + hasScope |= copyIfPresent(row, runIdField, "run_id", request); + if (!hasScope) { + throw new IOException("Mem0 requires at least one scope field"); + } + if (metadataField != null) { + JsonNode metadata = row.get(metadataField); + if (metadata != null && !metadata.isNull()) { + if (!metadata.isObject()) { + throw new IOException("Mem0 metadata field must contain a JSON object"); + } + request.set("metadata", metadata); + } + } + send(request); + } + + private ObjectNode parseRow(SeaTunnelRow element) throws IOException { + return (ObjectNode) objectMapper.readTree(serializationSchema.serialize(element)); + } + + private JsonNode required(ObjectNode row, String field, String logicalName) throws IOException { + JsonNode value = row.get(field); + if (value == null + || value.isNull() + || (value.isTextual() && value.textValue().trim().isEmpty())) { + throw new IOException("Mem0 " + logicalName + " field is missing or empty"); + } + return value; + } + + private boolean copyIfPresent(ObjectNode row, String field, String name, ObjectNode target) { + if (field == null) { + return false; + } + JsonNode value = row.get(field); + if (value == null + || value.isNull() + || (value.isTextual() && value.textValue().trim().isEmpty())) { + return false; + } + target.set(name, value); + return true; + } + + private void send(ObjectNode request) throws IOException { + final String body; + try { + body = objectMapper.writeValueAsString(request); + for (int attempt = 0; ; attempt++) { + HttpResponse response = httpClient.doPost(url, headers, body); + if (response.getCode() >= 200 && response.getCode() < 300) { + validateAcceptedResponse(response); + return; + } + if (!isRetryableStatus(response.getCode()) || attempt >= maxRetries) { + throw new IOException( + "Mem0 add request failed with HTTP status " + response.getCode()); + } + waitBeforeRetry(attempt); + } + } catch (IOException e) { + throw e; + } catch (Exception e) { + throw new IOException("Failed to send Mem0 add request", e); + } + } + + private void validateAcceptedResponse(HttpResponse response) throws IOException { + JsonNode responseBody = objectMapper.readTree(response.getContent()); + JsonNode eventId = responseBody == null ? null : responseBody.get("event_id"); + if (eventId == null || eventId.isNull() || eventId.asText().trim().isEmpty()) { + throw new IOException("Mem0 add response did not contain a nonblank event_id"); + } + } + + static boolean isRetryableStatus(int statusCode) { + return statusCode == 408 || statusCode == 429 || statusCode >= 500; + } + + private void waitBeforeRetry(int attempt) throws IOException { + long delay = + Math.min( + (long) retryBackoffMultiplierMillis * (attempt + 1), + (long) retryBackoffMaxMillis); + if (delay <= 0) { + return; + } + try { + TimeUnit.MILLISECONDS.sleep(delay); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while retrying Mem0 add request", e); + } + } + + @Override + public void close() throws IOException { + if (Objects.nonNull(httpClient)) { + httpClient.close(); + } + } +} diff --git a/seatunnel-connectors-v2/connector-http/connector-mem0/src/test/java/org/apache/seatunnel/connectors/seatunnel/mem0/sink/Mem0SinkWriterTest.java b/seatunnel-connectors-v2/connector-http/connector-mem0/src/test/java/org/apache/seatunnel/connectors/seatunnel/mem0/sink/Mem0SinkWriterTest.java new file mode 100644 index 0000000000..75185d4edd --- /dev/null +++ b/seatunnel-connectors-v2/connector-http/connector-mem0/src/test/java/org/apache/seatunnel/connectors/seatunnel/mem0/sink/Mem0SinkWriterTest.java @@ -0,0 +1,166 @@ +/* + * 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.mem0.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.api.table.type.ArrayType; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; +import org.apache.seatunnel.connectors.seatunnel.http.client.HttpClientProvider; +import org.apache.seatunnel.connectors.seatunnel.http.client.HttpResponse; +import org.apache.seatunnel.connectors.seatunnel.http.config.HttpParameter; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +import java.lang.reflect.Field; +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.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class Mem0SinkWriterTest { + @Mock private HttpClientProvider httpClient; + private SeaTunnelRowType rowType; + + @BeforeEach + void setUp() { + MockitoAnnotations.openMocks(this); + rowType = + new SeaTunnelRowType( + new String[] {"messages", "user_id", "agent_id"}, + new SeaTunnelDataType[] { + ArrayType.of(BasicType.STRING_TYPE), + BasicType.STRING_TYPE, + BasicType.STRING_TYPE + }); + } + + private Mem0SinkWriter writer() throws Exception { + HttpParameter parameter = new HttpParameter(); + parameter.setUrl("https://api.mem0.ai/v3/memories/add/";); + Map<String, String> headers = new HashMap<>(); + headers.put("Authorization", "Token secret"); + headers.put("Content-Type", "application/json"); + headers.put("Accept", "application/json"); + parameter.setHeaders(headers); + Mem0SinkWriter writer = + new Mem0SinkWriter( + rowType, parameter, "messages", "user_id", null, null, null, null); + Field field = Mem0SinkWriter.class.getDeclaredField("httpClient"); + field.setAccessible(true); + field.set(writer, httpClient); + return writer; + } + + @Test + void sendsV3AddPayloadAndRequiresEventId() throws Exception { + when(httpClient.doPost(anyString(), any(), anyString())) + .thenReturn(new HttpResponse(202, "{\"event_id\":\"evt-1\"}")); + + writer().write(new SeaTunnelRow(new Object[] {new String[] {"hello"}, "u-1", null})); + + org.mockito.ArgumentCaptor<String> body = org.mockito.ArgumentCaptor.forClass(String.class); + verify(httpClient).doPost(anyString(), any(), body.capture()); + JsonNode json = new ObjectMapper().readTree(body.getValue()); + assertEquals("hello", json.get("messages").get(0).asText()); + assertEquals("u-1", json.get("user_id").asText()); + } + + @Test + void rejectsAcceptedResponseWithoutEventId() throws Exception { + when(httpClient.doPost(anyString(), any(), anyString())) + .thenReturn(new HttpResponse(202, "{}")); + + assertThrows( + java.io.IOException.class, + () -> + writer().write( + new SeaTunnelRow( + new Object[] { + new String[] {"hello"}, "u-1", null + }))); + } + + @Test + void retriesRateLimitAndServerResponses() throws Exception { + when(httpClient.doPost(anyString(), any(), anyString())) + .thenReturn(new HttpResponse(429, "{}")) + .thenReturn(new HttpResponse(503, "{}")) + .thenReturn(new HttpResponse(202, "{\"event_id\":\"evt-3\"}")); + HttpParameter parameter = parameter(); + parameter.setRetry(2); + parameter.setRetryBackoffMultiplierMillis(1); + parameter.setRetryBackoffMaxMillis(2); + Mem0SinkWriter writer = + new Mem0SinkWriter( + rowType, parameter, "messages", "user_id", null, null, null, null); + injectClient(writer); + + writer.write(new SeaTunnelRow(new Object[] {new String[] {"hello"}, "u-1", null})); + + verify(httpClient, org.mockito.Mockito.times(3)).doPost(anyString(), any(), anyString()); + } + + @Test + void onlyTransientStatusesAreRetryable() { + org.junit.jupiter.api.Assertions.assertTrue(Mem0SinkWriter.isRetryableStatus(408)); + org.junit.jupiter.api.Assertions.assertTrue(Mem0SinkWriter.isRetryableStatus(429)); + org.junit.jupiter.api.Assertions.assertTrue(Mem0SinkWriter.isRetryableStatus(500)); + org.junit.jupiter.api.Assertions.assertFalse(Mem0SinkWriter.isRetryableStatus(400)); + org.junit.jupiter.api.Assertions.assertFalse(Mem0SinkWriter.isRetryableStatus(401)); + } + + @Test + void acceptsAnotherScopeWhenConfiguredUserIdIsNull() throws Exception { + when(httpClient.doPost(anyString(), any(), anyString())) + .thenReturn(new HttpResponse(202, "{\"event_id\":\"evt-2\"}")); + Mem0SinkWriter writer = + new Mem0SinkWriter( + rowType, parameter(), "messages", "user_id", "agent_id", null, null, null); + injectClient(writer); + writer.write(new SeaTunnelRow(new Object[] {new String[] {"hello"}, null, "a-1"})); + } + + private HttpParameter parameter() { + HttpParameter parameter = new HttpParameter(); + parameter.setUrl("https://api.mem0.ai/v3/memories/add/";); + Map<String, String> headers = new HashMap<>(); + headers.put("Authorization", "Token secret"); + headers.put("Content-Type", "application/json"); + headers.put("Accept", "application/json"); + parameter.setHeaders(headers); + return parameter; + } + + private void injectClient(Mem0SinkWriter writer) throws Exception { + Field field = Mem0SinkWriter.class.getDeclaredField("httpClient"); + field.setAccessible(true); + field.set(writer, httpClient); + } +} diff --git a/seatunnel-connectors-v2/connector-http/pom.xml b/seatunnel-connectors-v2/connector-http/pom.xml index 133cf63a74..05f73a000c 100644 --- a/seatunnel-connectors-v2/connector-http/pom.xml +++ b/seatunnel-connectors-v2/connector-http/pom.xml @@ -48,6 +48,7 @@ <module>connector-http-zendesk</module> <module>connector-http-stripe</module> <module>connector-http-linear</module> + <module>connector-mem0</module> </modules> </project> diff --git a/seatunnel-dist/pom.xml b/seatunnel-dist/pom.xml index 27ddbfc396..a029ecd37b 100644 --- a/seatunnel-dist/pom.xml +++ b/seatunnel-dist/pom.xml @@ -698,6 +698,12 @@ <version>${project.version}</version> <scope>provided</scope> </dependency> + <dependency> + <groupId>org.apache.seatunnel</groupId> + <artifactId>connector-mem0</artifactId> + <version>${project.version}</version> + <scope>provided</scope> + </dependency> <dependency> <groupId>org.apache.seatunnel</groupId> <artifactId>connector-rabbitmq</artifactId>
