CryoThrust commented on issue #12102:
URL: https://github.com/apache/seatunnel/issues/12102#issuecomment-5551354781
## STIP-38: Phase-1 design for a Mem0-compatible sink
This turns the approved MVP boundary into a reviewable design before
connector code starts. It intentionally remains a **connector-local source ->
memory sink**. It does not add an engine/core/SPI memory abstraction, a Mem0
source, conversational ingestion, or a scope-delete operation.
### 1. Connector boundary and configuration contract
Proposed module name: `connector-mem0`.
The connector accepts a normal SeaTunnel row and sends one logical operation
per row. The first version has a deliberately small configuration surface:
| Option | Type | Meaning |
| --- | --- | --- |
| `api_base_url` | required string | Base URL of the self-hosted
Mem0-compatible REST endpoint; validation rejects malformed URLs. |
| `api_key` | optional secret/string | Credential, supplied through the
connector's secret/config mechanism and never logged. Existing `headers`,
connect/socket timeout, and proxy/TLS configuration remain the HTTP-base
configuration surface. |
| `operation` | enum, default `UPSERT` | Static default: `UPSERT` or
`DELETE`. |
| `operation_field` | optional field name | When set, a row value may
override the default with `UPSERT` or `DELETE`; unknown/null values fail that
row rather than falling back silently. |
| `memory_field` | field name | Required for every possible UPSERT row;
string content to add/update. |
| `memory_id_field` | field name | Required for every possible DELETE row;
identifies exactly one backend memory. |
| `user_id_field` | field name | Required scope for UPSERT. |
| `agent_id_field`, `run_id_field` | optional field names | Optional Mem0
scope dimensions for UPSERT. |
| `metadata_field` | optional field name | JSON object copied as user
metadata for UPSERT. |
| `request_batch_size` | integer, default 1 | Batches transport only after
the backend endpoint and per-record acknowledgement semantics are confirmed. It
is not an acknowledgement shortcut. |
Field-name options refer to the resolved SeaTunnel row schema, not an
implicit JSONPath language. This keeps schema validation deterministic and
avoids a second selector syntax in phase 1.
Startup validation rejects:
1. unsupported static `operation`;
2. missing `memory_field` when an UPSERT is possible;
3. missing `memory_id_field` when a DELETE is possible;
4. missing `user_id_field` when an UPSERT is possible;
5. duplicate/empty field selectors and selectors absent from the resolved
input schema;
6. a metadata field whose declared type cannot represent an object.
Per-row validation rejects an invalid dynamic operation, non-string/empty
UPSERT content, absent/empty required scope, an absent/empty DELETE memory id,
or metadata that is not an object. Phase 1 has **no inferred scope deletion**:
null content, null id, or a CDC DELETE must never turn into deletion of all
memories for a user/agent/run.
### 2. Request mapping
The exact wire adapter is isolated behind a small Mem0 client because
self-hosted versions differ. The following logical mapping is stable:
```text
UPSERT row
memory_field -> request memory/content
user_id_field -> request user_id
agent_id_field -> request agent_id (when present)
run_id_field -> request run_id (when present)
metadata_field -> request metadata (when present)
DELETE row
memory_id_field -> request memory identifier
```
The connector owns only its reserved transport headers; it must not
overwrite user metadata. Response bodies, tokens, and credential-bearing
request details are never included in exception text or logs.
### 3. Delivery and retry contract
The initial delivery guarantee is **at-least-once per row**. It must not
claim exactly-once: a network failure after backend acceptance is
indistinguishable from a failure before acceptance unless the selected backend
exposes a durable idempotency mechanism.
A future backend-specific idempotency header can be enabled only after its
server-side semantics are documented and covered by integration tests. A
locally derived hash alone is not a backend guarantee.
Each logical row has an individual outcome:
| Outcome | Action |
| --- | --- |
| Connect/read timeout, connection reset, HTTP 408/429, 5xx | Retry subject
to the configured policy; preserve the original row identity in error
reporting. |
| HTTP 401/403, schema/request validation failure, other non-retryable 4xx |
Do not retry; fail the row/job through the normal connector error path with a
sanitized diagnostic. |
| Successful acknowledgement for exactly that row | Mark only that row
successful. |
| Batch response with mixed outcomes | Acknowledge successful rows only;
retry/surface unsuccessful rows individually. A whole batch must never be
acknowledged because one item succeeded. |
| Ambiguous response or unknown response shape | Treat as unsuccessful and
surface a sanitized protocol error; do not infer success. |
Ordering is preserved only inside the connector's single writer sequence. No
cross-batch ordering or exactly-once claim is made in Phase 1.
### 4. HTTP-base integration assessment
The implementation will reuse `HttpParameter` / `HttpCommonOptions` for
headers, timeouts and I/O retry settings, and `HttpClientProvider` for client
lifecycle where its APIs fit.
However, the current generic `HttpSinkWriter` is not sufficient as-is: it
logs and swallows non-200 responses, and `HttpClientProvider` retries I/O
exceptions rather than classifying response status. That behavior cannot
enforce the contract above. The implementation proposal will therefore either:
1. add a narrowly scoped HTTP-base extension that exposes status/body
handling and preserves headers for all needed verbs, with existing connector
regression coverage; or
2. use a connector-owned Mem0 client while reusing the existing common HTTP
parameter/configuration path.
The choice will be made in the implementation PR after a small compatibility
spike. It will not change engine/core/SPI contracts.
### 5. Test matrix and acceptance gates
**Unit tests**
- all startup invalid combinations and resolved-schema errors;
- UPSERT/DELETE serialization and optional scope/metadata mapping;
- dynamic-operation validation;
- response status classification and sanitization;
- retryability classification;
- no accidental scope deletion.
**Deterministic integration tests**
- local mock server verifies paths, headers, redaction, request order and
body;
- 200/201 success, 400/401/403 terminal failures, 408/429/5xx retry behavior;
- connection failure and ambiguous post-write failure;
- mixed per-record batch outcomes;
- duplicate delivery after an ambiguous failure, proving the documented
at-least-once behavior.
**E2E decision**
- a Mem0 Testcontainers profile is optional, not a prerequisite. It is added
only if an official image can start deterministically within CI limits. The
mock-server suite is the required acceptance gate.
**Docs/examples**
- JDBC -> Mem0 batch hydration;
- MySQL-CDC -> Mem0 UPSERT/DELETE;
- explicit delivery guarantee, required scopes, retry behavior, redaction,
and Phase-1 non-goals.
### 6. Compatibility and implementation sequence
1. Maintainer review of this contract.
2. A small implementation PR for the connector, tests, examples and docs
only.
3. Only after a second memory backend is real and reviewed, reconsider a
shared abstraction or a memory source/export path.
Please confirm whether the field-name contract and the at-least-once/per-row
failure boundary are acceptable. In particular, I would like confirmation that
Phase 1 should keep DELETE limited to a single `memory_id`, rather than
introducing scope deletion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
