Component Developer Guide

• 1: What the Framework Gives You
• 2: Developing New Components
• 3: Developing Stages
• 4: Developing Connectors
• 5: Developing Indexers
• 6: Developing File Handlers
• 7: Developing Storage Clients
• 8: SPEC Validation System
• 9: Quick Reference
• 10: Testing Pipelines
• 11: Javadoc

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

stages: [
  {
    class: "com.mycompany.ingest.MyStage"
    myParam: "value"
  }
]

1 - What the Framework Gives You

When you are developing components for Lucille, it is important to understand what the framework does and does not give you. That understanding lets you take full advantage of what is already built, and gives you realistic expectations about the responsibilities that will fall to you as the implementor.

This page begins with a general discussion of what you get when you adopt the framework overall. The following sections then focus on each component type — Stage, Connector, and Indexer. When you implement a component you will be extending the Stage or Indexer abstract classes, or implementing the Connector interface. Each section explains how the framework interacts with that component, what the base class or framework infrastructure does for you automatically, and what you are responsible for writing yourself.

What You Are Really Getting When You Adopt Lucille

When you adopt Lucille, you are not just getting a library of connectors and stages. You are getting a runtime that solves the hard problems of production search ingestion — problems that are invisible during a POC but dominate the engineering effort as a system matures.

You are getting a concurrency model you don’t have to build. Reading, enriching, and indexing run as independent components communicating through queues. You don’t write threading code, you don’t manage shared state between components, and you don’t debug race conditions. The framework handles the concurrent execution; you write sequential logic inside each component.

You are getting completion detection you don’t have to invent. In a distributed system where multiple workers and indexers process documents asynchronously — and where new child documents can be generated mid-pipeline — knowing when “all the work is done” is a genuinely hard problem. Lucille’s Publisher solves it with an event-driven accounting model that tracks every document from publication to terminal state. You call publisher.publish(doc) and the framework tells you when everything is finished.

You are getting fault tolerance you don’t have to implement. Proper Kafka offset management, consumer group rebalancing, at-least-once delivery guarantees, and poison pill detection are built into the framework’s messaging layer. You don’t write offset commit logic or crash recovery code. You write a stage that processes one document at a time, and the framework ensures that document is never silently lost.

You are getting a deployment model you don’t have to choose upfront. The same pipeline code runs in a single JVM for development and in a fully distributed Kafka deployment for production. You don’t architect for scale on day one and you don’t rewrite when scale arrives. The framework’s pluggable messenger abstraction means the transition from local to distributed is a command-line flag, not a code change.

You are getting batching, retry, and error handling you don’t have to get right. Sending documents to a search engine in production involves batch accumulation with dual thresholds, exponential backoff on transient failures, per-document vs. batch-level failure discrimination, and correct event reporting for every outcome. The base Indexer class handles all of this. You implement one method — sendToIndex(List<Document>) — and the framework handles everything around it.

You are getting a test infrastructure that makes correctness verifiable. The test mode captures the complete history of every document through the system — what was published, what was processed, what was indexed, what failed. You write assertions against this history without mocking the framework itself. The system under test is the real system, running in memory, with the search backend bypassed.

You are getting a configuration system that scales from a single file to a multi-team deployment. HOCON with environment variable substitution, file includes, and pre-run validation means your pipeline definition works unchanged from a developer’s laptop to a Kubernetes CronJob to a distributed Kafka deployment. Credentials come from environment variables. Shared settings come from included files. Typos are caught before the run starts.

You are getting a Document API designed for search. Single-valued and multi-valued fields, typed access without casts, update modes that match how search engines actually work, nested JSON support, and zero-cost serialization to the wire format that every search backend expects. Every stage you write is a few lines of domain logic rather than a page of field-access boilerplate.

You are getting a library of enrichment stages you don’t have to write. Text extraction, OCR, NER, embeddings, chunking, database lookups, HTTP enrichment, scripting, and dozens of field manipulation operations — all configurable, all composable, all tested. For many pipelines, the custom code you write is zero: the pipeline is pure configuration.

What you are not getting is a visual workflow editor, a managed cloud service, or a connector catalog for hundreds of SaaS sources. Lucille is a framework for engineers who write code. Its value proposition is that the code you write is small, focused on your specific problem, and surrounded by a runtime that handles everything else correctly.

The remainder of this document details exactly how this division of labor works for each component type — what the framework handles, and what you as the implementor are responsible for.

Stages

What the framework handles (you don’t have to)

Conditional execution. The base Stage class evaluates the conditions configuration block before calling processDocument(). If conditions don’t match, the stage is skipped entirely. You never write if (doc.has("myField")) checks for basic field-presence gating — configure it in HOCON and the framework handles it.

Dropped and skipped document handling. If a document is marked as dropped or skipped, shouldProcess() returns false and processDocument() is never called. You don’t need to check these flags.

Per-stage metrics. The framework automatically tracks:

• Processing time per document (a Codahale Timer)
• Error count (a Counter incremented when processDocument throws)
• Child document count (a Counter incremented for each child emitted)

These are registered with a shared MetricRegistry and reported at the end of the run. You get per-stage performance visibility for free.

Per-document logging. The DocLogger automatically logs stage entry and exit for every document processed, with the stage name and document ID. You don’t need to add logging for “stage X processed document Y.”

Child document lifecycle management. When processDocument() returns an iterator of children, the framework:

• Copies the parent’s run ID to each child
• Counts children in the metrics
• Ensures children flow through downstream stages (but not upstream ones)
• Ensures children are emitted before the parent in the result iterator (so the Publisher learns about children before the parent completes)

Thread safety via per-thread instantiation. Each Worker thread gets its own Pipeline with its own Stage instances. You can use instance fields freely without synchronization. The framework guarantees this isolation.

Configuration validation. The SPEC you declare is validated before the run starts. If a user provides an unrecognized parameter or omits a required one, the error is caught at startup — not when your stage tries to read it.

Automatic naming. If no name is configured, the framework assigns one based on position (stage_1, stage_2, etc.).

Reflective instantiation. The framework instantiates your stage from the class property in config. You don’t register it anywhere — just put it on the classpath.

What you must handle (with guidance from existing implementations)

Resource initialization and cleanup. If your stage needs external resources (database connections, HTTP clients, loaded models), initialize them in start() and clean them up in stop(). The framework calls these at the right time but doesn’t know what resources you need. See FetchUri (HTTP client), DatabaseLookup (JDBC connection), JlamaEmbed (model loading).

Error semantics. Decide whether an error should fail the document (throw StageException) or be handled gracefully (log and continue). The framework catches StageException and routes the document to a failure state. If you catch exceptions internally and continue, the document proceeds through the pipeline. See FetchUri for graceful degradation (logs the error, leaves the field empty) vs. stages that throw on any failure.

Child document creation. If your stage generates children, you must create them with unique IDs, populate their fields, and return them as an iterator. The framework handles everything after that. See ChunkText (attaches children to parent) and EmitNestedChildren (converts attached children to emitted children).

Idempotency. In distributed mode, a document may be processed more than once (crash + redelivery). Your stage should produce the same result if called twice on the same input. Most stages are naturally idempotent (setting a field to a computed value is idempotent). Stages with side effects (writing to an external system, incrementing a counter) need explicit consideration.

Multi-valued field handling. Decide whether your stage should process the first value of a field (getString) or all values (getStringList). The Document API supports both patterns, but you must choose the right one for your use case.

Memory management for expensive resources

If your Stage initializes a memory-consuming resource — a large dictionary, a machine learning model, a compiled rule set — it’s important to understand how many copies of that resource will exist in memory and when they are released.

The default behavior. Lucille creates a separate instance of your Stage for each Worker thread. If a pipeline has 4 Worker threads, there are 4 instances of your Stage, each with its own copy of any resource initialized in start(). When the pipeline finishes (all documents for that connector are processed), the Worker threads and their Stage instances go out of scope and become eligible for garbage collection.

Multiple pipelines in the same run. If a run has two connectors feeding two pipelines, and both pipelines use your Stage, the pipelines execute sequentially — not concurrently. The first pipeline’s components are created, used, and released before the second pipeline’s components are created. So the maximum number of copies of your resource in memory at any given time is equal to the number of Worker threads (e.g., 4), not the total across both pipelines (not 8). However, you pay the initialization cost once per Worker thread per pipeline — in this example, 8 initializations total across the run.

