Components

• 1: Document
• 1.1: Overview
• 1.2: Document IDs
• 1.3: Document Model
• 2: Connector
• 3: Publisher
• 3.1: Publisher Accounting
• 4: Event
• 5: Stage
• 6: Pipeline
• 6.1: Pipeline Internals
• 7: Worker
• 8: Indexer
• 9: Runner
• 9.1: Runner Orchestration
• 10: Config

1 - Document

Search engines are designed to handle messy, incomplete, heterogeneous, loosely structured data. The basic unit of data in a search engine is typically called a “document” — it is simply a set of named fields, where each field may hold a single value or a list of values.

In Lucille, a Document is the basic unit of data that flows through a pipeline and gets indexed. Lucille’s Document aims to be as close to a search engine document as possible. The idea is that you don’t want to wait until the last minute to convert your data into a search-engine-friendly representation; you want to start with a search-engine-friendly representation the moment you acquire data from the source system, and use that representation throughout the entire enrichment pipeline. This means no intermediate object model, no mapping layer at the end — just fields in, fields out.

Why not POJOs?

If you’re coming from a background where encapsulation and strong typing are second nature, your first instinct might be to define POJOs for each entity type — a Product, a SupportTicket, a LegalDocument — with transformation logic behind well-named methods. That approach works when your domain has a small number of well-understood record types with stable schemas.

Search ingestion rarely looks like that. A typical project pulls data from multiple source systems, each with its own schema. The fields vary wildly from one record to the next — a database row has structured columns, a PDF has extracted text and metadata, a JSON API response has nested objects. Even within a single source, records are often inconsistent: optional fields that are sometimes present and sometimes not, multi-valued fields with unpredictable cardinality, fields whose meaning changes depending on the record type. Trying to capture all of this in a POJO hierarchy quickly becomes impractical — you end up with dozens of classes, most of which are just bags of optional fields, and the type system works against you rather than for you.

The pragmatic alternative is a generic map — something like Map<String, Object>. This gives you the flexibility to handle arbitrary fields, but at a cost: no typed access, no distinction between single-valued and multi-valued fields, verbose null-checking on every read, manual serialization logic at every boundary, and no built-in support for the update patterns (overwrite, append, skip) that search ingestion requires constantly.

Lucille’s Document is the middle ground. It has the flexibility of a map — any field name, any number of fields, no fixed schema — but with a purpose-built API that eliminates the boilerplate. Typed getters, uniform single/multi-valued access, update modes as first-class operations, and zero-cost JSON serialization because the document is already in the format that search engines expect. You get the adaptability of a schemaless representation without giving up the ergonomics of a well-designed API.

1.1 - Overview

A Document is an ordered, named set of fields. Each field may hold a single value or a list of values (multi-valued). All field values are ultimately represented in JSON. Every Document has a unique id.

Creating a Document

Use the static factory methods on the Document interface:

// Create with an explicit ID
Document doc = Document.create("my-doc-123");

// Create with an auto-generated UUID
Document doc = Document.create();

Connectors typically create Documents and call publisher.publish(doc) to send them into the pipeline.

Reserved Fields

Lucille reserves several field names for internal use. Do not use these for application data.

Field Types

Lucille Documents support the following value types:

• String
• Boolean
• Integer
• Double
• Float
• Long
• java.time.Instant
• byte[]
• com.fasterxml.jackson.databind.JsonNode
• java.sql.Timestamp
• java.util.Date

Reading Fields

Single-Valued Access

String  title  = doc.getString("title");
int     count  = doc.getInt("count");
double  score  = doc.getDouble("score");
float   weight = doc.getFloat("weight");
long    size   = doc.getLong("size");
boolean flag   = doc.getBoolean("active");
Instant ts     = doc.getInstant("created_at");
byte[]  raw    = doc.getBytes("content");
JsonNode json  = doc.getJson("metadata");

Multi-Valued Access

These methods also wrap a single value in a list if needed:

List<String>  titles = doc.getStringList("title");
List<Integer> counts = doc.getIntList("counts");
List<Double>  scores = doc.getDoubleList("scores");
List<Long>    sizes  = doc.getLongList("sizes");

Checking Field Existence

if (doc.has("title")) {
    String t = doc.getString("title");
}

Writing Fields

setField — Overwrite

Replaces any existing value(s) and makes the field single-valued:

doc.setField("title", "Hello World");
doc.setField("count", 42);
doc.setField("active", true);
doc.setField("score", 0.95);
doc.setField("created_at", Instant.now());

addToField — Append

Appends a value, converting the field to multi-valued if it was single-valued:

doc.addToField("tags", "search");
doc.addToField("tags", "etl");
// tags is now ["search", "etl"]

setOrAdd — Create or Append

Creates the field as single-valued if it does not exist; appends if it does:

doc.setOrAdd("tags", "search");
doc.setOrAdd("tags", "etl");

update — Controlled Write with UpdateMode

The update method accepts an UpdateMode enum that covers the three most common write patterns:

import com.kmwllc.lucille.core.UpdateMode;

// OVERWRITE: first value replaces all existing values; additional values are appended
doc.update("title", UpdateMode.OVERWRITE, "New Title");

// APPEND: all values are appended (field becomes or remains multi-valued)
doc.update("tags", UpdateMode.APPEND, "search", "etl");

// SKIP: field is left unchanged if it already has a value
doc.update("title", UpdateMode.SKIP, "Default Title");

Nested JSON

Documents support reading and writing values within nested JSON objects and arrays using dot-path notation (e.g., "metadata.author.name") or structured List<Document.Segment> paths.

Reading Nested Values

JsonNode node = doc.getNestedJson("metadata.author.name");

// Or using structured path segments
List<Document.Segment> path = Document.Segment.parse("metadata.items[2].title");
JsonNode node2 = doc.getNestedJson(path);

Writing Nested Values

ObjectMapper mapper = new ObjectMapper();
doc.setNestedJson("metadata.score", mapper.valueToTree(0.95));

Removing Nested Values

doc.removeNestedJson("metadata.tempField");

Path Segments

List<Document.Segment> segments = Document.Segment.parse("a.b[2].c");
String path = Document.Segment.stringify(segments); // "a.b[2].c"

Dropping and Skipping

Dropping removes the document from the pipeline entirely. It will not reach the Indexer.

Use the DropDocument Stage in config, or call setDropped() in stage code:

doc.setDropped(true);

Skipping causes the document to bypass all downstream Stages but still reach the Indexer. This is used for deletion markers, so the Indexer can issue a delete against the search backend.

Use the SkipDocument Stage in config, or call setSkipped() in stage code:

doc.setSkipped(true);

Child Documents

A Stage may generate child documents — additional Documents that flow through the remaining pipeline stages as independent records and are indexed alongside the parent. A Stage returns children from processDocument() as an Iterator<Document>.

Children are always emitted before the parent document, ensuring the Publisher’s accounting registers child IDs before it sees the parent’s completion event.

Iterating Fields

for (String fieldName : doc) {
    // iterate over all field names in the Document
}

Serialization

Documents serialize to and from JSON. In distributed mode, Documents flow between components as JSON bytes. The id and run_id are always included.

String json = doc.toString();

1.2 - Document IDs

Why IDs Must Be Deterministic

In search ingestion, the document ID serves as the primary key in the search index. When you index a document with ID “ABC”, the search engine either creates a new record or updates the existing record with that ID (upsert semantics). This has a critical implication:

If you run the same ingest twice, you want the same documents to get the same IDs. Otherwise, the second run creates duplicates instead of updating existing records. A search index with 100,000 documents would become 200,000 documents after a re-run — all duplicates.

Deterministic IDs mean:

• Re-ingestion is safe. Running the pipeline again updates existing documents rather than creating duplicates.
• Incremental updates work. A document that changes in the source system gets re-indexed under the same ID, replacing the old version.
• Deletions are possible. To delete a document from the index, you need to know its ID. If the ID was random, you’d have no way to reference it later.
• Parent-child relationships are stable. Child documents reference their parent by ID. If the parent’s ID changes on re-run, the relationship breaks.

This is why Connectors are responsible for creating documents with meaningful, stable IDs derived from the source data — a file path, a database primary key, a URL — rather than random UUIDs.

Why IDs Are Immutable in Lucille

Once a Document is created with an ID, that ID cannot be changed through the normal Document API. The id field is in the RESERVED_FIELDS set, so setField("id", ...), addToField("id", ...), and renameField("id", ...) all throw IllegalArgumentException.

This immutability exists because the ID is used as a tracking key throughout the system:

1. The Publisher registers the ID in its accounting ledger when the document is published. If the ID changed mid-pipeline, the Publisher would never receive a terminal event for the original ID — the run would hang.
2. The Worker sends CREATE events for child documents using their IDs. If a child’s ID changed after the CREATE event was sent, the Publisher’s accounting would be corrupted.
3. The Kafka message key is the document ID. Changing the ID mid-pipeline would break ordering guarantees (the document would land on a different partition after the change).
4. The Indexer uses the ID to send upserts and deletions to the search backend. The ID must be the same value that was tracked through the entire pipeline.

In short: the ID is the document’s identity across all components. Mutating it would break accounting, ordering, and idempotency simultaneously.

How Lucille Handles Duplicate IDs

Duplicate IDs — multiple documents with the same ID published in the same run — are a legitimate scenario. They arise in CDC (Change Data Capture) scenarios where a source system emits multiple updates for the same record, or when a connector reads from a source that contains duplicate entries.

Lucille handles duplicates at two levels:

At the Publisher Level: The Bag Data Structure

The Publisher’s docIdsToTrack is a Bag<String> (multiset), not a Set<String>. If two documents with ID “doc-1” are published, the Bag count for “doc-1” becomes 2. The Publisher expects to receive two separate terminal events for that ID. Each terminal event decrements the count by one. The run is not considered complete until the count reaches zero for all IDs.

// Two documents with same ID published → bag count is 2
docIdsToTrack.add("doc-1");  // count: 1
docIdsToTrack.add("doc-1");  // count: 2

// First terminal event → count drops to 1
docIdsToTrack.remove("doc-1", 1);  // count: 1

// Second terminal event → count drops to 0
docIdsToTrack.remove("doc-1", 1);  // count: 0, now removed

This means duplicate IDs don’t corrupt the accounting — each published document is tracked independently even if it shares an ID with another.

At the Publisher Level: Collapsing Mode

For connectors that emit multiple consecutive documents with the same ID (common in CDC), the Publisher supports a collapsing mode (requiresCollapsingPublisher() = true). In this mode, consecutive same-ID documents are merged into a single document via setOrAddAll() before being sent for processing:

