Apache Ozone Output Connector

Apache Ozone Output Connector (Migration Mode)

oc-ozone-output-connector migrates enterprise content bit for bit into Apache Ozone object storage. It works only in Migration Mode (opencrawling.pipeline.mode=migration).

The AI/vector output connectors (pgvector, Solr, Luxir, …) index chunks and embeddings for RAG. This connector does the opposite: it skips Tika extraction, chunking and embeddings, and copies each original binary with its metadata and Zero-Trust ACLs into Ozone volumes and buckets.

Note

Migration Mode (this page) is not the Migration Guide. The Migration Guide covers moving Apache ManifoldCF configurations to OpenCrawling.

---

🧭 Pipeline Modes: RAG vs Migration

	rag (default)	migration
Purpose	Semantic search / RAG indexing	Content migration and data liberation
Processing	Tika extraction → chunking → embeddings	None: the original binary is copied unchanged
Output	Vector / search connectors	Apache Ozone Output Connector
Payload	Text chunks + vectors	Binary + OIS JSON sidecar (metadata, ACLs, SHA-256)

How the mode is resolved

• Global default: opencrawling.pipeline.mode (rag or migration, default rag). PipelineAutoConfiguration registers it as PipelineProperties.
• Per job: JobDTO.pipelineMode overrides the global value when set. Unknown values are rejected (PipelineMode.parseStrict).
• Guard: a job whose output is the Ozone connector but runs in rag mode is rejected with HTTP 409. Its status shows Rejected: Ozone output requires migration mode. The connector itself also fails fast with an IllegalStateException in RAG mode.

Where to set it

Client	How
CLI	oc job start --name <name> --connector <conn> --path <path> --mode migration, or oc job create --file job.json --mode migration (overrides pipelineMode in the file). oc job list shows a MODE column, and oc job status shows Pipeline Mode. See OpenCrawling CLI.
Java SDK	JobRequest.builder().pipelineMode("migration"); JobResponse.pipelineMode() (records). See Java Client SDK.
Admin UI	The Job form has a Pipeline Mode radio group (RAG / Migration). Choosing an Ozone output switches it to Migration automatically. Jobs show a Migration / RAG badge, and Ozone outputs have an amber Archive icon.
REST	Set "pipelineMode": "migration" in the job JSON.
Offline validation	oc schema validate --file <envelope.json> checks migration envelopes (contentRef, checksumSha256, security, …).

---

🏗️ Architecture

flowchart LR
    R["Repository Connector"] --> C["oc-crawler-service<br/>(parallel lanes)"]
    C -- "1. upload binary" --> CC[("Claim Check store<br/>source Ozone")]
    C -- "2. publish reference" --> K[["Kafka<br/>opencrawling-documents"]]
    K -- "3. consume<br/>opencrawling-ozone-migration-group" --> W["OzoneMigrationWriterConsumer"]
    W -- "4. stream binary" --> CC
    W -- "5. binary + sidecar + index" --> T[("Target Apache Ozone")]

Loading

Component	Role
org.opencrawling.ozone.OzoneOutputConnector	Output connector. Active when spring.opencrawling.output.type=ozone.
OzoneMigrationWriterConsumer	Decoupled writer. It uses its own consumer group (opencrawling-ozone-migration-group), separate from the RAG ingestion group, so the two never split partitions.
OzoneRpcClaimCheckStrategy / OzoneRpcClaimCheckStrategyFactory	Real NATIVE (Ozone RPC) claim-check client. See Apache Ozone Claim Check Store.
org.opencrawling.core.s3.S3MultipartUploader	Shared S3 uploader with parallel multipart uploads for large objects. Used by the crawler's S3 claim check and by the S3G writer.

---

⚙️ Configuration