Optimizing with a singleton. If your resource is read-only (immutable after initialization), you can avoid both the redundant memory and the redundant initialization by storing it as a singleton that is initialized once and reused across all Stage instances for the lifetime of the JVM. In this case, there is one initialization and one copy in memory, regardless of how many Worker threads or pipelines use the Stage.

Stage code accesses the resource through the singleton rather than holding it as an instance field. The singleton must be thread-safe for concurrent reads (which is trivially true for immutable data structures like an unmodifiable Map).

Example: DictionaryManager. Lucille’s built-in DictionaryManager class demonstrates this pattern. It maintains a static cache of loaded dictionaries, keyed by file path. The first Stage instance to request a dictionary triggers the load; subsequent requests (from other Worker threads or later pipelines) receive the already-loaded instance. The getDictionary() method is synchronized to prevent duplicate loading during concurrent initialization. This pattern can be copied and adapted for other expensive read-only resources.

Connectors

What the framework handles (you don’t have to)

Lifecycle orchestration. The Runner calls preExecute(), execute(), postExecute(), and close() in the correct order with the correct error-handling semantics:

• execute() is not called if preExecute() throws
• postExecute() is not called if execute() throws
• close() is always called regardless of success or failure
• The run is aborted if any lifecycle method throws

You implement the methods; the framework calls them at the right time.

Document publication and tracking. The Publisher passed to execute() handles:

• Stamping the run ID on each document
• Tracking document IDs for completion accounting
• Backpressure (blocking publish() when too many documents are in flight)
• Thread safety (multiple threads can call publish() concurrently)
• Collapsing mode (merging consecutive same-ID documents if configured)

You just call publisher.publish(doc) and the framework handles everything else.

Sequential connector composition. If multiple connectors are configured, the Runner ensures each one completes fully (all documents processed and indexed) before the next starts. You don’t coordinate with other connectors.

Configuration validation. The SPEC declared in your connector is validated before the run starts. The AbstractConnector base class handles parsing common config (name, pipeline, docIdPrefix, collapse) so you don’t repeat it.

Doc ID prefixing. If docIdPrefix is configured, AbstractConnector.createDocId(id) prepends it. You call this helper when creating document IDs.

Reflective instantiation. Like stages, connectors are instantiated from the class property in config.

What you must handle (with guidance from existing implementations)

Source system connection management. Open connections in preExecute() or at the start of execute(). Close them in close(). The framework doesn’t know how to connect to your source. See FileConnector (initializes StorageClients in execute()), DatabaseConnector (opens JDBC in preExecute()).

Document creation with meaningful IDs. Each document needs a unique ID. The ID should be stable across re-runs (so re-ingestion updates rather than duplicates). See FileConnector (uses file path as ID), DatabaseConnector (uses a configured ID column).

Incremental state tracking. If your connector needs to track what it has already published (to avoid re-processing unchanged data), you must implement that state management. See FileConnector’s FileConnectorStateManager (JDBC-backed state tracking of published files and their modification times).

Error handling during iteration. If reading one record from the source fails, decide whether to skip it and continue or abort the connector. The framework aborts the run if execute() throws, but individual record failures within execute() are your responsibility to handle. See FileConnector (logs and continues on individual file errors) vs. DatabaseConnector (wraps the entire query in a try-catch).

Pagination and memory management. If the source has millions of records, you can’t load them all into memory. Use streaming/cursor-based access and publish documents as you go. The Publisher’s backpressure (maxPendingDocs) will block you if you’re publishing faster than the system can process, but you must still avoid loading the entire dataset into memory before publishing. See DatabaseConnector (uses JDBC fetch size for streaming).

Tombstone/deletion document generation. If your connector needs to signal deletions (records that existed in a previous run but are now gone), you must create documents marked for deletion. See FileConnector’s sendExpiredFileTombstones() (creates skipped documents with an expired flag for files no longer in storage).

Indexers

What the framework handles (you don’t have to)

Batching with size and timeout. The base Indexer class accumulates documents in a Batch and flushes when either the configured batchSize is reached or batchTimeout milliseconds have elapsed. Your sendToIndex() method receives a pre-batched list of documents. You never manage batch accumulation or timeout logic.

MultiBatch for index routing. When indexOverrideField is configured, the framework uses a MultiBatch that maintains separate batches per destination index, flushing them independently. Your implementation receives documents already grouped by destination.

Retry with exponential backoff. When indexer.maxRetries is configured, the framework wraps your sendToIndex() call in a Resilience4j retry. If you throw an IndexerRetryableException with a status code in the configured retryable list, the framework retries automatically with exponential backoff. You just throw the right exception type; the framework handles retry logic.

Event reporting for accounting. After each batch, the framework sends FINISH events for successful documents and FAIL events for failed ones. You return a set of failed document/reason pairs from sendToIndex(); the framework handles all event communication with the Publisher.

Field filtering (whitelist/blacklist). The getIndexerDoc() method applies the configured field filter before you see the document. Reserved fields (___dropped, ___skipped, ___children) are stripped. You call getIndexerDoc(doc) and get a clean map ready for the search backend.

ID override. getDocIdOverride(doc) returns the override ID if configured, or null. You check this and use it as the document ID in the search backend.

Index override. getIndexOverride(doc) returns the destination index/collection override if configured, or null.

Connection validation. The framework calls validateConnection() before processing any documents. If it returns false, the run is aborted immediately — no documents are wasted on a broken connection.

Offset commitment (in WorkerIndexer mode). The batchComplete() call in the finally block of sendToIndexWithAccounting always fires, allowing the messenger to commit Kafka offsets. You don’t manage offset semantics.

Graceful shutdown. The terminate() method sets a flag that causes the polling loop to exit after the current batch. The framework handles the shutdown sequence.

Metrics. The framework tracks indexing rate (documents/second), backend latency per batch, and logs periodic status updates.

Bypass mode. When bypass=true (test mode), the framework skips your sendToIndex() entirely. You don’t need to handle test mode in your implementation.

What you must handle (with guidance from existing implementations)

Bulk API construction. Translate Lucille Documents into the search backend’s bulk request format. This is backend-specific: Solr uses SolrInputDocument, OpenSearch uses BulkRequest.Builder, Pinecone uses its own upsert API. See SolrIndexer.toSolrDoc(), OpenSearchIndexer.uploadDocuments().

Per-document failure extraction from bulk responses. The bulk API may succeed overall but report failures for individual documents. You must parse the response, identify which documents failed, and return them as Set<Pair<Document, String>>. See OpenSearchIndexer (iterates BulkResponseItem checking for errors) and SolrIndexer (catches exceptions per-collection).

Deletion handling. Detect documents marked for deletion (using deletionMarkerField/deletionMarkerFieldValue) and issue the appropriate delete operation (delete-by-ID or delete-by-query). The framework provides the config values; you implement the detection logic and the delete API call. See SolrIndexer.isDeletion() and OpenSearchIndexer.isMarkedForDeletion().

Ordering within a batch. If a batch contains both an upsert and a delete for the same document ID, you must ensure the operations are sent in the correct order. The SolrIndexer handles this by flushing pending upserts before processing a delete for the same ID. The OpenSearch indexer handles it by removing conflicting entries from the upload/delete maps.

Connection management. Create the search engine client in the constructor, validate it in validateConnection(), close it in closeConnection(). Handle TLS, authentication, and certificate validation as needed. See OpenSearchUtils.getOpenSearchRestClient() for TLS/auth setup.

Nested/child document transformation. If the search backend supports nested documents (Solr does), transform Lucille’s attached children into the backend’s nested format. See SolrIndexer.addChildren().

Version type handling. If the backend supports optimistic concurrency via version numbers (OpenSearch/Elasticsearch do), extract the version from the document (typically the Kafka offset) and include it in the request. See OpenSearchIndexer’s versionType and versionNum logic.