if (previousDoc.getId().equals(document.getId())) {
    previousDoc.setOrAddAll(document);  // merge fields into one document
} else {
    sendForProcessing(previousDoc);     // different ID — send the previous one
    previousDoc = document;             // hold the new one
}

This reduces N consecutive same-ID documents to one document with multi-valued fields, which is then processed and indexed once. numReceived counts all N inputs; numPublished counts only the single merged output.

At the Indexer Level: Upsert Semantics

When two documents with the same ID reach the Indexer (either because collapsing is not enabled, or because they were non-consecutive), the search engine’s upsert semantics handle it: the second document overwrites the first in the index. The final state of the index reflects the last document indexed with that ID. Combined with Lucille’s ordering guarantees (same ID → same partition → same consumer → sequential processing), this means the final state is deterministic.

At the Indexer Level: Ordering Within a Batch

If a batch contains both an upsert and a delete for the same ID, the Indexer implementations (SolrIndexer, OpenSearchIndexer) explicitly handle ordering — flushing pending upserts before processing a delete for the same ID, or vice versa. This ensures that the operations are applied in the correct sequence even within a single batch.

The idOverrideField: Decoupling Internal ID from Index ID

Sometimes the ID used internally for tracking is not the ID you want in the search index. For example:

• A Connector might use a composite key (source + record number) for tracking uniqueness, but the search index expects a simpler ID.
• A pipeline might compute a better ID during enrichment (e.g., hashing certain fields to create a deduplication key).
• A document might need different IDs in different destination indices.

Lucille solves this with indexer.idOverrideField — a configuration option that tells the Indexer to use a different field’s value as the document’s ID when sending to the search backend, without modifying the document’s internal ID:

indexer {
  idOverrideField: "computed_id"
}

How It Works

1. The Connector creates the document with a stable internal ID (e.g., "source-record-42")
2. A pipeline Stage computes a better ID and stores it in a field (e.g., doc.setField("computed_id", "hash-abc123"))
3. The document flows through the system tracked by its internal ID ("source-record-42")
4. At indexing time, the Indexer calls getDocIdOverride(doc) which returns "hash-abc123"
5. The document is sent to the search backend with ID "hash-abc123"

What Uses Which ID

The internal ID ("source-record-42") is used for:

• Publisher accounting
• Kafka message key (ordering)
• Event tracking (CREATE, FINISH, FAIL)
• Logging and debugging

The override ID ("hash-abc123") is used only for:

• The document’s ID in the search index

This separation means the pipeline’s internal correctness guarantees (ordering, accounting, fault tolerance) are never affected by what ID appears in the search index. The override is applied at the very last moment — after all tracking is complete — as a pure presentation concern.

ID Generation Strategies: Good and Bad

How Existing Connectors Generate IDs

FileConnector: MD5 hash of the full file path.

String docId = DigestUtils.md5Hex(fullPath);
Document doc = Document.create(StorageClient.createDocId(docId, params));

