Connectors

Connectors Reference

OpenCrawling is built around a connector-based architecture. Every step of the pipeline — ingesting content from a source, transforming it via embeddings, and writing it to a destination — is handled by a dedicated connector implementing a well-defined Java interface.

Connector Types

Type	Interface	Role
Repository Connector	`RepositoryConnector`	Scans a content source and emits `RepositoryDocument` objects as a reactive `Flux` stream
Output Connector	`OutputConnector`	Receives a `RepositoryDocument` and writes it directly to a destination (sync/direct mode)
Transformation Connector	Configured via Admin UI	Selects the embedding engine (Ollama, OpenAI, etc.) for the `oc-embedding-service`

Core Interfaces

RepositoryConnector (in oc-core):

public non-sealed interface RepositoryConnector extends Connector {
    Flux<RepositoryDocument> scan(String basePath);
}

OutputConnector (in oc-core):

public non-sealed interface OutputConnector extends Connector {
    Mono<Void> send(RepositoryDocument document);
}

Both extend Connector which provides getName(), connect(), and disconnect() lifecycle methods.

Repository Connectors

Repository connectors are responsible for discovering and streaming documents from content sources into the pipeline.

📁 FileSystem Repository Connector

Module: oc-filesystem-repository-connector
Class: FileSystemRepositoryConnector
Spring Bean: @Primary — selected by default when no other connector is specified

Description

Recursively scans a local filesystem directory using Java NIO and emits each regular file as a RepositoryDocument. Subdirectory traversal is parallelised using Java 25 Structured Task Scope for high-throughput I/O on deep directory trees.

Configuration

No external configuration properties are required. The path is provided at runtime via the scan(String basePath) call (e.g., from the Admin UI job form or the crawler SCAN_PATH environment variable).

Metadata extracted per document

Metadata key	Value
`extension`	File extension (e.g. `pdf`, `txt`, `docx`)
`uri`	`file://` URI of the absolute path

Connector type selection

Set the connector type to filesystem in the Admin UI or via the environment variable:

# docker-compose-decoupled.yml
environment:
  CONNECTOR_TYPE: filesystem
  SCAN_PATH: /data

Notes

Files are streamed; content is not loaded into memory until processed.
The uri is a file:// URI (e.g. file:///data/report.pdf), so no Claim Check Pattern is used — the IngestionConsumer reads directly from the local filesystem path.

🏢 Alfresco Repository Connector

Module: oc-alfresco-repository-connector
Class: AlfrescoRepositoryConnector
Spring Bean: @Component (activated when CONNECTOR_TYPE=alfresco)

Description

Connects to an Alfresco Content Services (Community or Enterprise) instance via the Alfresco REST API v1 and recursively scans the repository node tree. It supports scanning from the root node, a path (e.g. /Company Home/Sites), or a specific Node UUID. Folder traversal is parallelised using Java 25 Structured Task Scope. Document content is streamed via the /nodes/{nodeId}/content endpoint and saved to a Claim Check shared file before being published to Kafka.

Configuration Properties

Property	Environment variable	Default	Description
`spring.opencrawling.connector.alfresco.url`	`ALFRESCO_URL`	`http://localhost:8080/alfresco/api/-default-/public/alfresco/versions/1`	Base URL of the Alfresco REST API
`spring.opencrawling.connector.alfresco.username`	`ALFRESCO_USERNAME`	`admin`	Alfresco username
`spring.opencrawling.connector.alfresco.password`	`ALFRESCO_PASSWORD`	`admin`	Alfresco password (Basic Auth)
`spring.opencrawling.connector.alfresco.batch-size`	`ALFRESCO_BATCH_SIZE`	`100`	Number of children fetched per API page

Docker Compose example

oc-crawler:
  environment:
    CONNECTOR_TYPE: alfresco
    SCAN_PATH: /Company Home
    ALFRESCO_URL: http://alfresco:8080/alfresco/api/-default-/public/alfresco/versions/1
    ALFRESCO_USERNAME: admin
    ALFRESCO_PASSWORD: admin

`SCAN_PATH` formats

Format	Behaviour
(empty) or `-root-`	Scans from the Alfresco repository root
`/Company Home/Sites/my-site`	Scans using a relative path from root
`a1b2c3d4-...` (UUID)	Scans a specific node by its Alfresco Node ID