Retryable vs. non-retryable error classification. When a bulk call fails, decide whether to throw IndexerRetryableException (transient, worth retrying) or IndexerException (permanent, don’t retry). The distinction is typically based on HTTP status code. See OpenSearchIndexer (wraps OpenSearchException with status code, wraps IOException as unknown status).

Summary: The Division of Labor

Summary

The framework’s goal is to let you write the 20% of code that is specific to your source system, your enrichment logic, or your search backend — while the 80% that is common to all search ETL (batching, retry, accounting, metrics, error handling, threading, configuration) is handled once, correctly, in the framework. The opening section of this document describes what that means in practice; the tables above show exactly where the boundaries are.

2 - Developing New Components

Each component type has its own dedicated guide:

• Developing Stages — Skeleton, lifecycle, conditional execution, and the Document API.
• Developing Connectors — Skeleton, lifecycle, and publishing documents.
• Developing Indexers — Skeleton, lifecycle, and sending documents to a destination.

All components must declare a SPEC and follow the Javadoc Standards. See Testing Pipelines for testing conventions.

3 - Developing Stages

To create a Stage, extend the abstract Stage class and implement processDocument(). That is the only method you are required to provide. The base class handles everything else: config validation, condition evaluation, metrics, thread isolation, and error routing.

What the base class does for you:

• Config validation — The constructor calls getSpec().validate(config) using your class’s SPEC field. If the config has missing required properties, unknown properties, or type mismatches, validation fails at startup with a clear error message. You never call this yourself.
• Condition evaluation — If the user configures conditions on your stage, the base class evaluates them before calling processDocument(). If conditions are not met, your method is never invoked for that document. You write processDocument() as if conditions are always satisfied.
• Thread isolation — Each worker thread gets its own instance of your stage. Instance fields are effectively thread-local; no synchronization is needed.
• Metrics — The base class tracks per-stage document count, latency, error count, and child count automatically.
• Error handling — If processDocument() throws a StageException, the framework catches it, marks the document as failed, and continues processing other documents.

What you implement:

What you declare:

Your constructor must call super(config) — this triggers SPEC validation and condition parsing. After super() returns, you can safely read config values into instance fields.

What’s in config: The Config passed to your Stage constructor contains only the properties defined inside your stage’s config block — the { ... } element from the stages list. It does not contain the full Lucille config. You read your parameters directly: config.getString("myParam"). Your SPEC should declare only the properties that belong to your stage.

Stage Skeleton

Every stage must follow the Javadoc Standards.

package com.kmwllc.lucille.stage;

import com.kmwllc.lucille.core.Stage;
import com.kmwllc.lucille.core.StageException;
import com.kmwllc.lucille.core.spec.Spec;
import com.kmwllc.lucille.core.spec.SpecBuilder;
import com.kmwllc.lucille.core.ConfigUtils;
import com.typesafe.config.Config;
import java.util.Iterator;
import com.kmwllc.lucille.core.Document;

/**
 * One‑line summary.
 * <p>
 * Config Parameters -
 * <ul>
 *   <li>foo (String, Required) : Description.</li>
 *   <li>bar (Integer, Optional) : Description. Defaults to 10.</li>
 * </ul>
 */
public class ExampleStage extends Stage {
  
  public static final Spec SPEC = SpecBuilder.stage()
      .requiredString("foo")
      .optionalNumber("bar")
      .build();
  
  private final String foo;
  private final int bar;

  public ExampleStage(Config config) throws StageException {
    super(config);
    this.foo = config.getString("foo");
    this.bar = ConfigUtils.getOrDefault(config, "bar", 10);
  }

  @Override
  public Iterator<Document> processDocument(Document doc) throws StageException {
    // mutate doc as needed
    doc.setField("out", foo + ":" + bar);
    // return null unless emitting child docs
    return null;
  }
}

Conditional Execution

Conditions are configured by the user in the stage’s config block. Here’s what that looks like:

{
  class: "com.kmwllc.lucille.stage.MyStage"
  conditions: [
    { fields: ["status"], values: ["active"], operator: "must" }
  ]
}

Guideline: Prefer conditions over in-code skip logic. If your stage should only run on documents that have a certain field or a certain value, that decision belongs in the conditions config — not in processDocument(). This keeps stages reusable (the same stage can be applied with different conditions in different pipelines) and keeps the skip logic visible in the config rather than hidden in code.

There are legitimate exceptions. Some stages return early from processDocument() when a required source field is absent or empty — for example, FetchUri returns immediately if the URL field is missing or blank, DetectLanguage returns early if the accumulated text is shorter than a minimum length, and ParseJson returns early if its source field doesn’t exist. These are cases where the stage’s logic fundamentally cannot proceed and the check is more nuanced than a simple field-existence condition (e.g., checking isEmpty(), or evaluating a computed threshold). When you do return early, do so silently — don’t throw a StageException for ordinary data variation.

For the full conditions reference and all other control flow options — skipping, dropping, error handling, child documents, and connector sequencing — see Control Flow.

Stage Scope: One Stage or Several?

When designing a stage, you’ll sometimes face the question of whether to build one stage that performs multiple internal steps, or several smaller stages that the user composes in config.

Prefer a single stage when:

• The intermediate state is only needed to connect internal operations and is not used by any other stage in the pipeline. Creating a field, using it once, and deleting it is three stages of ceremony for what should be one stage of work.
• Splitting would require the user to configure matching field names across stages and then clean up afterward.
• The combined operation is conceptually one thing from the user’s perspective (e.g., “look up the hash of this field in a dictionary” — the hash is an implementation detail, not a user-visible artifact).

Prefer separate stages when:

• The intermediate result is actually used elsewhere in the pipeline — another stage reads it, it’s indexed, or it’s used in a condition.
• The operations are independently reusable and the user is actually reusing them independently in this pipeline.
• The operations need different conditions or different error handling.

Pipeline simplicity is a valid design goal. A pipeline where many stages exist only to create or clean up intermediate fields is harder to read, harder to maintain, and more error-prone than one where each stage does a complete unit of work. The user shouldn’t have to think about plumbing between stages when the plumbing serves no purpose beyond connecting internals.

That said, if a stage becomes so large that it’s doing several unrelated things and has a dozen config parameters, it’s probably too big — not because of the intermediate field question, but because it’s no longer a focused, testable unit.

The balance: One stage should do one user-visible thing, even if that thing involves multiple internal steps. The litmus test is: would a user in this pipeline benefit from the intermediate field existing as a separate, visible document field? If not, keep it internal to a single stage.

Reading & Writing Fields

For the Document API — reading fields, writing fields, update modes, nested JSON, and supported types — see The Document API in the Quick Reference.

Fetching File Content

If your stage needs to read file content from a path stored on a document (e.g., fetching a PDF for text extraction, loading a dictionary file, reading a template), use FileContentFetcher rather than opening files directly. This gives your stage transparent support for local files, classpath resources, and cloud storage (S3, Azure, GCS) — and allows users to plug in a custom fetcher via the fetcherClass config property.

private FileContentFetcher fileFetcher;

@Override
public void start() throws StageException {
  this.fileFetcher = FileContentFetcher.create(config);
  try {
    fileFetcher.startup();
  } catch (IOException e) {
    throw new StageException("Failed to initialize file fetcher", e);
  }
}

@Override
public Iterator<Document> processDocument(Document doc) throws StageException {
  String path = doc.getString("file_path");
  try (InputStream is = fileFetcher.getInputStream(path)) {
    // process the file content
  } catch (IOException e) {
    throw new StageException("Failed to fetch " + path, e);
  }
  return null;
}

@Override
public void stop() throws StageException {
  fileFetcher.shutdown();
}

By using FileContentFetcher.create(config), your stage automatically supports the fetcherClass config property — users can substitute a custom implementation that resolves paths differently (e.g., fetching from a CMS API or interpreting paths relative to document metadata). See Custom File Content Fetchers for details on implementing a custom fetcher.

Unit Testing

Stage unit tests follow a consistent pattern: create a stage from a config, create a document, call processDocument(), and assert on the document’s state afterward.

StageFactory

StageFactory eliminates the boilerplate of instantiating and starting a stage in tests. It handles reflection, config loading, and calling start() — returning a stage that’s ready to process documents.

private final StageFactory factory = StageFactory.of(MyStage.class);