spring:
  opencrawling:
    output:
      type: ozone
      ozone:
        client-type: NATIVE          # NATIVE (ofs/RPC) or S3G (S3 Gateway HTTP)
        volume: opencrawling
        bucket: migration
        om-host: localhost
        om-port: 9862
        s3-endpoint: http://localhost:9878
        access-key: any
        secret-key: any
        auto-create-bucket: true
        key-strategy: HIERARCHICAL   # HIERARCHICAL or FLAT
        sidecar-suffix: .ois.json
        tombstone-action: DELETE_KEY # DELETE_KEY or ARCHIVE_TOMBSTONE

opencrawling:
  pipeline:
    mode: migration                  # Required for the Ozone output connector

Additional properties

Property	Default	Description
spring.opencrawling.output.ozone.allow-in-memory-fallback	false	For testing only. When false, an unreachable OzoneManager makes the connector fail instead of silently writing to an in-memory store.
spring.opencrawling.output.ozone.consumer-group	opencrawling-ozone-migration-group	Kafka consumer group of the decoupled writer. It must differ from the RAG ingestion group.
spring.opencrawling.output.ozone.consumer-concurrency	3	Parallel writer threads, one Kafka consumer each.
spring.opencrawling.kafka.topic.partitions	3	Partitions of opencrawling-documents, created at startup by KafkaAdmin. Use the same value on every service.
spring.opencrawling.crawler.concurrency	4	Parallel claim-check upload + Kafka publish lanes of the standalone crawler, for connectors that opt in. 1 means sequential.
spring.opencrawling.output.ozone.multipart-threshold	256MB	S3G writer: objects of at least this size use a parallel multipart upload. 0 disables multipart.
spring.opencrawling.output.ozone.multipart-part-size	16MB	S3G writer: part size (minimum 5MB).
spring.opencrawling.output.ozone.multipart-concurrency	4	S3G writer: parallel part uploads per object.
spring.opencrawling.claim-check.ozone.multipart-threshold / -part-size / -concurrency	256MB / 16MB / 4	The same settings for the crawler's S3 claim-check uploads.

In the Admin UI, the output is listed as Apache Ozone (Migration Mode only). Its form fields are: transport (NATIVE/S3G), volume/bucket (defaults s3v/migration), OM host/port, S3 endpoint, access/secret key, key strategy and tombstone action.

Transports