Metadata extracted per document

Metadata key	Value source
`name`	`entry.name` from the API response
`nodeType`	`entry.nodeType` (e.g. `cm:content`)
`mimeType`	`entry.content.mimeType`
`sizeInBytes`	`entry.content.sizeInBytes`
`cm:title`, `cm:description`, etc.	All Alfresco `cm:*` properties returned via `?include=properties`

Claim Check Pattern

Because Alfresco documents are fetched from a remote HTTP server, the connector downloads the content stream and saves it to a local shared directory (/data/claims/) before publishing the reference to Kafka. This decouples the slow HTTP download from the fast Kafka publish step.

Note

The shared /data directory must be volume-mounted and accessible by both the crawler and the consumer containers in the decoupled deployment.

PGVector verification queries

After a crawl job completes, verify ingestion with:

-- Total chunks stored for Alfresco documents
SELECT count(*) FROM vector_store_1024;

-- Inspect content and source URI of stored chunks
SELECT id,
       LEFT(content, 200)     AS preview,
       metadata->>'name'      AS filename,
       metadata->>'mimeType'  AS mime_type,
       metadata->>'uri'       AS source_uri
FROM vector_store_1024
ORDER BY id
LIMIT 10;

-- Semantic similarity search
SELECT id,
       LEFT(content, 300)    AS preview,
       metadata->>'name'     AS filename,
       1 - (embedding <=> (
           SELECT embedding FROM vector_store_1024 LIMIT 1
       ))                    AS similarity
FROM vector_store_1024
ORDER BY similarity DESC
LIMIT 5;

🧊 Apache Iceberg Repository Connector

Module: oc-iceberg-repository-connector
Class: IcebergRepositoryConnector
Spring Bean: @Component (activated when CONNECTOR_TYPE=iceberg)

Description

Connects to an Apache Iceberg table catalog and scans structured data tables, emitting each Iceberg Record as a JSON-serialized RepositoryDocument. Supports REST, Hive Metastore, Hadoop, AWS Glue, and in-memory catalog backends. File scan tasks are parallelized using Java 25 Structured Task Scope for high-throughput Parquet file reading.

Each record is serialized to JSON and treated as a document, making Iceberg data lakes a first-class ingestion source for RAG pipelines. This connector reads Parquet data files directly from the catalog's underlying storage (local FS, S3, GCS, ADLS, etc.).

Configuration Properties (Admin UI / `connectors.json`)

Property key	Default	Description
`catalogType`	`in-memory`	Catalog backend type: `rest`, `hive`, `hadoop`, `glue`, or `in-memory`
`catalogUri`	(empty)	URI of the catalog server (required for `rest` and `hive` types)
`warehouse`	`tmp/iceberg-warehouse`	Warehouse root path (local path, `s3://`, `gs://`, `abfs://`, etc.)
`idColumn`	(empty)	Optional column name to use as the document ID. Falls back to Iceberg identifier fields, then SHA-256 of JSON content

Catalog Types

`catalogType`	Implementation	Required `catalogUri`
`rest`	`RESTCatalog`	REST catalog server URL (e.g. `http://localhost:8181`)
`hive`	`HiveCatalog`	Hive Metastore Thrift URI (e.g. `thrift://hive:9083`)
`hadoop`	`HadoopCatalog`	Not required — uses `warehouse` path directly
`glue`	`GlueCatalog` (via reflection)	Not required — uses AWS SDK credentials chain
`in-memory`	`InMemoryCatalog`	Not required — ephemeral, for testing only

`SCAN_PATH` format

The basePath argument to scan() is an Iceberg table identifier in the format <namespace>.<table>:

Format	Example	Description
`namespace.table`	`analytics.events`	Scans the `events` table in the `analytics` namespace
`db.schema.table`	`prod.dbo.orders`	Three-part identifier (catalog-specific)

Metadata extracted per document

Metadata key	Value
All Iceberg record columns	Included as metadata entries (column name → string value)
`mimeType`	Always `application/json`
`sizeInBytes`	Size of the JSON-serialized record in bytes
`uri`	`iceberg://<table-name>/<document-id>`

The document ID is resolved in order of priority:

The column specified in idColumn config property
Iceberg table's declared identifier fields
SHA-256 hash of the JSON record content (fallback)

Docker Compose example (with Iceberg REST Catalog + MinIO)