StageFactory provides several get() overloads:

Each get() call creates a new stage instance and calls start() on it before returning.

The basic pattern

public class MyStageTest {

  private final StageFactory factory = StageFactory.of(MyStage.class);

  @Test
  public void testBasicBehavior() throws StageException {
    Stage stage = factory.get("MyStageTest/basic.conf");

    Document doc = Document.create("doc1");
    doc.setField("input", "hello");

    stage.processDocument(doc);

    assertEquals("HELLO", doc.getString("output"));
  }
}

Testing conditions

Use processConditional() instead of processDocument() when you want to verify that conditions are evaluated correctly. processDocument() bypasses conditions; processConditional() respects them.

@Test
public void testConditionalExecution() throws StageException {
  Stage stage = factory.get("MyStageTest/conditional.conf");

  Document matching = Document.create("doc1");
  matching.setField("status", "active");
  stage.processConditional(matching);
  assertTrue(matching.has("enriched"));  // stage ran

  Document nonMatching = Document.create("doc2");
  nonMatching.setField("status", "archived");
  stage.processConditional(nonMatching);
  assertFalse(nonMatching.has("enriched"));  // stage skipped
}

Testing invalid configs

Use assertThrows to verify that bad configurations fail at startup (during construction or start()), not silently at runtime:

@Test
public void testBadConfig() {
  assertThrows(StageException.class, () -> factory.get("MyStageTest/missingRequired.conf"));
  assertThrows(StageException.class, () -> factory.get("MyStageTest/invalidValue.conf"));
}

Test config files

Store configs under src/test/resources/ in a subdirectory named after the test class:

src/test/resources/
  MyStageTest/
    basic.conf
    conditional.conf
    missingRequired.conf

Each config file contains just the stage’s config block — no pipelines, connectors, or indexer:

{
  class = "com.kmwllc.lucille.stage.MyStage"
  source = "input"
  dest = "output"
}

What to test

• Each parameter — at least one test path per required and optional parameter.
• Conditions — verify the stage is skipped when conditions are not met.
• Invalid configs — missing required fields, invalid values, type mismatches.
• Edge cases — empty fields, missing fields, multi-valued fields, null values.
• Child documents — if your stage emits children, assert on the returned iterator.

For full pipeline integration tests (running connectors, workers, and indexers end-to-end in memory), see Testing Pipelines.

4 - Developing Connectors

To create a Connector, extend AbstractConnector and implement execute(). That is the only method you are required to provide. The execute() method receives a Publisher — your connector reads from its data source and calls publisher.publish(doc) for each document it produces. That’s the core job of a connector: create documents and publish them. The base class handles config validation, name/pipeline resolution, document ID prefixing, and provides no-op defaults for the optional lifecycle methods.

Why Connector is an interface: Unlike Stage and Indexer, which are abstract classes, Connector is defined as an interface. This is because connectors have no processing loop managed by the framework — a connector owns its own execution entirely. The Runner calls execute() and the connector decides how to read data and when to publish documents. There is no framework-managed thread pool, no batching, and no message consumption loop to build into a base class. AbstractConnector provides the common config parsing and SPEC validation, but the execution model is entirely yours.

By contrast, Stage and Indexer are abstract classes because the framework manages their execution: stages are invoked per-document by the Worker’s processing loop, and indexers are driven by a message consumption loop with batching logic. Those base classes embed that execution machinery.

What AbstractConnector does for you:

• Config validation — The constructor calls getSpec().validate(config) using your class’s SPEC field. Missing or unrecognized properties fail at startup.
• Common config parsing — Reads name, pipeline, docIdPrefix, and collapse from the config automatically.
• Document ID prefixing — Provides createDocId(id) which prepends the configured docIdPrefix to your raw IDs.
• No-op lifecycle defaults — preExecute(), postExecute(), and close() are provided as no-ops so you only override what you need.

What you implement:

The Publisher:

The Publisher passed to execute() is your interface to the rest of the Lucille pipeline. Key things to know:

• Thread-safe — publish() can be called from multiple threads concurrently. If your data source supports parallel reads (e.g., fetching pages from an API, reading from multiple partitions), you can spawn threads inside execute() and have each one call publisher.publish(doc) independently. No synchronization on your part is needed.
• Backpressure — When publisher.maxPendingDocs is configured, publish() blocks automatically if the number of in-flight documents exceeds the threshold. This prevents a fast connector from overwhelming the pipeline. You don’t need to implement throttling yourself.
• Do not reuse documents after publishing — Once you call publisher.publish(doc), the document may be picked up by a worker thread immediately. Do not read from or write to the document after publishing it.
• Multi-threaded cleanup — If your connector uses multiple publishing threads, each thread should call publisher.preClose() when it is done publishing. This releases per-thread resources inside the publisher. Single-threaded connectors do not need to call preClose().
• Collapsing mode — If your connector sets collapse: true in its config, the publisher combines consecutive documents with the same ID into a single document with multi-valued fields. This is useful when your source emits multiple rows per logical record (e.g., a denormalized SQL join). Call publisher.flush() at the end of execute() if you use collapsing mode, to ensure the last held document is published.

What you declare:

Your constructor must call super(config) — this triggers SPEC validation and parses the common connector properties.

What’s in config: The Config passed to your Connector constructor contains only the properties defined inside your connector’s config block — the { ... } element from the connectors list. It does not contain the full Lucille config. You read your parameters directly: config.getString("sourceUri"). Your SPEC should declare only the properties that belong to your connector.

See Control Flow: Pre- and Post-Connector Actions for the full lifecycle contract, including what happens when each method throws.

Setup-Only Connectors

If your connector’s purpose is purely preparatory — it performs work but does not publish documents — omit the pipeline field in the config. The framework will call execute(null) synchronously, passing null for the publisher.

Put your logic in execute() and ignore the publisher parameter:

public class CreateCollectionConnector extends AbstractConnector {

  private final String solrUrl;
  private final String collection;

  public CreateCollectionConnector(Config config) {
    super(config);
    this.solrUrl = config.getString("solrUrl");
    this.collection = config.getString("collection");
  }

  @Override
  public void execute(Publisher publisher) throws ConnectorException {
    // publisher is null — this connector does not publish documents
    try (SolrClient client = new Http2SolrClient.Builder(solrUrl).build()) {
      CollectionAdminRequest.Create create =
          CollectionAdminRequest.createCollection(collection, 1, 1);
      create.process(client);
    } catch (Exception e) {
      throw new ConnectorException("Failed to create collection: " + collection, e);
    }
  }
}

This connector runs before any publishing connector in the same config. If it throws, the run aborts. Use close() for resource cleanup and postExecute() for success-only teardown, just as with any other connector.

See Control Flow: Setup-Only Connectors for the config-level view of this pattern.

Connector Skeleton

Every connector must follow the Javadoc Standards.

package com.kmwllc.lucille.connector;

import com.kmwllc.lucille.core.ConnectorException;
import com.kmwllc.lucille.core.Document;
import com.kmwllc.lucille.core.Publisher;
import com.kmwllc.lucille.core.spec.Spec;
import com.kmwllc.lucille.core.spec.SpecBuilder;
import com.typesafe.config.Config;

/**
* One-line summary of what this Connector reads and how it emits Documents.
* <p>
* Config Parameters -
* <ul>
*   <li>sourceUri (String, Required) : Where to read from (file://, s3://, http://, etc.).</li>
*   <li>batchSize (Integer, Optional) : Max items to read before publishing a batch. Defaults to 100.</li>
* </ul>
*/
public class ExampleConnector extends AbstractConnector {

  public static final Spec SPEC = SpecBuilder.connector()
      .requiredString("sourceUri")
      .optionalNumber("batchSize")
      .build();

  private final String sourceUri;
  private final int batchSize;

  public ExampleConnector(Config config) {
    super(config);
    this.sourceUri = config.getString("sourceUri");
    this.batchSize = config.hasPath("batchSize") ? config.getInt("batchSize") : 100;
  }
  