client-type	Transport	Notes
NATIVE (default)	Ozone RPC (ofs://) to the OM and datanodes	About 30% faster. Streams blocks directly to the datanodes. Every service needs network access to the OM and all datanodes.
S3G	S3 Gateway HTTP (s3://) via the AWS SDK	Only needs the gateway endpoint. Large objects use multipart uploads.

Claim-check transport

The writer reads each binary from the claim-check store that the crawler wrote. With spring.opencrawling.claim-check.store=ozone, spring.opencrawling.claim-check.ozone.client-type selects the transport:

Value	Transport	Settings
NATIVE (default)	Ozone RPC, ofs://<volume>/<bucket>/<key>, via OzoneRpcClaimCheckStrategy	om-host, om-port, volume, bucket, auto-create-bucket
S3	S3 Gateway HTTP, s3://<bucket>/<key>, with multipart for large objects	s3-endpoint, bucket, access-key, secret-key, multipart-*

• The real NATIVE client is registered by this module, and oc-runtime includes it. Applications built on oc-core alone fall back to an in-process store that is not shared across services. They log a warning when it is selected.
• A missing claim-check object fails with NoSuchFileException instead of returning empty content. The Kafka message is then retried, so a 0-byte binary is never migrated.

---

📦 Object Layout

For each UPSERT the connector writes:

Object	Content
<key>	The original binary, streamed bit for bit. SHA-256 is computed while streaming.
<key>.ois.json	OIS sidecar with id, action, metadata, security (inheritanceEnabled, permissions[]) and contentRef (key, volume, bucket, claimCheckUri, filename, mimeType, contentLength, checksumSha256).
.opencrawling/index/<sha256(docId)>	Internal index that maps the document id to its key, so tombstones without metadata remove the right objects.

• HIERARCHICAL keys come from relativePath/file_path (or the file: URI), sanitized against .. traversal. FLAT keys use the sanitized document id.
• The write order is always binary → sidecar → index.
• On DELETE, DELETE_KEY removes the binary, the sidecar and the index entry. ARCHIVE_TOMBSTONE also writes .tombstones/<key>.ois.json. See Document Lifecycle Actions & Tombstones.

---

🚀 Performance Tuning

Writer parallelism

• Effective parallelism is min(consumer-concurrency, topic.partitions). Extra threads stay idle (partitions assigned: []). Writer replicas share the same partitions.
• Ordering is preserved per document. Messages are keyed by document id, so all UPSERT/DELETE events of a document land on one partition and are processed in order.
• Choose the partition count before the first migration. KafkaAdmin can add partitions but never remove them. Adding partitions while a backlog exists remaps keys and can reorder pending events of the same document.
• The S3 Gateway client checks or creates the bucket once per process. The native client caches its connection.

Measured with the e2e load mode: 400 documents, 131 MB, 4 KB–1 MB files, on a single-datanode Ozone on Docker Desktop. Treat the numbers as a lower bound.

Transport	Threads (= partitions)	Throughput	Speed-up
S3G	1	19.1 docs/s, 6.3 MB/s	—
S3G	3	33.3 docs/s, 11.0 MB/s	1.75×
S3G	6	38.4 docs/s, 12.6 MB/s	2.0×
NATIVE	1	25.2 docs/s, 8.3 MB/s	—
NATIVE	3	42.2 docs/s, 13.9 MB/s	1.67×
NATIVE	6	51.1 docs/s, 16.8 MB/s	2.0×

Beyond 3 threads the single datanode is the bottleneck. A multi-datanode cluster scales further.

Crawler parallelism

The standalone crawler uploads each binary to the claim-check store and publishes its reference to Kafka. It can do this in parallel lanes on virtual threads (spring.opencrawling.crawler.concurrency, default 4):

• Ordering is preserved per document. A document always maps to the same lane (by a hash of its id).
• Connectors opt in via RepositoryConnector.supportsConcurrentProcessing() (default false). With lanes, the scan runs ahead and buffers documents. A connector should opt in only if a buffered document holds no open connection, cursor or large payload. Currently only the filesystem connector opts in.
• Processing is always sequential when a job writes directly to an OutputConnector (Admin UI/API jobs, embedded runtime), because output connectors don't have to be thread-safe.

Crawler lanes	Throughput	Speed-up
1	30.8 docs/s, 10.1 MB/s	—
4	67.3 docs/s, 22.2 MB/s	2.2×
8	88.9 docs/s, 29.3 MB/s	2.9×

With 4 or more lanes, the crawler publishes faster than a 3-thread writer drains, so the writer settings bound end-to-end throughput.

Large objects (S3 multipart)

• Objects below multipart-threshold use a single PUT. Larger ones are split into multipart-part-size parts, uploaded multipart-concurrency at a time. If a part fails, the other parts are cancelled and the upload is aborted, so no orphan parts are left.
• Streams of unknown length are buffered in memory up to 8 MB and spooled to a temp file beyond that, so memory stays bounded. (The crawler always passes -1.)
• Every request body is repeatable, as the AWS SDK checksums and retries require.
• The NATIVE writer is not affected.

Mode (256 MB file, S3G, single datanode)	Claim-check upload (crawler)	Target upload (writer)
Single PUT	1.6 s (164 MB/s)	2.2 s (117 MB/s)
Multipart, 16 MB parts × 4	2.3 s (109 MB/s)	3.1 s (82 MB/s)

On one disk, parallel parts compete with each other, so multipart is about 30% slower. That's why it's enabled only for very large objects (default 256MB). For those, it bounds each API call's duration (the claim-check client has a 30 s call timeout) and retries a single part instead of the whole object.

---

🧪 Testing

Suite	Script	What it checks
Unit	mvn -pl oc-ozone-output-connector test	Connector, writer, sidecar, and the RPC claim-check strategy
Migration mode	scripts/test-ozone-migration.sh	PipelineMode, orchestrator/consumer migration bypass and connector enforcement tests; verifies key storage if a live Ozone container is running
Migration, decoupled	scripts/test-ozone-migration-decoupled.sh	Two Ozone clusters (source claim check + target) with Kafka, crawler and writer
Decoupled RAG + Ozone claim check	scripts/test-ozone-decoupled.sh	See Apache Ozone Claim Check Store
Claim-check GC	scripts/test-ozone-claimcheck-gc.sh	TTL-based garbage collection

test-ozone-migration-decoupled.sh variables

Variable	Default	Description
OUTPUT_OZONE_CLIENT_TYPE	NATIVE	Writer transport into the target cluster: NATIVE or S3G.
CLAIM_CHECK_OZONE_CLIENT_TYPE	NATIVE	Claim-check transport on the source cluster: NATIVE or S3.
OC_KAFKA_PARTITIONS	3	Partitions of the documents topic.
OZONE_WRITER_CONCURRENCY	3	Writer consumer threads.
OC_CRAWLER_CONCURRENCY	4	Crawler lanes.
LARGE_FILE_MB	24	Step 5b crawls one random file of this size. It checks that the target object and the sidecar checksumSha256 match the source SHA-256. With S3 transports it also asserts that multipart uploads happened. 0 skips the step.
OC_MULTIPART_THRESHOLD / OC_MULTIPART_PART_SIZE / OC_MULTIPART_CONCURRENCY	8MB / 5MB / 4	The script lowers the compose defaults (256MB / 16MB / 4) so Step 5b exercises multipart. 0 disables it.
LOAD_DOCS	0	When > 0, generates N files and crawls them with the writer stopped, then times the writer draining the backlog. Prints CRAWLER_THROUGHPUT and THROUGHPUT.

The script always asserts that the crawler saved the claim check (Saved document content to Claim Check store: ofs:// or s3://).

# Default: NATIVE claim check -> NATIVE writer
scripts/test-ozone-migration-decoupled.sh

# S3 transports end to end
OUTPUT_OZONE_CLIENT_TYPE=S3G CLAIM_CHECK_OZONE_CLIENT_TYPE=S3 scripts/test-ozone-migration-decoupled.sh

# Throughput
OC_KAFKA_PARTITIONS=6 OZONE_WRITER_CONCURRENCY=6 LOAD_DOCS=400 scripts/test-ozone-migration-decoupled.sh

# Large file: single PUT vs multipart (S3 transports)
LARGE_FILE_MB=256 OC_MULTIPART_THRESHOLD=0 scripts/test-ozone-migration-decoupled.sh
LARGE_FILE_MB=256 OC_MULTIPART_THRESHOLD=16MB OC_MULTIPART_PART_SIZE=16MB scripts/test-ozone-migration-decoupled.sh

Large-file results (24 MB file):

Claim check → output	Claim-check upload	Writer migration
S3 → NATIVE	0.8 s, 29 MB/s	0.3 s, 73 MB/s
S3 → S3G	0.9 s, 26 MB/s	0.8 s, 29 MB/s
NATIVE → NATIVE	0.7 s, 34 MB/s	1.9 s, 13 MB/s (single sample)

CI

.github/workflows/integration-tests.yml runs the decoupled migration suite twice:

Matrix entry	OUTPUT_OZONE_CLIENT_TYPE	CLAIM_CHECK_OZONE_CLIENT_TYPE
Apache Ozone Migration Decoupled Pipeline Test Suite (NATIVE)	NATIVE	NATIVE
Apache Ozone Migration Decoupled Pipeline Test Suite (S3G)	S3G	S3