The FileConnector uses the MD5 hash of the file’s full URI (e.g., s3://bucket/path/to/file.pdf) as the document ID. This is a good strategy because:

• It’s deterministic — the same file always produces the same ID
• It’s stable across re-runs — re-ingesting the same file updates rather than duplicates
• It handles special characters — file paths with spaces, unicode, or long lengths are reduced to a fixed-length hex string that’s safe for any search engine
• It’s unique — different file paths produce different hashes (collision probability is negligible)

The docIdPrefix is then prepended via StorageClient.createDocId(), producing IDs like "file-a1b2c3d4e5f6...".

DatabaseConnector: Value from a configured ID column.

String id = createDocId(rs.getString(idColumn));
Document doc = Document.create(id);

The DatabaseConnector reads the ID from a column specified in config (idField). This is the natural choice for database sources because:

• The database already has a primary key that uniquely identifies each record
• It’s deterministic and stable across re-runs
• It matches what the user expects — the search index ID corresponds to the database primary key

ChunkText: Parent ID + chunk number.

String id = parentId + "-" + (i + 1);
Document childDoc = Document.create(id);

Child documents derive their IDs from the parent ID plus a positional suffix. This ensures:

• Children have unique IDs (parent ID is unique, suffix is unique within the parent)
• The relationship to the parent is visible in the ID itself
• Re-chunking the same parent produces the same child IDs (deterministic)

Good ID Strategies

Bad ID Strategies

When Random UUIDs Are Acceptable

There is one scenario where random UUIDs are acceptable: when the index is always rebuilt from scratch (full re-index, not incremental). If you drop and recreate the index on every run, duplicate IDs from re-runs are not a concern because the old index is gone. However, this limits you to full batch mode and prevents incremental updates.

Hashing as an ID Strategy

Hashing (MD5, SHA-256) is useful when:

• The natural key is too long for a search engine ID field (some have length limits)
• The natural key contains characters that are problematic in URLs or APIs
• You want to deduplicate across sources (hash the content, not the source path)

The FileConnector’s use of DigestUtils.md5Hex(fullPath) is a good example. The tradeoff is that the ID is opaque — you can’t look at it and know which file it represents. The FileConnector mitigates this by also storing the full path in a file_path field on the document.

The docIdPrefix: Namespacing IDs by Connector

When multiple connectors feed documents into the same index, their IDs might collide. A database connector and a file connector might both produce a document with ID “1”. The docIdPrefix configuration on a connector prepends a string to all document IDs it creates:

connectors: [
  {
    name: "database"
    class: "com.kmwllc.lucille.connector.DatabaseConnector"
    docIdPrefix: "db-"
    # produces IDs like "db-1", "db-2", ...
  },
  {
    name: "files"
    class: "com.kmwllc.lucille.connector.FileConnector"
    docIdPrefix: "file-"
    # produces IDs like "file-/path/to/doc.pdf", ...
  }
]

The prefix is applied by the connector when creating documents via AbstractConnector.createDocId(id). It becomes part of the document’s immutable ID from that point forward.

Summary

1.3 - Document Model

Why the Document API Matters

At first glance, the idea of a “document” in a search ingestion system seems simple: it’s a bag of named fields with values. A Map<String, Object> would seem to suffice. In practice, a well-designed Document API turns out to be one of the most important practical considerations in a search ETL framework, for reasons that only become apparent when you’ve written dozens of pipeline stages and dealt with the realities of search engine field models.

Every pipeline stage reads fields, transforms them, and writes results back. A stage might be three lines of logic surrounded by ten lines of field access boilerplate — checking if a field exists, handling null, deciding whether to overwrite or append, converting types, dealing with single-valued vs. multi-valued fields. If the Document API is clumsy, that boilerplate dominates every stage you write. If the API is well-designed, stages are concise and the intent is clear.

Lucille’s Document API is the result of iterating on this problem across many real-world pipelines. Every method exists because a common pattern in search ingestion code demanded it.

Matching the Search Engine’s Field Model

Search engines (Solr, Elasticsearch, OpenSearch) have a field model that differs from a typical programming language map in important ways:

Single-valued vs. multi-valued fields. In a search engine schema, a field can hold one value or a list of values. A document might have a single title but multiple tags. This distinction matters for how the field is indexed, how it’s queried, and how it’s displayed. A Map<String, Object> does not capture this distinction — is the value a String or a List<String>? You’d need to check at every access point.

Lucille’s Document makes this explicit:

• getString("title") returns the single value (or the first value if multi-valued).
• getStringList("tags") returns all values as a list (wrapping a single value in a list if necessary).
• setField("title", "Hello") creates a single-valued field.
• addToField("tags", "search") converts the field to multi-valued if it wasn’t already.
• setOrAdd("tags", "etl") creates the field as single-valued if absent, or appends if present.

This mirrors exactly how search engines think about fields. A stage author doesn’t need to write if (value instanceof List) checks — the API handles the single/multi distinction uniformly.

The three update patterns. When a stage writes to a field, there are exactly three things it might want to do:

1. Overwrite whatever was there before.
2. Append to whatever was there (creating a multi-valued field).
3. Skip — write only if the field doesn’t already exist (don’t clobber earlier enrichment).

These three patterns appear so frequently in search ingestion code that Lucille provides them as a first-class UpdateMode enum, usable with the update() method:

doc.update("title", UpdateMode.OVERWRITE, "New Title");
doc.update("tags", UpdateMode.APPEND, "tag1", "tag2");
doc.update("summary", UpdateMode.SKIP, "Default summary");

The update() method also accepts varargs, so a stage can write multiple values in a single call. Without this, every stage would implement its own if/else logic for these three cases — and get it subtly wrong in edge cases (e.g., forgetting to convert a single-valued field to multi-valued before appending).

Why JSON Backing (ObjectNode) Is the Right Choice

Lucille’s Document is implemented as a thin wrapper around a Jackson ObjectNode. This is not an obvious choice — a HashMap<String, Object> would be simpler to implement. The JSON backing is motivated by the realities of how documents flow through the system.

Serialization Without Type Metadata

Documents cross boundaries constantly in Lucille: they’re placed on queues (in-memory or Kafka), sent to search backends via bulk APIs, logged for debugging, and captured in test mode for assertions. Every boundary crossing requires serialization.

With a JSON-backed document, serialization is trivial: data.toString() produces valid JSON. No conversion step, no schema, no type registry.

Critically, JSON’s type system eliminates the need to store explicit type information with each field. Jackson’s ObjectNode stores values as typed nodes — TextNode, IntNode, BooleanNode, ArrayNode, etc. When serialized to JSON, the types are implicit in the syntax:

• "title": "Hello" — string (quoted)
• "count": 42 — integer (unquoted number)
• "active": true — boolean (literal)
• "tags": ["a", "b"] — array (brackets)

When deserialized, Jackson reconstructs the correct node types from the JSON syntax. No type annotations, no class names in the serialized form, no versioning concerns. Compare this to Java serialization or a HashMap-based approach where you’d need to store type discriminators alongside values to reconstruct them correctly on the other end.

This matters operationally: documents on Kafka topics are human-readable JSON. An administrator can inspect them with standard Kafka tooling and understand what they contain without a decoder ring.

Zero-Cost Boundary Crossings

Because the document is JSON internally, there is no impedance mismatch at any boundary:

• From a JSON source (file, HTTP response, Kafka message): Parse the JSON into an ObjectNode and it becomes the Document’s backing store directly. No field-by-field copying.
• To a search engine: The bulk API for Elasticsearch, OpenSearch, and Solr all accept JSON. The document is already in the right format.
• To Kafka: Serialize the ObjectNode to a string. Done.
• From Kafka: Parse the string back to an ObjectNode. Done.
• In test assertions: document.toString() gives you the complete state as readable JSON.

A HashMap-backed document would require a serialization pass at every one of these boundaries — and in a system where a document crosses 4+ boundaries (source → processing queue → worker → indexing queue → indexer → search backend), that overhead is significant.

Native Nested Structure Support

Modern search ingestion often involves complex nested data: JSON responses from HTTP enrichment stages, structured extraction results from LLMs, nested document schemas in Elasticsearch. An ObjectNode naturally represents nested JSON (objects within objects, arrays of objects).

Lucille’s getNestedJson/setNestedJson API works directly on the tree structure:

// Read a nested value
JsonNode author = doc.getNestedJson("metadata.author.name");

// Set a nested value (creates intermediate objects as needed)
doc.setNestedJson("metadata.source.url", TextNode.valueOf("https://..."));

// Array indexing
JsonNode thirdTag = doc.getNestedJson("results[0].metadata.tags[2]");

With a HashMap, nested access would require ((Map) ((Map) map.get("metadata")).get("author")).get("name") — casting at every level, null-checking at every level, and no type safety. The Jackson tree API provides a typed, null-safe traversal.

Interop with the Jackson Ecosystem

Stages that call external APIs (HTTP enrichment, LLM calls, search engine queries) typically use HTTP clients that return Jackson JsonNode objects. These can be stored directly on the Document without conversion:

JsonNode apiResponse = httpClient.get(url);  // returns JsonNode
doc.setField("enrichment_result", apiResponse);  // stored directly

Similarly, Lucille’s JSONata transformation support operates on the Jackson tree natively — the Document’s backing ObjectNode is the input to the JSONata expression, and the result is written back as a JsonNode.

The Tradeoffs

The JSON backing is not free:

Per-field access overhead. Getting a String from an ObjectNode means node.get("field").asText() rather than (String) map.get("field"). The Document API hides this behind typed getters, but the implementation does more work per access than a HashMap. For stages that read many fields in a tight loop, this is measurably slower.

Memory overhead. Each field value is wrapped in a JsonNode subclass (TextNode, IntNode, etc.) rather than stored as a raw Java object. For documents with many small fields, the per-field wrapper overhead adds up.

No arbitrary Java types. A HashMap can store any Java object. An ObjectNode can only store JSON-representable types. Lucille works around this (byte arrays are stored as base64-encoded binary nodes, Instants are stored as ISO-8601 strings), but the Document cannot natively hold arbitrary domain objects.

Deep copy cost. Copying a document requires objectNode.deepCopy(), which recursively copies the entire JSON tree. A HashMap with immutable String values would be cheaper to shallow-copy.

In practice, these tradeoffs are acceptable because pipeline stages typically read a small number of fields, do expensive work (API calls, model inference, text processing), and write a small number of fields. The per-access overhead is negligible relative to the actual enrichment work. The serialization savings at every boundary crossing more than compensate.

Notable Design Choices in the API

Uniform Single/Multi-Valued Access

The getStringList() method returns a List<String> regardless of whether the field is single-valued or multi-valued. If the field is single-valued, it wraps the value in a singleton list. This means a stage that processes “all values of a field” doesn’t need to check whether the field is single or multi-valued first — it can always iterate over the list.

Conversely, getString() always returns the first value, whether the field is single or multi-valued. A stage that only cares about the primary value doesn’t need to handle the list case.

The setOrAdd Pattern

setOrAdd() is a single method that handles the most common field-writing pattern in search ingestion: “if this field doesn’t exist yet, create it; if it does, append to it.” Without this, every stage that accumulates values would need:

if (doc.has("tags")) {
    doc.addToField("tags", newTag);
} else {
    doc.setField("tags", newTag);
}

With setOrAdd, it’s one line: doc.setOrAdd("tags", newTag). Across a pipeline with dozens of stages, this eliminates hundreds of lines of boilerplate.

Typed Getters with Null Semantics

Every getter returns null for both “field absent” and “field present but null.” The has() method distinguishes between these cases when it matters. This is a deliberate choice: most stage logic doesn’t care why a value is missing — it just needs to handle the missing case. The rare stage that needs to distinguish “field not set” from “field explicitly null” can use has() + hasNonNull().

Reserved Fields with Triple-Underscore Prefix

Internal control fields (___dropped, ___skipped, ___children) use a triple-underscore prefix. This keeps them out of the normal field namespace — a triple-underscore prefix is highly unlikely to collide with any user-defined field name — and makes them visually distinct from user data fields. The validateFieldNames() method prevents stages from accidentally writing to reserved fields by checking membership in RESERVED_FIELDS.

Field Name Validation on Every Write

Every setter calls validateFieldNames() before writing. This catches two classes of bugs immediately:

1. Attempting to write to a reserved field (like id or run_id).
2. Passing a null or empty field name.

The validation happens at write time, not at indexing time, so bugs are caught in the stage that caused them rather than surfacing later in the pipeline.

Insertion Order Preservation

ObjectNode uses a LinkedHashMap internally, so getFieldNames() returns fields in insertion order. This is a subtle but useful property: when a document is serialized to JSON, the fields appear in the order they were added. This makes debugging easier (the ID is always first, enrichment fields appear in pipeline order) and produces deterministic output for testing.

JSONata Integration

The transform() method applies a JSONata expression directly to the Document’s backing ObjectNode. JSONata is a query and transformation language for JSON — think of it as XPath/XSLT for JSON. Because the Document is already JSON, there’s no conversion step. A stage can reshape a document’s structure with a single expression:

Jsonata expr = Jsonata.jsonata("{ 'fullName': firstName & ' ' & lastName }");
doc.transform(expr);

This is particularly powerful for stages that need to restructure complex nested data without writing procedural Java code.

The asMap() Escape Hatch

asMap() converts the Document to a Map<String, Object> using Jackson’s MAPPER.convertValue(). This is the escape hatch for code that needs a plain Map — typically when interfacing with libraries that expect Map input. It’s deliberately not the primary API because it loses the typed access, the single/multi-valued distinction, and the zero-cost serialization. But it exists for interop.

Summary

Lucille’s Document API is designed around three principles:

1. Match the search engine’s field model — single/multi-valued distinction, typed access, update modes that reflect how search fields are actually populated.
2. Minimize serialization cost — JSON backing means zero-cost boundary crossings in a system where documents cross many boundaries.
3. Eliminate stage boilerplate — setOrAdd, update with UpdateMode, uniform list access, and null handling reduce the per-stage code that isn’t core logic.

The result is that a typical pipeline stage is a few lines of domain logic rather than a page of field-access ceremony. Across a pipeline with dozens of stages, this compounds into significantly less code, fewer bugs, and faster development.

2 - Connector

What a Connector Does

A Connector is the component responsible for acquiring data from a source system and introducing it into Lucille as Documents. It is the entry point for all data in the system.

A Connector reads from its source — a database, a filesystem, a Kafka topic, an RSS feed, a search engine — and emits Documents one at a time by calling publisher.publish(doc). It does not know how many Workers will process those documents, how long enrichment will take, or where the documents will ultimately be indexed. Its only job is to produce Documents and hand them off.

Lifecycle

Every Connector goes through four lifecycle phases on each run:

1. preExecute(runId) — Called before execute. Use for setup: acquiring locks, creating temporary tables, validating source accessibility.
2. execute(publisher) — The main phase. Read from the source and call publisher.publish(doc) for each Document.
3. postExecute(runId) — Called only if execute succeeds. Use for cleanup: releasing locks, writing completion markers.
4. close() — Always called, even on failure. Use for releasing resources.

This lifecycle is enforced by the framework. The separation of preExecute from execute allows setup that should not be repeated on retry. The guarantee that close() is always called — regardless of whether execute or postExecute threw — ensures resources are never leaked.

Sequential Execution

When multiple Connectors are defined in a single run, they execute in sequence. Each Connector runs to completion — all its documents processed and indexed — before the next begins. This ordering guarantee is enforced automatically by the Publisher’s accounting system, without external orchestration.

This enables patterns like indexing parent documents before child documents that reference them by ID, or running a full ingest followed by a deletion pass.

Decoupling from Downstream

A Connector is fully decoupled from the rest of the system. It does not know:

• How many Worker threads or processes will consume its output
• What pipeline will be applied to its documents
• Which search backend the documents will reach
• Whether the system is running in local or distributed mode

This decoupling is what allows the same Connector implementation to work identically in all deployment modes. The Connector publishes to a queue; everything downstream is the framework’s concern.

Practical Guide

For how to configure connectors — common parameters, config syntax, and the full catalogue of built-in connectors — see Connectors in the Ingest Designer Guide.

For how to build a custom Connector, see Developing Connectors.

3 - Publisher

The Publisher is the internal accounting system that tracks every Document from the moment it is submitted to the pipeline until it reaches a terminal state (indexed, failed, or dropped).

What the Publisher Does

1. Accepts documents from a Connector via publisher.publish(doc).
2. Stamps the run ID on each document before it enters the pipeline.
3. Registers the document in its accounting ledger so it can track completion.
4. Buffers documents on the source queue for Workers to consume.
5. Receives events (FINISH, FAIL, DROP, CREATE) from Workers and the Indexer.
6. Determines run completion when all submitted documents have reached a terminal state.

Run Completion

The Publisher declares a run complete only when all three of the following are simultaneously true:

• The Connector thread has finished publishing all documents.
• All document IDs in the accounting ledger have been accounted for (each has a terminal event).
• The event queue is drained (no more events arriving).

This ensures that even out-of-order events and child documents generated mid-pipeline are correctly accounted for before the run is declared complete.

The internal accounting ledger is a Bag (multiset), not a Set. This means a Connector can legitimately publish two documents with the same ID in a single run — each is tracked independently and must individually reach a terminal event before the run completes.

Document Lifecycle Events

Backpressure

The Publisher implements backpressure to prevent a fast Connector from overwhelming the system:

• In local mode: publisher.queueCapacity bounds the in-memory source and destination queues. publish() blocks when the queue is full.
• In distributed mode: publisher.maxPendingDocs blocks publish() when too many documents are in flight (pending completion).

publisher {
  # Local mode: max docs in each queue (source and destination queues share this limit)
  queueCapacity: 10000

  # Distributed mode: block the Connector when this many docs are pending
  maxPendingDocs: 80000
}

Collapsing Mode

When a Connector emits multiple consecutive Documents with the same user-visible ID (e.g., a CDC stream with multiple updates to the same record), the Publisher can merge them into a single Document with multi-valued fields before passing them to the pipeline. This is enabled by setting requiresCollapsingPublisher() to true in the Connector implementation.

• numReceived counts every call to publisher.publish().
• numPublished counts only the documents actually sent downstream after collapsing.

Run Statistics

The Publisher tracks the following counts for each run:

These are reported in the run summary at completion:

connector1: complete. 200000 docs succeeded. 0 docs failed. 0 docs dropped.

Pause and Resume

The Publisher supports pausing and resuming document publication. publish() blocks when paused and wakes when resume() is called. This is used internally in some specialized deployment patterns.

Event Handling in Distributed Mode

In local mode, events flow through an in-memory queue. In distributed mode, events flow through a dedicated Kafka event topic. The topic name is derived from the run ID, ensuring isolation between concurrent runs.

See Events for more details.

3.1 - Publisher Accounting

Overview

The Publisher is Lucille’s bookkeeper. It tracks every document from the moment it enters the system until it reaches a terminal state (indexed, failed, or dropped). This accounting is what allows the Runner to know when a connector’s work is truly complete.

The core implementation lives in PublisherImpl, which maintains an in-memory ledger of pending documents. By design, the Publisher does not remember all documents it has ever published — only those currently in-flight. This keeps memory bounded regardless of how many documents flow through the system.

The Central Data Structure: docIdsToTrack

private final Bag<String> docIdsToTrack = SynchronizedBag.synchronizedBag(new HashBag<>());

This is the Publisher’s primary ledger — a synchronized Bag<String> from Apache Commons Collections. Every document ID that is currently “in-flight” (published but not yet terminal) lives here.

Why a Bag Instead of a Set

A Bag (multiset) allows duplicate entries. This matters because the same document ID can legitimately appear multiple times in a single run. If a connector publishes two documents with ID “doc-1”, the Publisher expects to receive two separate terminal events for that ID. With a Set, removing the ID after the first terminal event would leave the second document untracked. With a Bag, each remove call decrements the count by one:

// Two documents with same ID published → bag count is 2
docIdsToTrack.add("doc-1");  // count: 1
docIdsToTrack.add("doc-1");  // count: 2

// First terminal event → count drops to 1
docIdsToTrack.remove("doc-1", 1);  // count: 1

// Second terminal event → count drops to 0
docIdsToTrack.remove("doc-1", 1);  // count: 0, now removed

The SynchronizedBag wrapper ensures thread safety since publish() and handleEvent() run on different threads.

The Secondary Ledger: docIdsIndexedBeforeTracking

private final Bag<String> docIdsIndexedBeforeTracking = SynchronizedBag.synchronizedBag(new HashBag<>());

This handles a race condition with child documents. When a Worker creates a child document during pipeline processing, two things happen asynchronously:

1. A CREATE event is sent to the Publisher (so it starts tracking the child)
2. The child is processed and eventually reaches a terminal state (FINISH or FAIL)

These events can arrive out of order. If the terminal event arrives before the CREATE event, the Publisher can’t find the ID in docIdsToTrack. Rather than ignoring this, it records the ID in docIdsIndexedBeforeTracking. When the late CREATE event eventually arrives, the Publisher checks this secondary ledger first:

// In handleEvent(), when event.isCreate():
if (!docIdsIndexedBeforeTracking.remove(docId, 1)) {
    docIdsToTrack.add(docId);
}

If the ID is found in docIdsIndexedBeforeTracking, the Publisher knows the child already completed — no need to start tracking it.

The waitForCompletion Polling Loop

This is the method that blocks the main thread until all work is done:

public PublisherResult waitForCompletion(ConnectorThread thread, int timeout) throws Exception {
    while (true) {
        Event event = messenger.pollEvent();
        if (event != null) {
            handleEvent(event);
        }
        // Three termination conditions:
        if (!thread.isAlive() && !hasPending() && event == null) {
            return new PublisherResult(!thread.hasException(), null);
        }
    }
}

The loop terminates when all three conditions are met simultaneously:

1. Connector thread is dead (!thread.isAlive()) — no more documents will be published
2. No pending documents (!hasPending()) — every published document and child has reached a terminal state
3. Event queue is empty (event == null) — the previous poll returned nothing, meaning no more events are in transit

Condition 3 is critical. Even if conditions 1 and 2 are met, there might be events still in the queue that would change the pending count (e.g., a CREATE event for a child that hasn’t been accounted for yet).

The messenger.pollEvent() call is a blocking operation with a timeout (typically 50ms for local, 2000ms for Kafka), preventing a busy-wait while still checking termination conditions periodically.

Thread Interaction: handleEvent() vs publish()

The Publisher is designed for concurrent access from two threads:

• Connector thread calls publish() — adds IDs to docIdsToTrack
• Main thread (in waitForCompletion) calls handleEvent() — removes IDs from docIdsToTrack

Both methods mutate docIdsToTrack, which is why it must be a SynchronizedBag. The publish() method can also be called from multiple connector threads simultaneously (except in collapsing mode).

The maxPendingDocs Backpressure Mechanism

When configured, this prevents the connector from overwhelming downstream components:

private final ReentrantLock lockForPendingDocs = new ReentrantLock();
private final Condition pendingDocsBelowMaxCondition = lockForPendingDocs.newCondition();

In publish(), if the pending count exceeds the threshold, the calling thread blocks:

if (maxPendingDocs != null) {
    lockForPendingDocs.lock();
    while (docIdsToTrack.size() >= maxPendingDocs) {
        pendingDocsBelowMaxCondition.await(10, TimeUnit.SECONDS);
    }
    lockForPendingDocs.unlock();
}

In handleEvent(), when a terminal event reduces the pending count below the max, blocked threads are signaled:

if (docIdsToTrack.size() < maxPendingDocs) {
    pendingDocsBelowMaxCondition.signalAll();
}

The 10-second timeout on await() is a safety net — if a signal is somehow missed, the thread will re-check the condition periodically.

Important concurrency note: If N threads are blocked on publish() and the pending count drops to maxPendingDocs - 1, all N threads are signaled simultaneously. Each may then publish a document, causing the actual pending count to temporarily exceed maxPendingDocs by up to N-1. This is acceptable because each thread will block again on its next publish() call.

Collapsing Mode

When isCollapsing == true, consecutive documents with the same ID are merged into one:

private void publishInternal(Document document) throws Exception {
    if (!isCollapsing) {
        sendForProcessing(document);
        return;
    }
    if (previousDoc == null) {
        previousDoc = document;
        return;
    }
    if (previousDoc.getId().equals(document.getId())) {
        previousDoc.setOrAddAll(document);  // merge fields
    } else {
        sendForProcessing(previousDoc);
        previousDoc = document;
    }
}

The Publisher holds onto the previous document. If the next document has the same ID, fields are merged. If the ID differs, the previous document is finally sent for processing. The flush() method handles the last held document.

Thread safety caveat: Collapsing mode is NOT thread-safe for multiple publishing threads because previousDoc is shared mutable state without synchronization.

numPublished vs numReceived

• numReceived — incremented every time publish() completes (counts inputs)
• numPublished — incremented every time sendForProcessing() is called (counts outputs)

In non-collapsing mode, these are equal. In collapsing mode, numPublished <= numReceived because multiple inputs may collapse into one output.

Registration Ordering

A critical invariant: the document ID is added to docIdsToTrack before the document is placed on the processing queue:

private void sendForProcessing(Document document) throws Exception {
    document.initializeRunId(runId);
    String docId = document.getId();

    // Track FIRST
    docIdsToTrack.add(docId);

    try {
        // Send SECOND
        messenger.sendForProcessing(document);
    } catch (Exception e) {
        // Rollback tracking if send fails
        docIdsToTrack.remove(docId, 1);
        throw e;
    }
    numPublished.incrementAndGet();
}

If the order were reversed (send first, then track), a fast Worker could process the document and emit a terminal event before the Publisher starts tracking it. The event would then be misclassified as “early” and placed in docIdsIndexedBeforeTracking, corrupting the accounting.

Pause/Resume Mechanism

The Publisher supports pausing all publishing threads:

private final ReentrantLock lockForPauseResume = new ReentrantLock();
private volatile Condition resumeCondition = null;

pause() creates a Condition object. Any thread calling publish() checks for this condition and blocks if it’s set:

if (resumeCondition != null) {
    lockForPauseResume.lock();
    if (resumeCondition != null) {  // double-check after acquiring lock
        while (resumeCondition != null) {
            resumeCondition.await();  // loop handles spurious wakeups
        }
    }
    lockForPauseResume.unlock();
}

resume() signals all waiting threads and nulls out the condition. The double-checked locking pattern (check volatile field, then acquire lock and re-check) avoids lock contention in the common case where the Publisher is not paused.

Thread Safety Summary

4 - Event

Lucille Events

As a Document passes through the Lucille pipeline, Event messages are generated at key transitions. The Publisher consumes these events to track the lifecycle of every document in the run and determine when all work is complete.

Event Types

Events include the document’s id and run_id, allowing the Publisher to match each event to its corresponding accounting entry.

Event Flow

Worker → [event queue] → Publisher
Indexer → [event queue] → Publisher

In local mode, events flow through an in-memory queue on the main thread’s polling loop.

In distributed mode, events flow through a dedicated Kafka event topic. The topic name is derived from the run ID, ensuring that events from different concurrent runs are always isolated.

Event Topics (Kafka)

In distributed mode, each run creates its own event topic named based on the run ID and pipeline name. This ensures that the Publisher for a given run only sees events for its own documents. Because Workers and Indexers are long-running processes that serve multiple runs over their lifetime, a document’s run_id is the mechanism that routes its events to the correct Publisher — enabling multiple concurrent Runner invocations to share the same Worker and Indexer pool without their accounting interfering with each other. See Long-Running Workers and Indexers for the full operational pattern.

When kafka.events is set to false in the config, event messages are not sent to Kafka. This is appropriate only in streaming mode (no Runner) where run completion tracking is not needed.

kafka {
  events: true  # default; set to false only in pure streaming mode
}

Connector-less (Streaming) Mode

In connector-less distributed mode, a third-party publisher writes Documents directly to a Kafka source topic. There is no Lucille Runner or Publisher. In this case:

• If kafka.events is true, the third-party publisher must include a run_id on each document (it can choose its own run ID value).
• Workers and the Indexer send events to the Kafka event topic as usual.
• Since there is no Publisher polling the event topic, events accumulate in the topic and are not consumed (unless you route them to your own consumer).

If run tracking is not needed in streaming mode, set kafka.events: false to suppress event production entirely.

Child Documents

Child documents generated by Stages must be registered with the Publisher before the parent document reaches the Indexer. The Worker sends a CREATE event for each child as soon as it is emitted, before the parent’s pipeline execution completes. This ordering guarantee ensures the Publisher never declares a run complete while child documents are still in flight.

Out-of-Order Events

The Publisher handles out-of-order events correctly. A child document can complete (receive a FINISH event from the Indexer) before the Publisher has even received the child’s CREATE event (because Workers and Indexers run concurrently). In this case, the Publisher stores the premature FINISH in a secondary buffer and reconciles it when the CREATE event subsequently arrives.

5 - Stage

What a Stage Does

A Stage is the fundamental unit of document transformation in Lucille. Each Stage performs a single, focused operation on a Document: extracting text, renaming fields, generating embeddings, looking up data from an external system, or any other enrichment task.

Stages are composed into Pipelines. When a Document flows through a Pipeline, it passes through each Stage in sequence. Each Stage receives the Document as mutated by all previous Stages and can read fields, write fields, or emit child documents.

The Stage Contract

A Stage implementation must provide one method: processDocument(Document doc). This method receives a Document, performs its transformation (typically by reading and writing fields), and returns an iterator of result documents. For most stages, the iterator contains just the input document (now modified). Stages that generate child documents return the children followed by the parent.

The framework handles everything else:

• Instantiation — Stages are created from class names in configuration via reflection.
• Lifecycle — start() is called once before processing begins (for resource acquisition); stop() is called once after processing ends (for cleanup).
• Condition evaluation — The framework checks conditions before calling processDocument(). If conditions are not met, the Stage is skipped entirely.
• Thread isolation — Each Worker thread gets its own Stage instance. No synchronization is needed.
• Error handling — If processDocument() throws, the framework catches the exception, marks the document as failed, and continues processing other documents.

Conditions as a Design Decision

Rather than supporting sub-pipelines or branching, Lucille provides per-stage conditions that control whether a Stage applies to a given Document. This keeps the pipeline linear while allowing different processing for different document types.

Conditions are evaluated by the framework before invoking the Stage. A Stage author never implements conditional logic — they write a Stage that does one thing, and the configuration determines which documents it applies to. This separation means Stages are simpler to write, simpler to test, and reusable across pipelines with different condition configurations.

Disabling Stages

As a convenience, you can set enabled: false on any Stage in your Config. By doing so, the Stage is not instantiated, start() and stop() are never called, and no Document is processed by that Stage. The Stage will still be validated against its Spec, warning you of any missing, invalid, or unknown properties in the Config.

Child Document Emission

A Stage can produce additional documents — children — that flow through the remaining pipeline stages independently. This is how Lucille handles 1-to-N fan-out (e.g., chunking a document into embedding-sized pieces). Children are tracked by the Publisher’s accounting system and indexed as independent records.

The iterator-based return type (Iterator<Document>) means children are produced lazily. Memory usage is bounded regardless of how many children a Stage generates.

Practical Guide

For how to configure stages — syntax, conditions, conditionPolicy, and the full catalogue of built-in stages — see Stages in the Ingest Designer Guide.

For how to build a custom Stage, see Developing Stages.

6 - Pipeline

A Pipeline is an ordered sequence of processing Stages. When a Connector publishes a Document, that Document is picked up by a Worker and passed through every Stage in the configured Pipeline before being sent to the Indexer.

Linear Execution Model

Stages execute in the order they are listed. Each Stage receives the Document as mutated by all previous Stages. If a Stage generates child documents, those children flow through the remaining stages of the same pipeline independently.

This is a deliberate architectural choice: a pipeline is a linear sequence, not an arbitrary graph. There are no branches, no sub-pipelines, and no conditional routing to different pipeline paths.

Why No Sub-Pipelines

Experience has shown that sub-pipelines and branching graphs introduce significant cognitive and testing complexity. They make pipelines harder to reason about and harder to troubleshoot — before you can diagnose a problem, you first have to determine which route a document took. In most real-world ingestion scenarios, sub-pipelines do not turn out to be necessary.

Lucille provides three mechanisms that cover the cases where branching might seem attractive:

Conditions. Every stage supports a conditions block in configuration that determines whether that stage should process a given document. Conditions can check field presence, field values, or combinations (with all/any policy). The framework evaluates conditions before invoking the stage — stage authors never implement conditional logic themselves. This allows a single linear pipeline to apply different processing to different document types without branching.

Custom stages for complex logic. If you need decision logic more complex than what conditions can express, the pathway is to write a custom stage where that logic is implemented and tested in Java — or in Python using Lucille’s EmbeddedPython or ExternalPython stages. A custom stage can inspect any aspect of the document and take arbitrary action, including setting fields that downstream conditions can check.

Config includes for reuse. If the concern is reusing common sequences of stages across multiple pipelines, those sequences can be defined in a separate config file and composed into a larger pipeline definition using HOCON’s array concatenation. The pipeline remains a single linear sequence at execution time — the composition happens at config resolution time, not at runtime.

Per-Thread Isolation

When multiple Worker threads are active, each thread gets its own Pipeline instance — and its own instance of every Stage. This means:

• Stages can safely hold stateful resources (database connections, loaded models, compiled patterns) without synchronization.
• Expensive setup (model loading, connection pool creation) happens once per Worker thread at startup.
• Per-thread isolation eliminates a whole class of concurrency bugs in Stage implementations.

This design means that pipeline authors write sequential code — read a field, transform it, write a field — and the framework handles parallelism. The complexity of concurrent execution is the framework’s responsibility, not the user’s.

Multiple Pipelines

Multiple pipelines can be defined in a single run, each serving different connectors. All pipelines feed the same Indexer. This allows a single Lucille invocation to ingest from multiple sources with different enrichment logic, all writing to the same search backend.

Practical Guide

For how to define pipelines in configuration — syntax, connecting connectors, conditions, reuse patterns, and examples — see Defining Pipelines in the Ingest Designer Guide.

6.1 - Pipeline Internals

What a Pipeline Is

A Pipeline is an ordered sequence of Stages. When a Document enters the Pipeline, it flows through each Stage in order. Each Stage modifies the Document in place and may optionally generate child Documents. The Pipeline returns an Iterator over all Documents that emerged from the processing — the original (possibly modified) Document plus any children generated along the way.

The key insight is that the Pipeline does not eagerly process everything and return a collection. It returns a lazy Iterator that processes Documents on demand as next() is called. This has profound implications for memory usage, especially when stages generate children.

How Pipeline.processDocument() Works

The implementation is deceptively simple:

public Iterator<Document> processDocument(Document document) throws StageException {
    Iterator<Document> result = document.iterator();

    for (Stage stage : stages) {
        result = stage.apply(result);
    }

    return result;
}

This builds a chain of lazy iterators — one per stage — and returns the outermost one. No processing has happened yet. The stages are not called until next() is called on the returned Iterator.

What happens when next() is called

Consider a pipeline with stages S1, S2, S3. When the Worker calls next() on the returned Iterator:

1. The outermost iterator (S3’s wrapper) calls next() on its input iterator (S2’s wrapper).
2. S2’s wrapper calls next() on its input iterator (S1’s wrapper).
3. S1’s wrapper calls next() on its input iterator (the original document’s singleton iterator).
4. The original document is returned.
5. S1’s wrapper calls apply(document) — which calls processConditional(document) — which calls S1’s processDocument(document). S1 modifies the document in place and returns an iterator of children (or null).
6. S1’s apply(document) returns an IteratorChain(children, parent) — children first, then the parent.
7. S2’s wrapper receives the first element from S1’s output (a child, if any, or the parent) and calls apply() on it.
8. This continues up the chain until S3 produces its first output document.

The critical property: each document is processed through the full pipeline one at a time. The pipeline doesn’t process all documents through S1, then all through S2, then all through S3. It processes one document through all stages before starting the next.

Children flow through downstream stages only

If S2 generates a child document, that child is passed through S3 but NOT back through S1. This is because the iterator chain is built left-to-right: S1’s output feeds S2, S2’s output feeds S3. A child generated by S2 enters the chain at S2’s output position and only flows forward.

This is the correct semantic for search ingestion: if S1 extracts text from a PDF and S2 chunks that text into pieces, the chunks should flow through S3 (which might generate embeddings) but should NOT flow back through S1 (which would try to extract text from them again).

Depth-first traversal order

The iterator chain traverses children depth-first. When a stage produces children, the first child flows through all downstream stages — and if those stages produce their own children, the first grandchild flows through all stages below it — before the second child at any level is even requested from its producing stage. This means that at any point during pipeline execution, only one document per level of the chain is actively being processed. Children do not accumulate in memory waiting for downstream processing — each child is fully processed and handed off to the Worker before the next sibling is pulled from the iterator.

This property holds as long as child-generating stages return lazy iterators. A stage that materializes all children into a list before returning will hold those children in memory at that level, but they still flow through downstream stages one at a time. For maximum memory efficiency in stages that produce many children (e.g., text chunking), return a lazy iterator that generates children on demand rather than building a complete list.

Why Iterators Instead of Lists

The memory problem with lists

Consider what would happen if processDocument() returned a List<Document> instead of an Iterator<Document>:

// HYPOTHETICAL: if stages returned lists
List<Document> processDocument(Document doc) {
    List<Document> results = List.of(doc);
    for (Stage stage : stages) {
        List<Document> nextResults = new ArrayList<>();
        for (Document d : results) {
            nextResults.addAll(stage.process(d));  // returns list of doc + children
        }
        results = nextResults;
    }
    return results;
}

This approach has a critical flaw: all documents must be held in memory simultaneously.

Consider a pipeline that chunks a large document into 1,000 pieces (S1), then generates an embedding for each piece (S2), then formats each for indexing (S3):

• After S1: 1,000 documents in memory
• After S2: still 1,000 documents in memory (each now with an embedding vector)
• After S3: still 1,000 documents in memory

With the list approach, all 1,000 chunks exist in memory at once. With the iterator approach, only one chunk at a time flows through S2 and S3. The previous chunk has already been handed off to the Worker (which sends it to the indexing queue) before the next chunk is generated.

Compounding with multiple child-generating stages

The problem compounds when multiple stages generate children. Consider:

• S1: Extracts 10 files from a zip archive (10 children)
• S2: Extracts text from each file, producing 5 pages each (50 children)
• S3: Chunks each page into 20 pieces (1,000 children)

With lists: after S3, you’d have 1,000 documents in memory simultaneously.

With iterators: at any given moment, you have at most one document at each level of the chain being processed. The zip is opened lazily, files are extracted one at a time, pages are produced one at a time, chunks are produced one at a time. Memory usage is proportional to the depth of the pipeline, not the breadth of the output.

The lazy evaluation model

The iterator chain implements pull-based lazy evaluation. Nothing happens until the consumer (the Worker) calls next(). Each next() call pulls exactly one document through the full pipeline. This means:

• Documents are produced one at a time
• Each document is fully processed (through all stages) before the next begins
• Memory usage is bounded by the pipeline depth, not the number of children
• The Worker can send each document to the indexing queue immediately after receiving it, freeing memory

In-Place Modification vs. Copying

Lucille stages modify Documents in place rather than creating new copies. This is a deliberate design choice with significant implications.

How it works

@Override
public Iterator<Document> processDocument(Document doc) throws StageException {
    // modifies doc directly — no copy is made
    String value = doc.getString("input");
    doc.setField("output", value.toUpperCase());
    return null;
}

The same Document object flows through all stages. S1 modifies it, then S2 sees the modifications S1 made, then S3 sees the modifications both S1 and S2 made.

Why in-place modification is the right choice for search ETL

Memory efficiency. A search ingestion document can be large — it might contain extracted text from a PDF (megabytes), binary content, or dozens of fields. Copying the entire document at each stage would multiply memory usage by the number of stages. With 20 stages and a 5MB document, that’s 100MB per document in flight vs. 5MB.

Simplicity for stage authors. The mental model is straightforward: “I receive a document, I modify it, I’m done.” There’s no need to construct a new Document, copy all existing fields, add new fields, and return it. The boilerplate savings are significant across dozens of stages.

Natural accumulation. In search ingestion, stages typically add information to a document: S1 extracts text, S2 detects language, S3 extracts entities, S4 generates embeddings. Each stage enriches the document with new fields. The in-place model makes this accumulation natural — each stage sees everything previous stages added.

Performance. No allocation, no copying, no garbage collection pressure from intermediate Document objects. For pipelines processing millions of documents, this matters.

The tradeoff: no rollback on failure

If S3 throws an exception after S1 and S2 have already modified the document, the document is in a partially-enriched state. There’s no way to “undo” the modifications from S1 and S2. In Lucille, this is acceptable because:

1. A failed document is routed to a failure state — it’s not indexed in its partial form.
2. Search ingestion is generally idempotent — if the document is retried, it will be processed from scratch (a fresh copy from the source).
3. The alternative (copying at each stage for rollback capability) would impose a performance cost on every document for a scenario that affects a tiny minority.

The tradeoff: stage ordering matters

Because stages see each other’s modifications, the order of stages in the pipeline is significant. S2 can depend on fields that S1 created. This is a feature, not a bug — it’s how enrichment pipelines naturally work (extract text before you can detect language, detect language before you can apply language-specific NER). But it means reordering stages can change behavior, and a stage that expects a field to exist will fail if the stage that creates it is moved downstream.

When copies are necessary

The one case where copies are needed is child document creation. When a stage generates a child, it creates a new Document object (via Document.create(childId)). The child is a separate object from the parent — modifications to the child don’t affect the parent and vice versa. This is correct because children are independent documents that will be indexed separately.

The Stage.apply() Contract

Each Stage has two apply methods that implement the iterator chaining:

apply(Document doc) — process one document

public Iterator<Document> apply(Document doc) throws StageException {
    Iterator<Document> children = processConditional(doc);
    Iterator<Document> parent = doc.iterator();

    if (children == null) {
        return parent;  // no children — just the (modified) parent
    }

    // wrap children to copy run ID and count metrics
    Iterator<Document> wrappedChildren = new Iterator<>() { ... };

    return new IteratorChain(wrappedChildren, parent);  // children first, then parent
}

Key details:

• Children come before the parent in the returned iterator. This ensures the Worker sends CREATE events for children before the parent completes, so the Publisher knows about children before it might declare the parent done.
• The run ID is copied from parent to child automatically.
• Child count metrics are incremented as children are produced (lazily, as next() is called).

apply(Iterator<Document> docs) — wrap an iterator

public Iterator<Document> apply(Iterator<Document> docs) throws StageException {
    return new Iterator<>() {
        Iterator<Document> current = null;

        public boolean hasNext() {
            return (current != null && current.hasNext()) || docs.hasNext();
        }

        public Document next() {
            if (current != null && current.hasNext()) {
                return current.next();
            }
            Document d = docs.next();
            current = apply(d);  // process this document, get iterator of children + parent
            return current.next();  // return first element
        }
    };
}

This wraps an input iterator so that each document pulled from it is processed through the stage. If a document produces children, they are returned before the next document from the input iterator is pulled. This is how the “children flow through downstream stages” behavior works — a child produced by S2 enters S3’s input iterator and is processed by S3 before the next document from S2’s output.

What the Pipeline Framework Gives You

As a stage author, you don’t think about:

• Iterator mechanics. You implement processDocument(Document doc) and return null or an iterator of children. The framework handles the chaining.
• Conditional execution. The framework evaluates conditions before calling your method. If conditions don’t match, your code is never called.
• Dropped/skipped documents. The framework checks these flags before calling your method.
• Metrics. Processing time, error count, and child count are tracked automatically.
• Logging. Stage entry/exit is logged per document automatically.
• Child document lifecycle. Run ID copying, CREATE event sending, and downstream routing are handled by the framework.
• Thread safety. Each Worker thread has its own Pipeline instance with its own Stage instances.
• Memory management. The lazy iterator model ensures bounded memory usage regardless of how many children are generated.

As a pipeline designer, you get:

• Composability. Stages are independent units that can be reordered, added, or removed without modifying other stages.
• Conditional execution via config. Skip stages for certain documents without writing code.
• Heterogeneous document handling. A single pipeline can process different document types differently using conditions.
• Predictable ordering. Stages execute in config order. Children flow through downstream stages only.
• Bounded memory. Even pipelines that generate millions of children from a single input document operate in bounded memory.

Compared to writing your own processing loop:

A naive processing loop (for each stage: stage.process(doc)) would need to handle:

• What if a stage produces children? Do you process them through remaining stages?
• What if children produce grandchildren?
• How do you avoid holding all children in memory?
• How do you ensure children are emitted before parents for accounting?
• How do you handle conditional execution?
• How do you handle dropped/skipped documents?
• How do you track metrics per stage?
• How do you handle errors in one stage without affecting others?

Lucille’s Pipeline handles all of this in ~50 lines of iterator-chaining code that stage authors never see. The stage author’s world is simple: receive a document, modify it, optionally return children. Everything else is the framework’s problem.

Summary

The Pipeline’s design rests on three key decisions:

1. Lazy iterators instead of eager lists. Documents are produced one at a time, bounding memory usage regardless of how many children are generated. Processing is pull-based — nothing happens until the consumer asks for the next document.

2. In-place modification instead of copying. Stages enrich a document by adding fields to it directly. No allocation overhead, no copy overhead, natural accumulation of enrichment across stages. The tradeoff (no rollback on failure) is acceptable because failed documents are discarded, not indexed in a partial state.

3. Children before parents in the output order. This ensures the accounting system learns about children before it might consider the parent complete, preventing premature run-completion detection.

Together, these decisions produce a pipeline framework that is memory-efficient, simple for stage authors, and correct for the accounting system — without requiring stage authors to understand any of the underlying mechanics.

7 - Worker

A Worker is a thread that pulls Documents from the source queue, runs them through a Pipeline of Stages, and pushes the processed results onto the destination queue for the Indexer to consume.

What the Worker Does

When a Worker starts, it:

1. Constructs its own instance of the configured Pipeline (including a private instance of every Stage).
2. Enters a polling loop, pulling Documents from the source queue one at a time.
3. Passes each Document through every Stage in the Pipeline in order.
4. Pushes the processed Document (and any child documents) to the destination queue.
5. Sends lifecycle events (FINISH, FAIL, DROP, CREATE) to the Publisher via the event queue.

Per-Thread Pipeline Isolation

Each Worker thread has its own isolated Pipeline instance. This is a deliberate design choice:

• Stages can hold stateful resources (database connections, loaded ML models, compiled regexes) initialized once in start() and reused across all documents that thread processes — no synchronization needed.
• A large NLP model loads once per Worker thread at startup and lives for the thread’s lifetime.
• The memory cost of N model instances is the price of N-way parallelism without lock complexity.

Multiple Workers

In local mode, you can run multiple Worker threads within a single JVM:

worker {
  threads: 4
}

Each thread runs its own Pipeline instance concurrently.

In distributed mode, you start multiple Worker processes. Each process consumes from the same Kafka source topic, and Kafka’s consumer group protocol distributes work across them automatically.

Configuration

worker {
  # Number of worker threads to start in local mode (default: 1)
  threads: 2

  # Maximum time (seconds) between Kafka polls before the worker shuts down
  # (only relevant in Kafka mode; requires exitOnTimeout: true)
  # Must be greater than Lucille's internal poll timeout (50ms local, 2s distributed),
  # otherwise an idle worker can be incorrectly flagged as stuck.
  maxProcessingSecs: 600

  # Shut down if no message is polled within maxProcessingSecs
  exitOnTimeout: true

  # Maximum number of processing attempts for any document across all workers.
  # Requires zookeeper.connectString to be configured.
  # Documents exceeding this limit are routed to a dead-letter queue and
  # do not block the rest of the run. Omit to disable retry tracking entirely.
  maxRetries: 3

  # Write a heartbeat.log file periodically for liveness checks.
  # Frequency is controlled by log.seconds.
  enableHeartbeat: true
}

# Required when worker.maxRetries is set
zookeeper {
  connectString: "localhost:2181"
}

# Controls how often Workers, Publishers, and Indexers log status updates and heartbeats
log {
  seconds: 30
}

Error Handling

Per-Document Failures

If a Stage throws an exception while processing a document, the Worker:

1. Logs the failure (including document ID and run ID in the MDC).
2. Sends a FAIL event to the Publisher.
3. Continues processing the next document.

The run does not stop on a per-document failure. Individual document failures are counted and reported in the run summary.

Poison Pills

A “poison pill” is a document that repeatedly causes the Worker process itself to crash. If worker.maxRetries is configured (requires ZooKeeper), the retry counter tracks crash counts across all Worker instances. When a document exceeds the retry limit, it is routed to a dead-letter queue, and the rest of the ingest continues.

Metrics

Each Worker reports Codahale metrics to the shared registry:

• Document processing time: Mean latency per document through the full pipeline.
• Error counts: Number of documents that caused exceptions.

The WorkerPool logs a periodic status update every log.seconds seconds:

INFO WorkerPool: 27017 docs processed. One minute rate: 1787.10 docs/sec. Mean pipeline latency: 10.63 ms/doc.

Lifecycle Events

The Publisher’s accounting system uses these events to determine when a run is complete.

Running a Worker Standalone

In distributed mode, start a Worker as a separate process:

java \
  -Dconfig.file=<PATH/TO/YOUR/CONFIG.conf> \
  -cp 'lucille-core/target/lucille.jar:lucille-core/target/lib/*' \
  com.kmwllc.lucille.core.Worker \
  <pipeline-name>

The pipeline name argument tells the Worker which pipeline to run and which Kafka source topic to consume from.

WorkerIndexer

WorkerIndexer is a hybrid entry point that pairs one Worker thread with one Indexer in a single JVM. It is useful in distributed deployments where you want a single process to handle both pipeline processing and indexing, without the overhead of coordinating separate Worker and Indexer processes.

How it differs from a standalone Worker:

• Consumes documents from Kafka (source topic) — same as a standalone Worker.
• Routes processed documents to an in-memory queue rather than back to Kafka.
• The co-located Indexer reads from that in-memory queue and sends to the search backend.
• Eliminates the Kafka hop between processing and indexing, reducing latency.

Start a WorkerIndexer:

java \
  -Dconfig.file=<PATH/TO/YOUR/CONFIG.conf> \
  -cp 'lucille-core/target/lucille.jar:lucille-core/target/lib/*' \
  com.kmwllc.lucille.core.WorkerIndexer \
  <pipeline-name>

Internally, WorkerIndexer creates a WorkerIndexerPool that manages multiple Worker+Indexer thread pairs within the same JVM. The worker.threads config controls how many pairs run (default: 1):

worker {
  threads: 4   # 4 Worker+Indexer thread pairs in this JVM
}

In a multi-node deployment, you can run multiple WorkerIndexer processes consuming from the same Kafka topic. Kafka’s consumer group protocol distributes source documents across them automatically. WorkerIndexer is particularly useful in streaming mode where no Runner is coordinating the run.

Practical Guide

For deployment instructions — starting Workers and WorkerIndexers in local and distributed mode, scaling, and operational considerations — see Deployment.

For Worker-related configuration parameters, see Writing a Config.

8 - Indexer

What an Indexer Does

An Indexer is the component responsible for delivering processed Documents to their final destination — typically a search engine or vector database. It is the last component in the data flow: Connectors produce Documents, Workers enrich them, and the Indexer sends them to the search backend.

Batching

Indexers do not send documents one at a time. They accumulate documents into batches and flush them as a single bulk API call. This is essential for search engine performance — bulk writes are significantly faster than individual indexing requests, often by an order of magnitude.

A batch is flushed when either of two conditions is met: the batch reaches a configured size, or a timeout expires since the last flush. The timeout ensures documents are not left waiting indefinitely in low-volume scenarios.

Single Indexer Per Run

Only one Indexer can be defined in a Lucille run. All pipelines feed to the same Indexer. This simplifies the system — there is one destination, one set of batching parameters, one connection to manage — and reflects the common reality that a search ingestion project writes to a single search backend.

When documents from different pipelines need to land in different indices within the same backend, Lucille supports index routing: a field on the document determines which index it is sent to, without requiring multiple Indexer definitions.

Deletion Support

Indexers support two deletion mechanisms — delete-by-ID and delete-by-query — triggered by marker fields on the document. This enables CDC patterns and incremental ingestion: a Connector can emit a document that represents “delete this record from the index” rather than “index this record,” and the Indexer translates that intent into the appropriate backend operation.

Field Filtering

The Indexer applies field filtering at the boundary — stripping internal fields, applying whitelist/blacklist rules — so that only the intended fields reach the search backend. This filtering happens at indexing time, not during pipeline processing, so Stages always see the full document.

Error Handling at the Batch Level

Search engine bulk APIs can return mixed results: some documents accepted, others rejected. The Indexer inspects per-document responses and reports individual failures without failing the entire batch. Documents that succeed are marked complete; documents that fail are marked failed. Both are tracked in the run summary.

Practical Guide

For how to configure indexers — generic parameters, field filtering, deletion mechanics, and backend-specific settings — see Indexers in the Ingest Designer Guide.

For how to build a custom Indexer, see Developing Indexers.

9 - Runner

The Runner is the command-line entry point for launching a Lucille run. When invoked, it reads the configuration file, validates all component configurations, generates a unique runId, launches the configured components, waits for all work to complete, and prints a run summary.

What a Run Is

A Lucille Run is a sequence of Connectors executed one after the other. Each Connector feeds a specific Pipeline. A run can include multiple Connectors feeding multiple Pipelines, all sharing the same Indexer.

Connectors run strictly in sequence: the next Connector does not start until all documents from the previous Connector have been fully processed and indexed. This ordering guarantee is enforced automatically by the Publisher’s accounting system.

Run Lifecycle

For each Connector in the configured sequence, the Runner:

1. Validates the full configuration (fails fast on any misconfiguration).
2. Starts a WorkerPool (N Worker threads based on worker.threads).
3. Starts an Indexer thread.
4. Creates a PublisherImpl and launches the Connector in a ConnectorThread.
5. Blocks on publisher.waitForCompletion() until all work is done.
6. Logs the run summary and moves to the next Connector (or exits).

Starting a Run

Local mode (default):

java \
  -Dconfig.file=/path/to/config.conf \
  -cp 'lucille-core/target/lucille.jar:lucille-core/target/lib/*' \
  com.kmwllc.lucille.core.Runner

Distributed mode:

java \
  -Dconfig.file=/path/to/config.conf \
  -cp 'lucille-core/target/lucille.jar:lucille-core/target/lib/*' \
  com.kmwllc.lucille.core.Runner \
  -distributed

External mode (single JVM, Kafka messaging):

java \
  -Dconfig.file=/path/to/config.conf \
  -cp 'lucille-core/target/lucille.jar:lucille-core/target/lib/*' \
  com.kmwllc.lucille.core.Runner \
  -external

Config validation only (no run):

Validates every Connector, Stage, and Indexer spec and prints all errors. Exits without executing anything.

java \
  -Dconfig.file=/path/to/config.conf \
  -cp 'lucille-core/target/lucille.jar:lucille-core/target/lib/*' \
  com.kmwllc.lucille.core.Runner \
  -validate

Render effective config (no run):

Prints the fully resolved configuration after HOCON substitutions (environment variables, include directives, etc.). Useful for debugging config variable expansion.

java \
  -Dconfig.file=/path/to/config.conf \
  -cp 'lucille-core/target/lucille.jar:lucille-core/target/lib/*' \
  com.kmwllc.lucille.core.Runner \
  -render

Run Configuration

runner {
  # Log detailed stage-by-stage metrics at end of run (default: INFO)
  metricsLoggingLevel: "INFO"

  # Connector timeout in milliseconds (default: 86400000 = 24 hours; set <= 0 to disable)
  connectorTimeout: 86400000
}

Run ID

The Runner generates a UUID runId for each run. The run ID is:

• Stamped on every Document by the Publisher (run_id field).
• Used as part of the Kafka event topic name in distributed mode.
• Included in log MDC so all log lines during a run include the run ID for filtering.

Run Summary

At the end of every run, the Runner logs a structured summary:

RUN SUMMARY: Success. 1/1 connectors complete. All published docs succeeded.
connector1: complete. 200000 docs succeeded. 0 docs failed. 0 docs dropped. Time: 416.47 secs.
Run took 417.46 secs.

A connector that failed entirely is distinguished from one that completed with individual document failures. Connectors after a failed one are listed as skipped.

Graceful Shutdown

The Runner handles SIGINT (Ctrl+C) and SIGTERM. On signal receipt:

1. The Connector stops publishing.
2. Workers drain remaining documents.
3. The Indexer flushes its current batch.
4. A partial run summary is logged.

RunType

Lucille supports four run types, selected via command-line flags:

Practical Guide

For deployment instructions — starting runs in each mode, command-line flags, and operational considerations — see Deployment.

For runner and other top-level configuration parameters, see Writing a Config.

9.1 - Runner Orchestration

Overview

The Runner is Lucille’s top-level orchestrator. It coordinates the full lifecycle of a run: validating configuration, instantiating components, executing connectors sequentially, and reporting results. All methods are static — the Runner is never instantiated.

A “run” is a sequential execution of one or more Connectors. Each Connector’s work must complete before the next begins. If any Connector fails, the run aborts.

How Runner.run() Coordinates the Full Lifecycle

The main execution flow:

public static RunResult run(Config config, RunType type, String runId) throws Exception {
    if (runId == null) {
        runId = Runner.generateRunId();  // UUID
    }
    MDC.put(RUNID_FIELD, runId);

    // 1. Validate FIRST
    Map<String, List<Exception>> validationErrors = runInValidationMode(config);
    if (!validationErrors.isEmpty()) {
        return new RunResult(false, ...);
    }

    // 2. Create connectors
    List<Connector> connectors = Connector.fromConfig(config);

    // 3. Execute each connector sequentially
    for (Connector connector : connectors) {
        // Create messenger factories based on RunType
        // Run connector with components
        ConnectorResult result = runConnectorWithComponents(...);
        if (!result.getStatus()) {
            return new RunResult(false, ...);  // Abort on failure
        }
    }

    return new RunResult(true, ...);
}

The Validation Step

Before any work begins, the Runner validates the entire configuration:

Map<String, List<Exception>> validationErrors = runInValidationMode(config);

This validates:

• Pipelines — each pipeline’s stages are instantiated to check for config errors
• Connectors — each connector’s config is checked for required/optional properties
• Indexer — the indexer config block is validated
• Other parents — publisher, runner, kafka, and other top-level config blocks

Validation is fail-all, not fail-fast. All errors are collected and reported together. If any validation errors exist, the run returns immediately with a failure result.

The Connector Loop

For each connector, the Runner:

1. Creates messenger factories appropriate for the RunType
2. Calls runConnectorWithComponents() which:
   • Starts a WorkerPool (if local mode)
   • Creates and starts an Indexer thread (if local mode)
   • Creates a Publisher
   • Calls runConnector() which:
     • Calls connector.preExecute(runId)
     • Launches a ConnectorThread that calls connector.execute(publisher) then publisher.flush()
     • Calls publisher.waitForCompletion(connectorThread, timeout)
     • Calls connector.postExecute(runId) (only if publishing succeeded)
3. Stops WorkerPool and Indexer in the finally block

private static ConnectorResult runConnectorWithComponents(...) {
    try {
        if (startWorkerAndIndexer && connector.getPipelineName() != null) {
            workerPool = new WorkerPool(config, pipelineName, localRunId, workerMessengerFactory, metricsPrefix);
            workerPool.start();

            IndexerMessenger indexerMessenger = indexerMessengerFactory.create();
            indexer = IndexerFactory.fromConfig(config, indexerMessenger, bypassIndexer, metricsPrefix, localRunId);
            indexerThread = new Thread(indexer);
            indexerThread.start();
        }

        publisher = new PublisherImpl(config, publisherMessenger, runId, ...);
        return runConnector(config, runId, connector, publisher);
    } finally {
        if (workerPool != null) { workerPool.stop(); workerPool.join(3000); }
        if (indexerThread != null) { indexer.terminate(); indexerThread.join(3000); }
    }
}

How RunType Affects Component Startup

public enum RunType {
    LOCAL,             // Workers + Indexer as threads; in-memory queues
    TEST,              // Same as LOCAL but bypass indexer backend; record message history
    EXTERNAL,          // Workers + Indexer as threads; Kafka for messaging
    DISTRIBUTED        // No local Workers/Indexer; Kafka messaging; assume external processes
}

The key decision:

boolean startWorkerAndIndexer = !type.equals(RunType.DISTRIBUTED);
boolean bypassSolr = type.equals(RunType.TEST);

The MessengerFactory Pattern

The Runner uses factory interfaces to decouple component creation from the RunType decision:

if (RunType.TEST.equals(type)) {
    TestMessenger messenger = new TestMessenger();
    history.put(connector.getName(), messenger);
    workerMessengerFactory = WorkerMessengerFactory.getConstantFactory(messenger);
    indexerMessengerFactory = IndexerMessengerFactory.getConstantFactory(messenger);
    publisherMessengerFactory = PublisherMessengerFactory.getConstantFactory(messenger);
} else if (RunType.LOCAL.equals(type)) {
    LocalMessenger messenger = new LocalMessenger(config);
    workerMessengerFactory = WorkerMessengerFactory.getConstantFactory(messenger);
    // ...
} else {
    workerMessengerFactory = WorkerMessengerFactory.getKafkaFactory(config, connector.getPipelineName());
    // ...
}

For LOCAL/TEST modes, a single messenger instance is shared (via getConstantFactory). For Kafka modes, each factory call creates a new messenger with its own Kafka consumer/producer.

Signal Handling for Clean Shutdown

When running via main(), the Runner registers an INT signal handler:

state = new RunnerState();

Signal.handle(new Signal("INT"), signal -> {
    if (state != null) {
        state.close();  // Close connector, publisher, workerPool, indexer
    }
    SystemHelper.exit(0);
});

RunnerState holds references to the currently-active components. When a connector starts, the state is populated:

if (state != null) {
    state.set(publisher, connector, workerPool, indexer, indexerThread);
}

When the connector finishes, the state is cleared. The RunnerState.close() method attempts orderly shutdown of each component, logging errors but not throwing.

ConnectorResult and RunResult Reporting

ConnectorResult captures per-connector outcomes:

• Status (success/failure)
• Error message (if failed)
• Duration in seconds
• Document counts from the Publisher (succeeded, failed)

RunResult aggregates across all connectors:

• Overall status
• List of all ConnectorResults
• Summary message (e.g., “2/3 connectors complete. Some docs failed.”)
• For TEST mode: a Map<String, TestMessenger> containing message history per connector

The Connector Timeout Mechanism

Each connector has a configurable timeout (default: 24 hours):

final int connectorTimeout = config.hasPath("runner.connectorTimeout") ?
    config.getInt("runner.connectorTimeout") : DEFAULT_CONNECTOR_TIMEOUT;
pubResult = publisher.waitForCompletion(connectorThread, connectorTimeout);

Inside waitForCompletion, the timeout is checked on each poll iteration:

if (timeout > 0 && ChronoUnit.MILLIS.between(start, Instant.now()) > timeout) {
    return new PublisherResult(false, "Connector timeout.");
}

This prevents a stuck connector from blocking the entire run indefinitely.

Sequential Connector Composition

Connectors execute strictly sequentially. The next connector only starts after the previous one fully completes (all documents indexed or failed):

for (Connector connector : connectors) {
    ConnectorResult result = runConnectorWithComponents(...);
    if (!result.getStatus()) {
        log.error("Aborting run because " + connector.getName() + " failed.");
        return new RunResult(false, ...);
    }
}

Each connector gets its own WorkerPool, Indexer, and Publisher. Components from one connector are fully stopped before the next connector’s components are created.

The main() Method and CLI Options

Option distributedOpt = Option.builder("distributed").hasArg(false)
    .desc("Uses Kafka for inter-component communication and doesn't execute pipelines locally.")
    .build();

Option external = Option.builder("external").hasArg(false)
    .desc("Modified local mode where workers and indexers are separate threads within the JVM communicating "
        + "through kafka")
    .build();

OptionGroup distributedType = new OptionGroup().addOption(distributedOpt).addOption(external);

Options cliOptions = new Options()
    .addOptionGroup(distributedType)
    .addOption(Option.builder("validate").hasArg(false)
        .desc("Validate the configuration and exit").build())
    .addOption(Option.builder("render").hasArg(false)
        .desc("Print out the configuration file with substitutions applied and exit").build());

The -validate and -render flags can be combined. -distributed and -external cannot as they are mutually exclusive, supplying both is rejected.

Anything the parser doesn’t recognize (unknown flags or leftover positional arguments) cause the Runner to log the usage text and exit 1 rather than starting a run.

Thread Model (Local Mode)

For each connector in a local run, there are 4+ threads:

1. Main thread — runs waitForCompletion(), polling for events
2. ConnectorThread — calls connector.execute(publisher), publishes documents
3. Worker thread(s) — poll documents, run pipeline, emit results (configurable count)
4. Indexer thread — polls processed documents, sends to search engine

Plus a WorkerPool watcher thread that monitors worker health and logs statistics.

10 - Config

Configuration as an Architectural Principle

Lucille is a configuration-driven system. Every aspect of an ingest — which sources to read, which enrichment stages to apply, which search backend to write to, how many worker threads to run, what retry policy to use — is declared in a configuration file rather than hardcoded in application logic.

This is not merely a convenience. It is a fundamental architectural decision with consequences that ripple through the entire system:

Separation of concerns. The framework provides the execution engine; the configuration provides the instructions. A developer can change what an ingest does — add a stage, switch a connector, adjust batch sizes — without modifying or rebuilding any code. The same compiled JAR serves every ingest; only the config file changes.

Composability. Because ingests are defined declaratively, they can be composed from reusable pieces. A connector definition, a pipeline fragment, or a set of connection parameters can be defined once and included in multiple configs. This prevents drift and duplication across an organization’s ingests.

Validatability. Because every component declares what configuration it expects (via the SPEC system), the entire config can be validated before any work begins. Errors are caught at startup — all at once, not one at a time — rather than surfacing mid-run after hours of processing.

Environment portability. The same config file works across development, staging, and production by substituting environment-specific values (URLs, credentials, index names) from environment variables at load time. No code changes, no separate config files per environment, no rebuild.

Why HOCON and Typesafe Config

The choice of configuration format and library is not a peripheral implementation detail. In a system where every component — every Stage, Connector, and Indexer — receives its instructions through configuration, the config library is effectively the API between the user and the framework. Its capabilities and limitations shape what users can express and how the system behaves.

Lucille uses HOCON (Human-Optimized Config Object Notation) parsed by the Typesafe Config library. This choice provides several properties that matter architecturally:

Comments. A config file that defines an entire ingestion pipeline — potentially dozens of stages, multiple connectors, connection details, tuning parameters — requires annotation. JSON does not support comments. HOCON does. This is not a minor ergonomic preference; it is the difference between a config that is self-documenting and one that requires external documentation to understand.

Environment variable substitution. The ${?ENV_VAR} syntax allows credentials and environment-specific values to be injected at load time without any application code involvement. This means the framework never needs to implement its own env-var resolution logic, and the precedence rules (file value vs. env var) are defined in one place — the config file itself — rather than scattered across component implementations.

File includes. HOCON’s include directive enables config composition. Shared settings (connection strings, common pipeline fragments) are defined once and included everywhere. This is what makes it practical for an organization to maintain dozens of ingests against shared infrastructure without duplicating connection details that drift out of sync.

Internal variable substitution. HOCON supports references within a config file, so a value defined once can be reused across multiple blocks. Combined with environment variable substitution, this eliminates repetition and ensures consistency — change a value in one place and it propagates everywhere it’s referenced.

Relaxed syntax. HOCON allows omitting quotes around keys, using = or : for assignment, and trailing commas. This makes configs more readable and less error-prone to edit by hand than strict JSON.

List concatenation. HOCON supports appending to lists and concatenating adjacent arrays. This enables patterns like composing a stages list from multiple included fragments — each fragment contributes its stages, and HOCON merges them into a single list at resolution time.

Typed access with path expressions. The Typesafe Config library provides getString(), getInt(), getConfigList(), and other typed accessors with dot-path navigation. This means component code reads configuration through a clean, typed API rather than parsing raw text. Type errors are caught at access time with clear messages.

Automatic resolution order. The library merges configuration from system properties, the application config file, and reference defaults in a defined precedence. This means any config value can be overridden via a system property (-Dproperty=value) without modifying the file — useful for one-off test runs and CI overrides.

Config file selection at runtime. The -Dconfig.file system property tells Lucille which config file to use. This keeps the application generic — the same JAR, the same classpath, the same entrypoint — with only the config file varying between runs. In containerized deployments, this means a single Docker image can serve any ingest by varying an environment variable at launch time.

These properties combine to make configuration a first-class architectural concern rather than an afterthought. The config file is not just “where you put settings” — it is the primary interface through which users interact with the framework, and its expressiveness directly determines how maintainable, portable, and correct an ingest can be.

Practical Guides

For how to write a config file, see Writing a Config in the Ingest Designer Guide.

For operational patterns — environment variable substitution, containerized deployment, config composition, and distributed mode configuration — see Configuration Management in the Operations Guide.