  @Override
  public void execute(Publisher publisher) throws ConnectorException {
    // Read from sourceUri and publish Documents.
    for (int i = 0; i < batchSize; i++) {
      Document d = Document.create(createDocId("item-" + i));
      // Populate fields on d as needed, e.g.: d.setField("source_uri", sourceUri);
      try {
        publisher.publish(d);
      } catch (Exception e) {
        throw new ConnectorException("Failed to publish document " + d.getId(), e);
      }
    }
  }
    
  @Override
  public void close() throws ConnectorException {
    // Optional: Close network or file handlers.
  }
}

Unit Testing

Connector tests verify that execute() publishes the expected documents. The standard pattern is: create a TestMessenger, wrap it in a PublisherImpl, instantiate your connector, call execute(), and assert on the documents captured by the messenger.

The basic pattern

public class MyConnectorTest {

  @Test
  public void testExecute() throws Exception {
    Config config = ConfigFactory.parseResourcesAnySyntax("MyConnectorTest/config.conf");
    TestMessenger messenger = new TestMessenger();
    Publisher publisher = new PublisherImpl(config, messenger, "run1", "pipeline1");

    Connector connector = new MyConnector(config);
    connector.execute(publisher);

    List<Document> docs = messenger.getDocsSentForProcessing();
    assertEquals(10, docs.size());
    assertEquals("expected-value", docs.get(0).getString("title"));
  }
}

TestMessenger captures all documents published during execute(). Use messenger.getDocsSentForProcessing() to retrieve them for assertion.

Mocking external services

Connectors that talk to external systems (databases, APIs, search engines) should inject a mock client rather than making real network calls. The common pattern is a constructor overload that accepts the client:

// Production constructor — creates a real client
public MyConnector(Config config) {
  this(config, createRealClient(config));
}

// Test constructor — accepts a mock
public MyConnector(Config config, MyClient client) {
  super(config);
  this.client = client;
}

In tests, pass a Mockito mock:

@Test
public void testWithMockClient() throws Exception {
  Config config = ConfigFactory.parseResourcesAnySyntax("MyConnectorTest/config.conf");
  MyClient mockClient = mock(MyClient.class);
  when(mockClient.query(any())).thenReturn(testData);

  TestMessenger messenger = new TestMessenger();
  Publisher publisher = new PublisherImpl(config, messenger, "run1", "pipeline1");
  Connector connector = new MyConnector(config, mockClient);
  connector.execute(publisher);

  assertEquals(5, messenger.getDocsSentForProcessing().size());
}

SolrConnector follows this pattern — its test constructor accepts a SolrClient mock.

Testing preExecute and postExecute

Test lifecycle methods directly when your connector overrides them:

@Test
public void testPreAndPostActions() throws Exception {
  Config config = ConfigFactory.parseResourcesAnySyntax("MyConnectorTest/actions.conf");
  MyClient mockClient = mock(MyClient.class);
  Connector connector = new MyConnector(config, mockClient);

  connector.preExecute("run1");
  verify(mockClient, times(1)).deleteByQuery("runId:run1");

  connector.postExecute("run1");
  verify(mockClient, times(1)).commit();
}

Testing error cases

Verify that your connector throws ConnectorException when it should:

@Test(expected = ConnectorException.class)
public void testExecuteFailsOnBadSource() throws Exception {
  Config config = ConfigFactory.parseResourcesAnySyntax("MyConnectorTest/badPath.conf");
  TestMessenger messenger = new TestMessenger();
  Publisher publisher = new PublisherImpl(config, messenger, "run1", "pipeline1");
  Connector connector = new MyConnector(config);
  connector.execute(publisher);
}

Test config files

Store configs under src/test/resources/MyConnectorTest/. Each config contains just the connector’s config block:

{
  name: "test-connector"
  class: "com.kmwllc.lucille.connector.MyConnector"
  pipeline: "pipeline1"
  sourceUri: "src/test/resources/MyConnectorTest/test-data.csv"
}

What to test

• Document output — correct number of documents, correct field values, correct IDs (including docIdPrefix behavior).
• Each parameter — at least one test path per required and optional parameter.
• Error handling — bad configs, unreachable sources, malformed data.
• Lifecycle methods — if you override preExecute() or postExecute(), test them independently.
• Edge cases — empty sources, sources with one record, large sources.

For full pipeline integration tests (running connectors through pipelines and indexers end-to-end in memory), see Testing Pipelines.

5 - Developing Indexers

To create an Indexer, extend the abstract Indexer class and implement three methods: sendToIndex(), validateConnection(), and closeConnection(). The base class manages the entire consumption loop — polling documents from the indexing queue, accumulating them into batches, flushing batches on size or timeout, sending completion/failure events, retrying transient failures, and logging throughput metrics.

Indexer is an abstract class (like Stage) rather than an interface because the framework drives the indexer’s execution. The base class implements Runnable and its run() method contains the message consumption loop, batching logic, retry machinery, and event accounting. Your implementation only provides the transport-specific operations: validate the connection, send a batch, and close the connection.

What the base class does for you:

• Message consumption loop — Polls the indexing queue, accumulates documents into batches, and flushes on batch size or timeout. You never write this loop.
• Config validation — Validates both the generic indexer config block (batchSize, field filtering, deletion markers, retry settings) and your implementation-specific config block (e.g., solr, opensearch) using your SPEC.
• Field filtering — Applies whitelist/blacklist filtering before documents reach sendToIndex(). Use getIndexerDoc(doc) to get the filtered field map.
• Batching — Configurable batch size and timeout, with support for per-document index routing via indexOverrideField.
• Retry with backoff — When indexer.maxRetries is configured, retries failed batches with exponential backoff and jitter. Your sendToIndex() returns per-document failures; the base class decides whether to retry based on status codes.
• Event accounting — Sends FINISH or FAIL events for each document so the Runner can track run completion.
• Metrics — Tracks documents indexed, throughput rate, and batch latency automatically.
• ID and index overrides — Supports idOverrideField and indexOverrideField for routing documents to different IDs or indices. Use getDocIdOverride(doc) and getIndexOverride(doc) in your sendToIndex().
• Deletion support — Detects documents marked for deletion via deletionMarkerField/deletionMarkerFieldValue or deleteByFieldField/deleteByFieldValue. Your implementation checks these in sendToIndex() and issues the appropriate delete operation.

What you implement:

What you declare:

Your constructor must call super(config, messenger, bypass, metricsPrefix, localRunId) — this triggers config validation, sets up batching, and initializes metrics. The constructor signature must be (Config, IndexerMessenger, boolean, String, String) because the IndexerFactory instantiates indexers reflectively using this signature.

What’s in config: Unlike Stages and Connectors, the Config passed to your Indexer constructor is the full root config for the entire Lucille run. The base class reads generic settings from config.getString("indexer.idOverrideField") etc. Your implementation reads its own block via config.getConfig("mykey") (e.g., config.getConfig("solr")). Your SPEC declares only the properties within your implementation-specific block — write "url", not "solr.url". The base class validates the generic indexer block separately.

Two-level config: Unlike Stages and Connectors, Indexers have a split config. The generic indexer {} block (batch size, field filtering, deletion markers, retries) is validated and consumed by the base class. Your implementation-specific block (e.g., solr {}, opensearch {}) is validated against your SPEC and read in your constructor. Your SPEC should only declare properties within your block — write "url", not "solr.url".

Indexer Skeleton

Every indexer must follow the Javadoc Standards.

package com.kmwllc.lucille.indexer;

import com.kmwllc.lucille.core.Document;
import com.kmwllc.lucille.core.Indexer;
import com.kmwllc.lucille.core.ConfigUtils;
import com.kmwllc.lucille.core.spec.Spec;
import com.kmwllc.lucille.core.spec.SpecBuilder;
import com.kmwllc.lucille.message.IndexerMessenger;
import com.typesafe.config.Config;
import java.util.List;
import java.util.Set;
import org.apache.commons.lang3.tuple.Pair;

/**
 * One-line summary of what this Indexer does and where it sends documents.
 * <p>
 * Config Parameters -
 * <ul>
 *   <li>url (String, Required) : Destination endpoint (e.g., base URL).</li>
 *   <li>index (String, Optional) : Default index/collection name. Defaults to "index1".</li>
 * </ul>
 */
public class ExampleIndexer extends Indexer {
  