Use the dedicated compose file provided in the oc-iceberg-repository-connector module to spin up a local Iceberg REST catalog backed by MinIO (S3-compatible object storage):

# oc-iceberg-repository-connector/docker-compose.yml
services:
  minio:
    image: minio/minio:RELEASE.2024-05-10T01-39-19Z
    container_name: minio-iceberg
    environment:
      MINIO_ROOT_USER: admin
      MINIO_ROOT_PASSWORD: password
    ports:
      - "9000:9000"
      - "9001:9001"
    command: ["server", "/data", "--console-address", ":9001"]

  iceberg-rest:
    image: apache/iceberg-rest-fixture:1.5.0
    container_name: iceberg-rest
    ports:
      - "8181:8181"
    environment:
      - AWS_ACCESS_KEY_ID=admin
      - AWS_SECRET_ACCESS_KEY=password
      - AWS_REGION=us-east-1
      - CATALOG_WAREHOUSE=s3://warehouse/
      - CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO
      - CATALOG_S3_ENDPOINT=http://minio:9000
      - CATALOG_S3_PATH_STYLE_ACCESS=true
    depends_on:
      - minio

Then configure the crawler to use this catalog:

oc-crawler:
  environment:
    CONNECTOR_TYPE: iceberg
    SCAN_PATH: analytics.events
    ICEBERG_CATALOG_TYPE: rest
    ICEBERG_CATALOG_URI: http://iceberg-rest:8181
    ICEBERG_WAREHOUSE: s3://warehouse/

Notes

The connector does not use the Claim Check Pattern — Iceberg records are serialized to JSON in-process and streamed directly as RepositoryDocument objects. No shared volume is required.
The uri field uses an iceberg:// scheme that identifies the source table and record ID.
For timestamp-based last-modified detection, the connector inspects well-known column names: last_modified, updated_at, timestamp, and date.
AWS Glue catalog requires the iceberg-aws dependency on the classpath and AWS credentials configured via environment variables or instance profile.

Output Connectors

Output connectors receive a RepositoryDocument and persist it directly (bypassing Kafka — used in the direct/standalone execution mode).

🗄️ Vector Output Connector

Module: oc-vector-output-connector
Class: VectorOutputConnector
Spring Bean: @Component

Description

Parses document content using Apache Tika (with full parser support for PDF, Word, PowerPoint, Excel, HTML, plain text, and more), splits the extracted text into overlapping chunks, generates embeddings by calling the configured embedding model, and writes the resulting vectors to PGVector (PostgreSQL with the pgvector extension).

This connector is used in the direct output mode (where the crawler and vectorisation happen in the same JVM process). In the decoupled mode, the equivalent pipeline is executed across three separate microservices: oc-ingestion-consumer, oc-embedding-service, and oc-runtime (VectorStoreWriterConsumer).

Configuration Properties

Property	Default	Description
`spring.ai.vectorstore.pgvector.url`	—	JDBC URL of the PGVector database
`spring.ai.vectorstore.pgvector.username`	—	Database username
`spring.ai.vectorstore.pgvector.password`	—	Database password
`spring.ai.vectorstore.pgvector.initialize-schema`	`true`	Auto-creates the `vector_store` tables on startup
`spring.ai.vectorstore.pgvector.dimensions`	`1536`	Default vector dimensions (must match the embedding model)

Vector Store Tables

The connector maintains four separate tables to support different embedding model dimension sizes simultaneously:

Table	Dimensions	Typical model
`vector_store`	1536 (default)	OpenAI `text-embedding-3-small`
`vector_store_384`	384	`all-minilm` (Ollama)
`vector_store_768`	768	`nomic-embed-text` (Ollama)
`vector_store_1024`	1024	`mxbai-embed-large` (Ollama)

The correct table is selected automatically based on the length of the embedding vector returned by the model.

Text parsing

Apache Tika's AutoDetectParser handles format detection and extraction. If parsing fails for a specific file (e.g., a FreeMarker .ftl template mistakenly detected as RSS/XML), the connector logs a warning and falls back to reading the file as UTF-8 plain text.

🍃 Milvus Output Connector

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

Description

Persists precomputed vector embeddings and text chunks directly into a Milvus vector database collection. It supports standard index types (e.g., HNSW), custom metric types (e.g., COSINE), and user token authentication.

