Apache SeaTunnel Output Connector

Apache SeaTunnel Output Connector

Module: oc-seatunnel-output-connector
Class: SeaTunnelOutputConnector
Spring Bean: @Component (activated when spring.opencrawling.output.type=seatunnel)

---

📌 Overview

The Apache SeaTunnel Output Connector integrates Apache SeaTunnel (v2.3.13) into OpenCrawling as a distributed data integration engine and fan-out orchestrator. It enables high-throughput document crawling, Apache Tika text extraction, token chunking, zero-trust ACL security mapping, and dense vector embeddings to fan out simultaneously to multiple downstream destinations across your enterprise data fabric.

Using SeaTunnel's native Zeta Engine, OpenCrawling pipelines can automatically route and replicate Open Ingestion Standard (OIS) streams to 100+ destination sinks without writing custom sync pipelines:

• Vector Databases: Milvus, Qdrant, Vespa
• Data Lakehouses & Warehouses: Apache Iceberg, ClickHouse, StarRocks, Snowflake, Amazon S3
• Search Engines: Elasticsearch, OpenSearch, Apache Solr, Luxir
• Message Brokers: Apache Kafka, Apache Pulsar

flowchart TD
    A["Crawled Document Chunk"] --> B["oc-embedding-consumer (Spring AI / Ollama)"]
    B --> C["Kafka Topic (opencrawling-embedded)"]
    C --> D["SeaTunnelStoreWriterConsumer (oc-seatunnel-output-connector)"]
    D --> E["Fan-Out Kafka Topic (seatunnel-ois-in)"]
    
    subgraph ControlPlane["Control Plane"]
        F["SeaTunnelOutputConnector"] --> G["SeaTunnelJobConfigBuilder"]
        G --> H["SeaTunnelRestClient (HTTP REST API)"]
        H -->|Deploy HOCON DAG via REST| I["Apache SeaTunnel Zeta Cluster (Port 8088)"]
    end
    
    subgraph ExecutionEngine["Execution Engine"]
        I --> J["Kafka Source Connector (connector-kafka-2.3.13)"]
        E -.-> J
        J --> K{"SeaTunnel Fan-Out Sinks"}
        K --> L["ClickHouse Sink"]
        K --> M["Milvus Sink"]
        K --> N["Apache Iceberg Sink"]
        K --> O["Qdrant / Console Sink"]
    end

Loading

---

🚀 Key Features

• Automated HOCON DAG Generation: SeaTunnelJobConfigBuilder automatically compiles OpenCrawling job parameters, parallelism settings, checkpointing intervals, Kafka source specifications, and target sinks into standardized SeaTunnel HOCON configuration DAGs.
• REST API v2 Job Orchestration: SeaTunnelRestClient automatically communicates with the SeaTunnel Zeta cluster REST API over HTTP to submit streaming jobs (POST /submit-job), monitor running jobs (GET /running-jobs), check finished jobs (GET /finished-jobs), inspect metrics (GET /job-info/{jobId}), and retrieve cluster health (GET /overview).
• Zero-Cluster Footprint (Stock SeaTunnel Compatibility): Uses standard stock apache/seatunnel:2.3.13 distributions with the pre-bundled connector-kafka. No custom plugins or custom JARs are required on the SeaTunnel workers, avoiding Java 25 preview classpath conflicts (UnsupportedClassVersionError).
• Open Ingestion Standard (OIS) Catalog Mapping: Maps all document fields, precomputed vector embeddings (array<float>), and metadata attributes to SeaTunnel's strongly typed catalog schema:
  • id (string): Chunk / Document identifier
  • doc_id (string): Parent document identifier
  • action (string): OIS action (UPSERT, DELETE, PURGE)
  • uri (string): Document source URI
  • text (string): Chunk extracted text content
  • vector (array<float>): Dense embedding vector
  • acl (array<string>): Zero-Trust document ACL tokens
  • security_allowed_read (array<string>): Allowed reader SIDs
  • security_denied_read (array<string>): Denied reader SIDs
  • security_inheritance (boolean): Security inheritance flag
  • last_modified (string): ISO-8601 timestamp
  • metadata (string): JSON serialized metadata map
• Document Lifecycle & Tombstone Delete CDC: Full tombstone purges (DocumentAction.DELETE) map to SeaTunnel's Change Data Capture (CDC) row kinds (-D / RowKind.DELETE), enabling downstream sinks to purge removed documents without performing costly re-embeddings.
• Admin UI & Health Checker Integration: Fully registered in oc-admin-ui (ConnectorForm.tsx, JobTable.tsx) and verified dynamically by ConnectorCheckerService using SeaTunnelRestClient.isReachable().
• Decoupled Architecture: Powered by SeaTunnelStoreWriterConsumer consuming from the Kafka opencrawling-embedded topic for horizontal scalability and zero-loss streaming.

---

⚙️ Configuration Properties

All properties are prefixed with spring.opencrawling.output.seatunnel.*:

Property	Default	Description
spring.opencrawling.output.seatunnel.rest-url	http://localhost:8080	HTTP endpoint URL of the SeaTunnel Zeta master REST API (typically mapped to port 8088 or 8080)
spring.opencrawling.output.seatunnel.job-name	opencrawling_ingestion_pipeline	SeaTunnel job name identifier
spring.opencrawling.output.seatunnel.job-mode	STREAMING	Execution mode: STREAMING or BATCH
spring.opencrawling.output.seatunnel.checkpoint-interval-ms	5000	State checkpoint interval in milliseconds
spring.opencrawling.output.seatunnel.parallelism	4	Execution parallelism degree across Zeta workers
spring.opencrawling.output.seatunnel.kafka-bootstrap-servers	localhost:9092	Kafka broker bootstrap servers for the intermediate fan-out topic
spring.opencrawling.output.seatunnel.kafka-topic	opencrawling-embedded	Topic containing embedded document chunks
spring.opencrawling.output.seatunnel.kafka-group-id	seatunnel-ingestion-consumer	Consumer group ID for the SeaTunnel Kafka source
spring.opencrawling.output.seatunnel.target-sinks	console	Comma-separated list of target SeaTunnel sinks (e.g. console, clickhouse, milvus, qdrant, iceberg)
spring.opencrawling.output.seatunnel.dimensions	1024	Vector dimensionality (matches default mxbai-embed-large)
spring.opencrawling.output.seatunnel.auto-submit-job	true	Automatically submit the generated HOCON DAG to SeaTunnel on connector startup
spring.opencrawling.output.seatunnel.timeout-seconds	30	HTTP connection and socket timeout in seconds

---

📋 Generated SeaTunnel HOCON DAG Example

When auto-submit-job is enabled or SeaTunnelJobConfigBuilder is invoked, the following HOCON DAG is compiled and submitted to SeaTunnel:

env {
  execution.parallelism = 4
  job.mode = "STREAMING"
  checkpoint.interval = 5000
}

source {
  Kafka {
    bootstrap.servers = "localhost:9092"
    topic = "opencrawling-embedded"
    consumer.group = "seatunnel-ingestion-consumer"
    result_table_name = "ois_embedded_chunks"
    format = "json"
    schema = {
      fields {
        id = "string"
        doc_id = "string"
        action = "string"
        uri = "string"
        text = "string"
        vector = "array<float>"
        acl = "array<string>"
        security_allowed_read = "array<string>"
        security_denied_read = "array<string>"
        security_inheritance = "boolean"
        last_modified = "string"
        metadata = "string"
      }
    }
  }
}

sink {
  Clickhouse {
    source_table_name = "ois_embedded_chunks"
    host = "localhost:8123"
    database = "default"
    table = "document_chunks"
  }
  Milvus {
    source_table_name = "ois_embedded_chunks"
    url = "http://localhost:19530"
    collection = "enterprise_kb"
  }
  Console {
    source_table_name = "ois_embedded_chunks"
  }
}

---

🔍 Interacting with Apache SeaTunnel Zeta REST API

1. Cluster Overview & Health

Retrieve master node information, memory usage, and cluster status:

curl -s http://localhost:8088/overview | jq .

2. Check Running Jobs

List all actively executing streaming pipelines:

curl -s http://localhost:8088/running-jobs | jq .

3. Check Finished Jobs

List completed or terminated jobs:

curl -s http://localhost:8088/finished-jobs | jq .

4. Query Job Detail & Metrics

Inspect execution metrics for a specific SeaTunnel job ID:

curl -s http://localhost:8088/job-info/{jobId} | jq .

5. Stop a Running Pipeline

Gracefully cancel a streaming job:

curl -X POST "http://localhost:8088/stop-job" \
     -H "Content-Type: application/json" \
     -d '{"jobId": "<job-id>"}'

---

🐳 Docker & Integration Test

Standalone SeaTunnel Zeta Docker Compose

Spin up an Apache SeaTunnel Zeta cluster using the standalone Compose definition:

docker compose -f oc-seatunnel-output-connector/docker/docker-compose.yml up -d

Decoupled Multi-Service Deployment

The decoupled architecture definition is located at oc-seatunnel-output-connector/docker/docker-compose-decoupled-with-seatunnel.yml and includes:

• Apache SeaTunnel Zeta Cluster: apache/seatunnel:2.3.13 exposing REST on port 8088 and Hazelcast RPC on 5801.
• Kafka & Zookeeper: Distributed messaging backbone.
• Redis: Distributed deduplication and Claim Check cache.
• Ollama & Ollama Model Puller: Automatically pulls mxbai-embed-large (1024 dims).
• OpenCrawling Microservices: oc-crawler, oc-ingestion-consumer, oc-embedding-consumer, oc-writer-consumer, and oc-mcp-server.

Automated Decoupled Integration Test Script

Run the automated end-to-end multi-service pipeline integration test:

./scripts/test-seatunnel-decoupled.sh

The script features detailed live logging, including:

1. Docker Compose stack startup and status verification (compose ps).
2. Health-check readiness counters for SeaTunnel Zeta (:8088), Kafka (:9092), and Ollama (:11434).
3. Init container tracking for the mxbai-embed-large embedding model.
4. Crawler execution tracking, exit code validation (System.exit(0)), and crawler scan log output (compose logs oc-crawler).
5. Live writer consumer log output (compose logs oc-writer-consumer) confirming document chunk serialization and Kafka streaming.
6. SeaTunnel REST API job submission and running job status verification (/running-jobs).
7. MCP Server HTTP endpoint accessibility check on port 8080.
8. OIS Document Lifecycle Tombstone DELETE action integration testing via SeaTunnelStoreWriterConsumerTest.
9. Clean environment teardown.