  public static final Spec SPEC = SpecBuilder.indexer()
      .requiredString("url")
      .optionalString("index")
      .build();
  
  private final String url;
  private final String defaultIndex;
  
  public ExampleIndexer(Config config, IndexerMessenger messenger, boolean bypass,
      String metricsPrefix, String localRunId) {
    super(config, messenger, bypass, metricsPrefix, localRunId);
    Config implConfig = config.getConfig("example");
    this.url = implConfig.getString("url");
    this.defaultIndex = ConfigUtils.getOrDefault(implConfig, "index", "index1");
  }

  @Override
  protected String getIndexerConfigKey() {
    return "example";  // matches the config block name: example { url: "...", index: "..." }
  }

  @Override
  public boolean validateConnection() {
    // Health check to the destination — return false if unreachable
    return true;
  }

  @Override
  protected Set<Pair<Document, Exception>> sendToIndex(List<Document> documents) throws Exception {
    // Send the batch using your destination client.
    // Use getIndexerDoc(doc) to get the filtered field map.
    // Use getDocIdOverride(doc) if idOverrideField may be configured.
    // Return any failed docs as Pair<Document, Exception>; empty set if all succeeded.
    return Set.of();
  }

  @Override
  public void closeConnection() {
    // Close client resources
  }
}

Unit Testing

Indexer tests verify that sendToIndex() sends the correct data to the destination. The standard pattern is: create a TestMessenger, place documents on it for indexing, instantiate your indexer with a mock client, run it for a fixed number of iterations, and assert on what was sent to the mock.

The basic pattern

public class MyIndexerTest {

  @Test
  public void testBasicIndexing() throws Exception {
    Config config = ConfigFactory.empty()
        .withValue("indexer.batchSize", ConfigValueFactory.fromAnyRef(1));
    TestMessenger messenger = new TestMessenger();

    Document doc = Document.create("doc1", "test_run");
    doc.setField("title", "Hello");

    MyClient mockClient = mock(MyClient.class);
    Indexer indexer = new MyIndexer(config, messenger, false, "", null, mockClient);

    messenger.sendForIndexing(doc);
    indexer.run(1);

    verify(mockClient, times(1)).send(any());
  }
}

run(iterations) — the fixed-iteration polling loop

The Indexer base class provides run(int iterations) alongside the standard run(). While run() polls indefinitely until terminate() is called (designed for production use), run(iterations) polls exactly N times and then flushes the final batch. This is the key testing mechanism — it lets you control exactly how many poll cycles the indexer executes without needing threads or timeouts.

Each iteration polls one document from the messenger’s indexing queue. If you place 3 documents on the messenger and call indexer.run(3), the indexer will poll all 3, batch them according to batchSize, send them to your sendToIndex(), and then flush any remaining partial batch before closing.

TestMessenger as the indexing queue

TestMessenger simulates the messaging layer. Use messenger.sendForIndexing(doc) to place documents on the indexing queue before calling run(). After the run, use messenger.getSentEvents() to verify that the indexer sent the expected FINISH or FAIL events.

messenger.sendForIndexing(doc1);
messenger.sendForIndexing(doc2);
indexer.run(2);

List<Event> events = messenger.getSentEvents();
assertEquals(2, events.size());
assertEquals(Event.Type.FINISH, events.get(0).getType());
assertEquals(Event.Type.FINISH, events.get(1).getType());

Mocking the destination client

Indexers that talk to external systems (Solr, OpenSearch, Elasticsearch) should inject a mock client. The common pattern is a constructor overload:

// Production constructor — creates a real client
public MyIndexer(Config config, IndexerMessenger messenger, boolean bypass,
    String metricsPrefix, String localRunId) {
  this(config, messenger, bypass, metricsPrefix, localRunId, createRealClient(config));
}

// Test constructor — accepts a mock
public MyIndexer(Config config, IndexerMessenger messenger, boolean bypass,
    String metricsPrefix, String localRunId, MyClient client) {
  super(config, messenger, bypass, metricsPrefix, localRunId);
  this.client = client;
}

Use Mockito’s ArgumentCaptor to inspect what was sent to the mock:

ArgumentCaptor<List<MyDocument>> captor = ArgumentCaptor.forClass(List.class);
verify(mockClient, times(1)).bulkIndex(captor.capture());
assertEquals(2, captor.getValue().size());
assertEquals("doc1", captor.getValue().get(0).getId());

What to test

• Successful indexing — documents reach the destination with correct fields and IDs.
• Field translation — your sendToIndex() correctly maps Document fields to the destination’s format.
• ID override — getDocIdOverride(doc) is used when idOverrideField is configured.
• Deletion — documents marked for deletion trigger delete operations instead of adds.
• Per-document failures — your sendToIndex() returns the correct Pair<Document, Exception> set when individual documents fail.
• validateConnection() — returns false when the destination is unreachable.
• closeConnection() — releases client resources without throwing.

For full pipeline integration tests (running connectors through pipelines and indexers end-to-end in memory), see Testing Pipelines.

6 - Developing File Handlers

A FileHandler turns a file’s content into Documents. Each implementation handles a specific file format — CSV, JSON, XML, or any custom format you need. The FileConnector uses FileHandlers to parse files it discovers during traversal.

To create a FileHandler, implement the FileHandler interface. The framework handles discovery, instantiation, and integration with the FileConnector.

What the framework does for you:

• Discovery by file extension — FileHandlers are mapped to file extensions in the connector’s fileHandlers config block. The framework instantiates your handler and routes files to it based on their extension.
• Config validation — Your getSpec() method declares legal properties; the framework validates the config at startup.
• InputStream management — The framework opens the file (from local disk, S3, Azure, GCS, or inside archives) and passes you an InputStream. You don’t need to know where the file came from.

What you implement:

Constructor: Your handler must have a public constructor that takes a single Config argument.

What’s in config: The Config passed to your FileHandler constructor contains only the properties defined inside your handler’s config block — the { ... } value for your file extension key within the fileHandlers map. For example, if the user writes:

fileHandlers: {
  yaml: {
    class: "com.mycompany.lucille.filehandler.YamlFileHandler"
    idField: "name"
  }
}

…your constructor receives a Config containing class and idField. It does not contain the connector config or the full Lucille config. Your SPEC should declare only your handler’s own properties.

Built-in defaults: If the user configures csv, json, jsonl, or xml without specifying a class, Lucille uses its built-in handlers. Specifying a class for any extension (including the built-in ones) overrides the default.

Referencing a Custom FileHandler from Config

Register your FileHandler by adding an entry to the fileHandlers block on a FileConnector, keyed by the file extension it handles. Include the class property with the fully qualified class name:

connectors: [{
  name: "ingest-yaml"
  class: "com.kmwllc.lucille.connector.FileConnector"
  pipeline: "my-pipeline"
  paths: ["/data/configs/"]
  fileHandlers: {
    yaml: {
      class: "com.mycompany.lucille.filehandler.YamlFileHandler"
      idField: "name"
    }
    csv: {
      # no class needed — uses built-in CSVFileHandler
      separator: ","
    }
  }
}]

When the FileConnector encounters a file ending in .yaml, it passes its content to your YamlFileHandler. Files ending in .csv use the built-in handler. Files with extensions not listed in fileHandlers are skipped.

Skeleton

Every file handler must follow the Javadoc Standards.

package com.mycompany.lucille.filehandler;

import com.kmwllc.lucille.core.Document;
import com.kmwllc.lucille.core.Publisher;
import com.kmwllc.lucille.core.fileHandler.FileHandler;
import com.kmwllc.lucille.core.fileHandler.FileHandlerException;
import com.kmwllc.lucille.core.spec.Spec;
import com.kmwllc.lucille.core.spec.SpecBuilder;
import com.typesafe.config.Config;
import java.io.InputStream;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import org.yaml.snakeyaml.Yaml;

/**
 * Parses YAML files where each YAML document (separated by ---) becomes a Lucille Document.
 * <p>
 * Config Parameters -
 * <ul>
 *   <li>idField (String, Optional) : Field to use as the document ID. Defaults to generating an ID from the file path and index.</li>
 *   <li>docIdPrefix (String, Optional) : Prefix to prepend to document IDs. Defaults to empty string.</li>
 * </ul>
 */