In the decoupled architecture, MilvusStoreWriterConsumer consumes embedded chunks from the Kafka opencrawling-embedded topic and saves them using this connector.

Configuration Properties

Property	Default	Description
`SPRING_OPENCRAWLING_OUTPUT_MILVUS_URI`	`http://localhost:19530`	URI/Host endpoint of the Milvus standalone instance
`SPRING_OPENCRAWLING_OUTPUT_MILVUS_TOKEN`	`root:Milvus`	Authentication token or username:password
`SPRING_OPENCRAWLING_OUTPUT_MILVUS_COLLECTION_NAME`	`enterprise_kb`	Target Milvus collection name
`SPRING_OPENCRAWLING_OUTPUT_MILVUS_VECTOR_FIELD_NAME`	`embeddings`	Name of the field containing the vector embeddings
`SPRING_OPENCRAWLING_OUTPUT_MILVUS_DIMENSIONS`	`1024`	Vector dimension size (e.g., 1024 for `mxbai-embed-large`)

Transformation Connectors (Embedding Engines)

Transformation connectors are not Java classes but named configurations stored in connectors.json and managed via the Admin UI. They define which embedding engine and model the oc-embedding-service should use when processing chunks from Kafka.

Supported Engines

Ollama (local)

Runs a local embedding model via the Ollama server. No API key required.

Config key	Example value	Description
`engine`	`ollama`	Engine type identifier
`model`	`mxbai-embed-large`	Ollama model name (must be pre-pulled)
`baseUrl`	`http://localhost:11434`	Ollama server base URL

Recommended models by dimension:

Model	Dimensions	Pull command
`mxbai-embed-large`	1024	`ollama pull mxbai-embed-large`
`nomic-embed-text`	768	`ollama pull nomic-embed-text`
`all-minilm`	384	`ollama pull all-minilm`

OpenAI

Uses the OpenAI Embeddings API. Requires an API key.

Config key	Example value	Description
`engine`	`openai`	Engine type identifier
`model`	`text-embedding-3-small`	OpenAI model name
`apiKey`	`sk-...`	OpenAI API key

Adding a New Connector

To implement a custom Repository Connector:

Create a new Maven module (e.g. oc-s3-repository-connector).
Add oc-core as a dependency.
Implement RepositoryConnector:

@Component
public class S3RepositoryConnector implements RepositoryConnector {

    @Override
    public String getName() { return "S3Connector"; }

    @Override
    public void connect() throws Exception { /* initialise S3 client */ }

    @Override
    public void disconnect() throws Exception { /* shutdown */ }

    @Override
    public Flux<RepositoryDocument> scan(String basePath) {
        return Flux.create(sink -> {
            // list S3 objects under basePath prefix
            // for each object: sink.next(new RepositoryDocument(...))
            sink.complete();
        });
    }
}

Add the module to the root pom.xml and to oc-runtime's dependencies.
Register it in docker-compose-decoupled.yml with CONNECTOR_TYPE: s3.

See Repository Connector Execution Modes for how connectors are wired into the runtime.

🗺️ Documentation

🧪 Testing

Integration Test — Decoupled Pipeline

Uh oh!

Connectors

Connectors Reference

Connector Types

Core Interfaces

Repository Connectors

📁 FileSystem Repository Connector

Description

Configuration

Metadata extracted per document

Connector type selection

Notes

🏢 Alfresco Repository Connector

Description

Configuration Properties

Docker Compose example

SCAN_PATH formats

Metadata extracted per document

Claim Check Pattern

PGVector verification queries

🧊 Apache Iceberg Repository Connector

Description

Configuration Properties (Admin UI / connectors.json)

Catalog Types

SCAN_PATH format

Metadata extracted per document

Docker Compose example (with Iceberg REST Catalog + MinIO)

Notes

Output Connectors

🗄️ Vector Output Connector

Description

Configuration Properties

Vector Store Tables

Text parsing

🍃 Milvus Output Connector

Description

Configuration Properties

Transformation Connectors (Embedding Engines)

Supported Engines

Ollama (local)

OpenAI

Adding a New Connector

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

🗺️ Documentation

🧪 Testing

🤝 Community

Clone this wiki locally

`SCAN_PATH` formats

Configuration Properties (Admin UI / `connectors.json`)

`SCAN_PATH` format