public class YamlFileHandler implements FileHandler {

  public static final Spec SPEC = SpecBuilder.fileHandler()
      .optionalString("idField", "docIdPrefix")
      .build();

  private final String idField;
  private final String docIdPrefix;

  public YamlFileHandler(Config config) {
    this.idField = config.hasPath("idField") ? config.getString("idField") : null;
    this.docIdPrefix = config.hasPath("docIdPrefix") ? config.getString("docIdPrefix") : "";
  }

  @Override
  public Spec getSpec() {
    return SPEC;
  }

  @Override
  public Iterator<Document> processFile(InputStream inputStream, String pathStr)
      throws FileHandlerException {
    Yaml yaml = new Yaml();
    // loadAll returns an Iterable of parsed YAML documents (each as a Map)
    Iterator<Object> yamlDocs = yaml.loadAll(inputStream).iterator();

    return new Iterator<>() {
      private int index = 0;

      @Override
      public boolean hasNext() {
        return yamlDocs.hasNext();
      }

      @Override
      public Document next() {
        Object raw = yamlDocs.next();
        if (!(raw instanceof Map)) {
          throw new RuntimeException("YAML document at index " + index + " in " + pathStr
              + " is not a mapping");
        }

        @SuppressWarnings("unchecked")
        Map<String, Object> fields = (Map<String, Object>) raw;

        // Determine document ID
        String id;
        if (idField != null && fields.containsKey(idField)) {
          id = docIdPrefix + fields.get(idField).toString();
        } else {
          id = docIdPrefix + pathStr + "-" + index;
        }

        Document doc = Document.create(id);
        for (Map.Entry<String, Object> entry : fields.entrySet()) {
          doc.setField(entry.getKey(), entry.getValue().toString());
        }

        index++;
        return doc;
      }
    };
  }

  @Override
  public void processFileAndPublish(Publisher publisher, InputStream inputStream, String pathStr)
      throws FileHandlerException {
    Iterator<Document> docs = processFile(inputStream, pathStr);
    while (docs.hasNext()) {
      try {
        publisher.publish(docs.next());
      } catch (Exception e) {
        throw new FileHandlerException("Error publishing document from " + pathStr, e);
      }
    }
  }
}

Guidelines

• Return a lazy iterator — Don’t load the entire file into memory. Parse incrementally when the format supports it (YAML’s loadAll, JSON streaming, etc.).
• Close resources when exhausted — If your iterator opens readers or parsers, close them when hasNext() returns false or when an exception is thrown.
• Use pathStr for IDs and logging — The path string identifies the file for error messages and can be used as part of document ID generation. Don’t use it to open the file — the InputStream is already open.
• Handle malformed input gracefully — Throw FileHandlerException with a descriptive message rather than letting raw parsing exceptions propagate.
• processFileAndPublish can usually delegate — The pattern shown above (iterate processFile() and publish each document) works for most handlers. Override with a custom implementation only if you need to manage resources differently or publish in batches.

Unit Testing

FileHandler tests follow a simple pattern: construct the handler with a config, open a test file as an InputStream, call processFile(), and assert on the returned documents.

public class YamlFileHandlerTest {

  @Test
  public void testBasicParsing() throws Exception {
    Config config = ConfigFactory.parseMap(Map.of("idField", "name"));
    YamlFileHandler handler = new YamlFileHandler(config);

    InputStream input = getClass().getClassLoader()
        .getResourceAsStream("YamlFileHandlerTest/sample.yaml");
    Iterator<Document> docs = handler.processFile(input, "sample.yaml");

    assertTrue(docs.hasNext());
    Document first = docs.next();
    assertEquals("my-service", first.getString("name"));
    assertEquals("8080", first.getString("port"));
  }
}

Place test YAML files under src/test/resources/YamlFileHandlerTest/.

For full pipeline integration tests, see Testing Pipelines.

Custom File Content Fetchers

Several built-in stages (ApplyFileHandlers, FetchFileContent, TextExtractor, ExtractEntities) need to read file content from a path stored on a document. They do this through a FileContentFetcher — an interface that resolves a path string to an InputStream. The default implementation handles local files, classpath resources, and cloud storage (S3, Azure, GCS) transparently.

If you need custom path resolution logic — for example, looking up a path in a database, fetching content from a proprietary CMS API, or interpreting paths relative to a document’s metadata — you can provide a custom FileContentFetcher implementation.

How it works

Any stage that uses FileContentFetcher.create(config) supports the fetcherClass config property. When present, the factory instantiates your class instead of the default:

{
  class: "com.kmwllc.lucille.stage.ApplyFileHandlers"
  filePathField: "file_path"
  fetcherClass: "com.mycompany.lucille.fetcher.CmsFetcher"
  cmsUrl: "https://cms.example.com/api"
  cmsToken: ${CMS_TOKEN}
  fileHandlers: { pdf: {} }
}

The entire stage config is passed to your fetcher’s constructor, so you can read any additional properties you need (like cmsUrl and cmsToken above).

The interface

public interface FileContentFetcher {

  void startup() throws IOException;

  void shutdown();

  InputStream getInputStream(String path) throws IOException;

  InputStream getInputStream(String path, Document doc) throws IOException;

  BufferedReader getReader(String path) throws IOException;

  BufferedReader getReader(String path, Document doc) throws IOException;

  BufferedReader getReader(String path, String encoding) throws IOException;

  BufferedReader getReader(String path, String encoding, Document doc) throws IOException;

  int countLines(String path) throws IOException;

  int countLines(String path, Document doc) throws IOException;
}

The Document-accepting overloads allow your fetcher to make decisions based on document metadata — for example, using a field on the document to determine which storage system to query.

Lifecycle

• startup() is called once when the stage’s start() method runs (once per worker thread). Open connections here.
• shutdown() is called when the stage’s stop() method runs. Close connections here.
• getInputStream() / getReader() are called per document during processDocument().

Skeleton

package com.mycompany.lucille.fetcher;

import com.kmwllc.lucille.core.Document;
import com.kmwllc.lucille.util.FileContentFetcher;
import com.typesafe.config.Config;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;

/**
 * Fetches file content from a CMS API. The path is interpreted as a CMS asset ID.
 */
public class CmsFetcher implements FileContentFetcher {

  private final String cmsUrl;
  private final String cmsToken;

  public CmsFetcher(Config config) {
    this.cmsUrl = config.getString("cmsUrl");
    this.cmsToken = config.getString("cmsToken");
  }

  @Override
  public void startup() throws IOException {
    // Validate connectivity if needed
  }

  @Override
  public void shutdown() {
    // Close any persistent connections
  }

  @Override
  public InputStream getInputStream(String path) throws IOException {
    URL url = new URL(cmsUrl + "/assets/" + path + "/content");
    HttpURLConnection conn = (HttpURLConnection) url.openConnection();
    conn.setRequestProperty("Authorization", "Bearer " + cmsToken);
    return conn.getInputStream();
  }

  @Override
  public InputStream getInputStream(String path, Document doc) throws IOException {
    // Could use doc metadata to determine the CMS tenant, version, etc.
    return getInputStream(path);
  }

  @Override
  public BufferedReader getReader(String path) throws IOException {
    return new BufferedReader(new InputStreamReader(getInputStream(path), "utf-8"));
  }

  @Override
  public BufferedReader getReader(String path, Document doc) throws IOException {
    return new BufferedReader(new InputStreamReader(getInputStream(path, doc), "utf-8"));
  }

  @Override
  public BufferedReader getReader(String path, String encoding) throws IOException {
    return new BufferedReader(new InputStreamReader(getInputStream(path), encoding));
  }

  @Override
  public BufferedReader getReader(String path, String encoding, Document doc) throws IOException {
    return new BufferedReader(new InputStreamReader(getInputStream(path, doc), encoding));
  }

  @Override
  public int countLines(String path) throws IOException {
    try (BufferedReader reader = getReader(path)) {
      int lines = 0;
      while (reader.readLine() != null) lines++;
      return lines;
    }
  }

  @Override
  public int countLines(String path, Document doc) throws IOException {
    try (BufferedReader reader = getReader(path, doc)) {
      int lines = 0;
      while (reader.readLine() != null) lines++;
      return lines;
    }
  }
}

Stages that support fetcherClass

Any stage that calls FileContentFetcher.create(config) in its constructor supports this extension point:

• ApplyFileHandlers — applies FileHandlers to content fetched from a path field
• FetchFileContent — fetches raw file content into a document field
• TextExtractor (lucille-tika plugin) — extracts text from binary files
• ExtractEntities — loads entity dictionaries from file paths

What’s in config

Your fetcher’s constructor receives the full stage config — the same Config object the stage itself received. This means you can add custom properties alongside the stage’s own properties. Your fetcher reads what it needs; the stage reads what it needs; the SPEC validates both (add your fetcher’s properties to the stage’s SPEC, or accept that they’ll be flagged as unknown unless you also declare them).

In practice, stages that support fetcherClass include it in their SPEC via FileContentFetcher.SPEC, which declares fetcherClass as an optional string. Additional properties specific to your fetcher implementation are not validated by the stage’s SPEC — they will be flagged as unknown properties unless the stage’s SPEC is extended to include them. This is a known limitation; a future improvement may allow fetcher-specific SPEC declarations.

7 - Developing Storage Clients

A StorageClient traverses a storage system, discovers files, and provides access to their content. The FileConnector uses StorageClients to list and read files from local disk, S3, Azure Blob Storage, and Google Cloud Storage. If you need to read files from a storage system that Lucille doesn’t support out of the box (e.g., SFTP, SMB/CIFS, WebDAV), you implement a StorageClient.

Note: It is not currently possible to reference a custom StorageClient from a Lucille config. The mapping from URI scheme to StorageClient implementation is hardcoded in StorageClient.create(). Adding a new storage backend today requires a PR to lucille-core — see the Contributor Guide for project structure and contribution workflow. Config-driven pluggable StorageClients are a roadmap feature. This page is included in the Component Developer Guide because the implementation pattern is the same regardless of how the client is registered.

How StorageClients Fit In

The FileConnector determines which StorageClient to use based on the URI scheme of each path in its paths config:

When you add a new StorageClient, you add a new case to this mapping for your URI scheme (e.g., sftp).

What BaseStorageClient Does for You

To create a StorageClient, extend BaseStorageClient. The base class manages the heavy lifting so you only provide the storage-specific operations:

• Lifecycle management — Tracks initialization state and ensures init() is called before traversal.
• File filtering — Applies include/exclude patterns, file size limits, and modification time cutoffs from TraversalParams.
• Archive handling — Automatically detects and extracts .zip, .tar, .tar.gz, and .gz files, processing their contents through the appropriate FileHandlers.
• Compressed file handling — Decompresses .gz and .bz2 files transparently before passing to FileHandlers.
• FileHandler delegation — Routes files to the correct FileHandler based on extension and publishes the resulting Documents.
• Incremental mode — Integrates with FileConnectorStateManager to track which files have been processed and skip unchanged files on subsequent runs.
• Success/error directory handling — Moves files to configured success or error directories after processing.

What You Implement

What’s in config

The Config passed to your StorageClient constructor is the cloud/storage-provider-specific config block extracted from the connector config. For example, if the connector config contains:

sftp: {
  host: "files.example.com"
  port: 22
  username: "ingest"
  privateKeyPath: "/home/ingest/.ssh/id_rsa"
}

…your constructor receives a Config containing host, port, username, and privateKeyPath. It does not contain the connector config or the full Lucille config.

FileReference

When traversing, you call processAndPublishFileIfValid() for each file. This method expects a FileReference — an object that describes a file’s metadata and provides access to its content. Extend BaseFileReference for your storage system:

public class SftpFileReference extends BaseFileReference {

  private final ChannelSftp channel;
  private final String remotePath;

  public SftpFileReference(LsEntry entry, String basePath, ChannelSftp channel, TraversalParams params) {
    super(
        URI.create("sftp://" + channel.getSession().getHost() + basePath + entry.getFilename()),
        entry.getAttrs().getMTime(),   // last modified (epoch seconds)
        entry.getAttrs().getSize(),    // file size
        null                           // creation time (not available via SFTP)
    );
    this.channel = channel;
    this.remotePath = basePath + entry.getFilename();
  }

  @Override
  public String getName() {
    return remotePath;
  }

  @Override
  public boolean isValidFile() {
    return !remotePath.endsWith("/");
  }

  @Override
  public InputStream getContentStream(TraversalParams params) {
    try {
      return channel.get(remotePath);
    } catch (SftpException e) {
      throw new RuntimeException("Failed to open " + remotePath, e);
    }
  }

  @Override
  protected byte[] getFileContent(TraversalParams params) {
    try (InputStream is = getContentStream(params)) {
      return is.readAllBytes();
    } catch (Exception e) {
      throw new RuntimeException("Failed to read " + remotePath, e);
    }
  }
}

Skeleton

package com.mycompany.lucille.storage;

import com.jcraft.jsch.*;
import com.kmwllc.lucille.connector.FileConnectorStateManager;
import com.kmwllc.lucille.connector.storageclient.BaseStorageClient;
import com.kmwllc.lucille.connector.storageclient.TraversalParams;
import com.kmwllc.lucille.core.Publisher;
import com.typesafe.config.Config;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.Vector;

/**
 * A StorageClient for SFTP servers. Traverses a remote directory and publishes files
 * through the standard FileHandler pipeline.
 */
public class SftpStorageClient extends BaseStorageClient {

  private Session session;
  private ChannelSftp channel;

  public SftpStorageClient(Config config) {
    super(config);
  }

  @Override
  protected void validateOptions(Config config) {
    if (!config.hasPath("host")) {
      throw new IllegalArgumentException("SFTP StorageClient requires 'host' in config.");
    }
    if (!config.hasPath("username")) {
      throw new IllegalArgumentException("SFTP StorageClient requires 'username' in config.");
    }
  }

  @Override
  protected void initializeStorageClient() throws IOException {
    try {
      String host = config.getString("host");
      int port = config.hasPath("port") ? config.getInt("port") : 22;
      String username = config.getString("username");

      JSch jsch = new JSch();
      if (config.hasPath("privateKeyPath")) {
        jsch.addIdentity(config.getString("privateKeyPath"));
      }

      session = jsch.getSession(username, host, port);
      session.setConfig("StrictHostKeyChecking", "no");

      if (config.hasPath("password")) {
        session.setPassword(config.getString("password"));
      }

      session.connect();
      channel = (ChannelSftp) session.openChannel("sftp");
      channel.connect();
    } catch (JSchException e) {
      throw new IOException("Failed to connect to SFTP server", e);
    }
  }

  @Override
  protected void shutdownStorageClient() throws IOException {
    if (channel != null) channel.disconnect();
    if (session != null) session.disconnect();
  }

  @Override
  protected void traverseStorageClient(Publisher publisher, TraversalParams params,
      FileConnectorStateManager stateMgr) throws Exception {
    String remotePath = params.getURI().getPath();

    @SuppressWarnings("unchecked")
    Vector<ChannelSftp.LsEntry> entries = channel.ls(remotePath);

    for (ChannelSftp.LsEntry entry : entries) {
      if (entry.getFilename().startsWith(".")) continue;
      if (entry.getAttrs().isDir()) continue;

      SftpFileReference ref = new SftpFileReference(entry, remotePath, channel, params);
      processAndPublishFileIfValid(publisher, ref, params, stateMgr);
    }
  }

  @Override
  protected InputStream getFileContentStreamFromStorage(URI uri) throws IOException {
    try {
      return channel.get(uri.getPath());
    } catch (SftpException e) {
      throw new IOException("Failed to read " + uri, e);
    }
  }

  @Override
  public void moveFile(URI filePath, URI folder) throws IOException {
    try {
      String source = filePath.getPath();
      String destDir = folder.getPath();
      String fileName = source.substring(source.lastIndexOf('/') + 1);
      channel.rename(source, destDir + "/" + fileName);
    } catch (SftpException e) {
      throw new IOException("Failed to move " + filePath + " to " + folder, e);
    }
  }
}