Connectors

• 1: File Connector
• 2: Database Connector
• 3: Solr Connector
• 4: Kafka Connector
• 5: Parquet Connector
• 6: RSS Connector

{
  name: "my-connector"
  class: "com.kmwllc.lucille.connector.FileConnector"
  pipeline: "my-pipeline"
  docIdPrefix: "files-"
  paths: ["/data/files"]
}

connectors: [
  { name: "parent-docs",  class: "...", pipeline: "pipeline1" },
  { name: "child-docs",   class: "...", pipeline: "pipeline1" }
]

1 - File Connector

Source Code

The FileConnector traverses a file system and publishes a Lucille Document for each file it encounters. It supports local filesystems, Amazon S3, Azure Blob Storage, and Google Cloud Storage through a unified interface — a single connector config can traverse paths across multiple providers simultaneously. Optional File Handlers extract structured Documents from files that themselves contain data (CSV rows, JSON objects, XML elements).

Cloud Storage Configuration

When traversing cloud storage, provide authentication under the appropriate top-level config block alongside your connector config. Each provider also accepts an optional maxNumOfPages to limit how many file listings are loaded into memory per request.

Azure

azure {
  connectionString: "DefaultEndpointsProtocol=https;AccountName=..."
  # or:
  accountName: "myaccount"
  accountKey: "mykey"
  maxNumOfPages: 100
}

You must provide either connectionString, or both accountName and accountKey.

Google Cloud Storage

gcp {
  pathToServiceKey: "/path/to/service-account.json"
  maxNumOfPages: 100
}

Amazon S3

s3 {
  accessKeyId: "AKIA..."
  secretAccessKey: "..."
  region: "us-east-1"
  maxNumOfPages: 100
}

For S3 paths, percent-encode special characters in paths (e.g., s3://bucket/folder%20with%20spaces).

A single connector can traverse multiple paths across providers:

paths: ["file:///local/data", "s3://my-bucket/prefix", "az://container/path"]

The URI scheme selects the appropriate storage backend automatically. A local filesystem client is always available without additional configuration.

File Handlers

File Handlers process individual files and extract one or more Lucille Documents from their contents. Without a handler, the FileConnector publishes one document per file containing only the file metadata fields. With a handler, a CSV file becomes one document per row, a JSON file becomes one document per object, and so on.

File Handlers are configured under the fileHandlers block. The key (csv, json, xml) determines which handler applies to files with that extension. fileHandlers and fileOptions are two separate top-level keys in the connector config block — they are not nested inside each other.

fileHandlers: {
  csv {
    separatorChar: "|"
    docIdPrefix: "csv-"
  }
  json {}
  xml {
    chunkPath: "//record"
  }
}

All File Handlers support docIdPrefix (prepended to every generated Document ID).

CSV File Handler

com.kmwllc.lucille.core.fileHandler.CSVFileHandler

Extracts one Document per row from a CSV file.

fileHandlers: {
  csv {
    idField: "article_id"
    separatorChar: "|"
    filenameField: "source_file"
    lowercaseFields: true
    docIdPrefix: "article-"
  }
}

JSON File Handler

com.kmwllc.lucille.core.fileHandler.JSONFileHandler

Extracts one Document per JSON object. Supports both standard JSON files (a single object or array) and JSON Lines (.jsonl) format where each line is a separate JSON object.

fileHandlers: {
  json {
    idField: "doc_id"
    blacklist: ["internal_metadata", "_rev"]
    docIdPrefix: "doc-"
  }
}

XML File Handler

com.kmwllc.lucille.core.fileHandler.XMLFileHandler

Extracts Documents from an XML file by selecting elements matching an XPath expression. Each matched element becomes a Document; element child text content and attributes become Document fields.

fileHandlers: {
  xml {
    chunkPath: "//product"
  }
}

Video File Handler

Available via the lucille-video plugin (com.kmwllc:lucille-video). Requires FFmpeg to be installed on the system.

Extracts frames from video files as individual Lucille Documents. See the plugin documentation for configuration parameters and supported formats.

Custom File Handlers

To process a file format not covered by the built-in handlers:

1. Extend BaseFileHandler.
2. Declare public static final Spec SPEC = SpecBuilder.fileHandler()... (required).
3. Implement Iterator<Document> processFile(InputStream stream, String pathStr).

Reference the handler by its fully-qualified class name in the fileHandlers block. The class field can also override the built-in handler for a standard extension:

fileHandlers: {
  csv {
    class: "com.example.MyCustomCSVHandler"
    myCustomParam: "value"
  }
}

File Handlers accept an InputStream and return Documents via an Iterator. The InputStream and any underlying resources are closed when the Iterator’s hasNext() returns false. When working directly with File Handlers in code, always exhaust the returned Iterator.

File Options

File options control traversal behaviour — how the connector handles the files it finds.

When archive or compressed file handling is enabled, entries inside archives get composite paths using the ! separator:

s3://bucket/archive.zip!path/inside/archive/file.csv

Modification and publish cutoffs apply to both the container archive and its entries.

Filter Options

Filter options control which files are processed and published. All filter options are optional. When multiple options are specified, a file must satisfy all of them to be processed. Evaluation order:

1. File name must match at least one includes pattern (if any are specified).
2. File name must not match any excludes pattern.
3. File’s modification time must fall within lastModifiedCutoff (if specified).
4. File must not have been published within lastPublishedCutoff (if specified — requires state configuration).
5. In incremental publish mode, only new or modified files since the last run are published (requires state configuration).

Example using regex patterns and time-based filtering:

filterOptions: {
  includes: [".*\\.pdf$", ".*\\.docx$"]
  excludes: [".*\\.tmp$", ".*~$"]
  lastModifiedCutoff: "24h"
  lastPublishedCutoff: "7d"
  publishMode: "incremental"
}

Note on pathsToSkip: while you could create an excludes regex detailing directories you do not want to publish files for, there are performance benefits to using pathsToSkip instead. With this parameter, the FileConnector will not actually traverse the path and its contents. If you use excludes, the path will be traversed, but the individual documents are not processed & published.

Each entry may be a URI with a scheme (for example s3://bucket/path, gs://bucket/path, or file:///path/to/dir). As a convenience, local file system paths may also be provided without a scheme - either absolute (e.g. /path/to/dir) or relative to the working directory (e.g. path/to/dir) - and will be resolved to an absolute file:// path.

Incremental Mode and State

The FileConnector can persist state to a JDBC-compatible database to track which files have been published and when. State enables lastPublishedCutoff, incremental publish mode, and tombstone detection.

Configure a state database alongside the connector:

state {
  enabled: true                    # Set to false to disable state without removing the block
  driver: "org.h2.Driver"          # Default: embedded H2
  connectionString: "jdbc:h2:./lucille-state"
  jdbcUser: ""
  jdbcPassword: ""
  tableName: "file_state"          # Defaults to the connector name
  performDeletions: true
  pathLength: 200                  # Max length of the file path column
  runsBeforeExpiration: 1          # Consecutive absent runs before tombstone (default: 1)
}

If connectionString is omitted, an embedded H2 database is created at ./state/{CONNECTOR_NAME}.

A few constraints to be aware of when using state:

• Files that are moved or renamed will not have lastPublishedCutoff applied — their new path is not recognised as previously published.
• Similarly, remember that the files in a directory that is skipped (by pathsToSkip), since they are not traversed, are also not tracked at all during a stateful run.
• Capitalise directory names in paths consistently across runs. State lookups are case-sensitive.
• Each database table should be used by only one connector configuration. Sharing a table across connectors will corrupt state. This can happen in several ways: reusing the same explicit tableName across multiple connectors in the same config file; running two separate configs concurrently where both reference the same tableName; or omitting tableName in two different configs that happen to use the same connector name (since tableName defaults to the connector name). In all cases, two connectors writing to the same state table will overwrite each other’s records, leading to incorrect incremental behavior — files may be skipped or reprocessed unexpectedly.

Tombstone Generation

When filterOptions.sendTombstones: true is set (requires incremental mode and state), the connector detects files that existed in the previous run but are no longer present. For each deleted file it publishes a tombstone document — a document with file_expired: true and ___skipped: true. The skipped flag causes the tombstone to bypass all pipeline stages. Downstream, configure the Indexer to issue a delete when it encounters the marker field:

indexer {
  type: "solr"
  deletionMarkerField: "file_expired"
  deletionMarkerFieldValue: "true"
}

By default, a file is considered expired after being absent from a single run. You can increase this threshold with state.runsBeforeExpiration to provide a safety margin against transient storage issues (e.g., a temporary listing failure):

state {
  runsBeforeExpiration: 3   # File must be absent for 3 consecutive runs before tombstone
}

The value must be at least 1. When set to 1 (default), a file missing from one run immediately triggers a tombstone — the original behavior.

This parameter only applies in incremental mode.

Document Fields

Every document published by the FileConnector carries these fields:

When a File Handler (CSV, JSON, XML) processes a file, it produces child documents with their own fields rather than a single file-level document.

Implementation Notes

This section is for Component Developers implementing new connectors. Pipeline Authors can stop here.

StorageClient Pattern

The connector handles four storage backends through a StorageClient abstraction. During traversal, the appropriate client is selected by URI scheme:

private void traverseStoragePath(Publisher publisher, URI pathToTraverse) throws ConnectorException {
    String clientKey = pathToTraverse.getScheme() != null ? pathToTraverse.getScheme() : "file";
    StorageClient storageClient = storageClientMap.get(clientKey);
    // ...
    storageClient.traverse(publisher, params, stateManager);
}

StorageClient is an interface with implementations for each backend. Each implementation handles listing/paginating files, applying filter criteria, reading file content, publishing documents, and updating state. The createClients(config) factory method inspects the config for cloud provider blocks and instantiates the appropriate clients.

Lifecycle: execute → close

Initialisation of storage clients and the state manager is deferred to inside execute(), not the constructor. This avoids opening network connections before the connector actually runs:

@Override
public void execute(Publisher publisher) throws ConnectorException {
    initialize();  // Init storage clients + state manager

    for (URI resource : storageURIs) {
        traverseStoragePath(publisher, resource);
    }

    if (sendTombstones) {
        sendExpiredFileTombstones(publisher);
    }
}

@Override
public void close() {
    if (stateManager != null) stateManager.shutdown();
    for (StorageClient client : storageClientMap.values()) {
        client.shutdown();
    }
}

SPEC Declaration

The FileConnector SPEC demonstrates how to declare a complex, nested configuration:

public static final Spec SPEC = SpecBuilder.connector()
    .requiredList("paths", new TypeReference<List<String>>(){})
    .optionalParent(
        SpecBuilder.parent("filterOptions")
            .optionalList("includes", new TypeReference<List<String>>(){})
            .optionalList("excludes", new TypeReference<List<String>>(){})
            .optionalString("lastModifiedCutoff", "lastPublishedCutoff", "publishMode", "sendTombstones").build(),
        SpecBuilder.parent("fileOptions")
            .optionalBoolean("getFileContent", "handleArchivedFiles", "handleCompressedFiles")
            .optionalString("moveToAfterProcessing", "moveToErrorFolder").build(),
        SpecBuilder.parent("state")
            .optionalString("driver", "connectionString", "jdbcUser", "jdbcPassword", "tableName")
            .optionalBoolean("performDeletions")
            .optionalNumber("pathLength").build(),
        GCP_PARENT_SPEC,
        AZURE_PARENT_SPEC,
        S3_PARENT_SPEC)
    .optionalParent("fileHandlers", new TypeReference<Map<String, Map<String, Object>>>(){})
    .build();

Key patterns: SpecBuilder.connector() provides base properties; nested SpecBuilder.parent(...) blocks define each optional config section; cloud provider specs are defined as reusable constants; fileHandlers uses a TypeReference because its structure is dynamic (keys are file extensions).

Reusable Patterns for Other Complex Connectors

1. StorageClient pattern — abstract the data source behind an interface; select implementation by URI scheme or config key.
2. Deferred initialisation — don’t open connections in the constructor; do it in execute().
3. Constraint validation early — check for incompatible config combinations (e.g., multiple paths + moveToAfterProcessing) in the constructor so failures surface before any traversal begins.
4. Nested SPEC declarations — group related config into parent blocks; define cloud provider specs as reusable constants shared across connectors.
5. JDBC-backed state — make state optional; check config.hasPath("state") before constructing the state manager.
6. Filter pipeline — layer multiple filter criteria (regex, time-based, state-based) with a clear evaluation order.
7. Tombstone generation — detect deletions by comparing the current file set to the previous state snapshot; mark documents with a field the Indexer can act on.
8. FileHandler delegation — use pluggable handlers (with their own SPECs) for format-specific processing rather than embedding format logic in the connector.

2 - Database Connector

Source Code

The DatabaseConnector reads rows from any JDBC-compatible relational database and publishes each row as a Lucille Document. Column names become field names on the Document.

Basic Configuration

connectors: [
  {
    name: "db-connector"
    class: "com.kmwllc.lucille.connector.jdbc.DatabaseConnector"
    pipeline: "my-pipeline"

    driver: "org.postgresql.Driver"
    connectionString: "jdbc:postgresql://localhost:5432/mydb"
    jdbcUser: "username"
    jdbcPassword: ${?DB_PASSWORD}
    sql: "SELECT id, title, body, published_at FROM articles WHERE active = true"
    idField: "id"
  }
]

Configuration Parameters

Pre and Post SQL

Use preSQL and postSQL to run setup and teardown logic that must happen before and after the main query:

{
  name: "db-connector"
  class: "com.kmwllc.lucille.connector.jdbc.DatabaseConnector"
  pipeline: "my-pipeline"

  driver: "org.postgresql.Driver"
  connectionString: "jdbc:postgresql://localhost:5432/mydb"
  jdbcUser: ${?DB_USER}
  jdbcPassword: ${?DB_PASSWORD}

  preSQL: "CREATE TEMP TABLE export_snapshot AS SELECT * FROM articles WHERE active = true"
  sql: "SELECT id, title, body FROM export_snapshot ORDER BY id"
  postSQL: "DROP TABLE IF EXISTS export_snapshot"
  idField: "id"
}

postSQL runs only if preSQL and the main query both succeed. If the connector throws, postSQL is skipped but close() is always called.

Multi-Query Joins

otherSQLs allows you to enrich primary rows with data from additional queries at read time, without a SQL JOIN. Each secondary query must be ordered by its join key (which must be an integer column matching a column in the primary query).

{
  sql: "SELECT id, title FROM articles ORDER BY id"
  idField: "id"
  otherSQLs: ["SELECT article_id, tag_name FROM article_tags ORDER BY article_id"]
  otherJoinFields: ["article_id"]
}

For each primary row, the connector merges matching rows from otherSQLs as multi-valued fields onto the Document.

Incremental Ingest

The DatabaseConnector does not maintain internal state. For incremental ingest (only rows modified since the last run), filter at the SQL level:

SELECT id, title, updated_at FROM articles
WHERE updated_at > TIMESTAMP '2025-01-01 00:00:00'
ORDER BY updated_at ASC

Store the high-water mark externally (e.g., in a config file, environment variable, or the database itself) and substitute it via HOCON variable substitution.

MySQL Streaming

For large MySQL tables, set fetchSize to avoid loading the entire result set into memory:

{
  driver: "com.mysql.cj.jdbc.Driver"
  connectionString: "jdbc:mysql://localhost:3306/mydb?useCursorFetch=true"
  fetchSize: -2147483648
  sql: "SELECT id, title FROM large_table ORDER BY id"
  idField: "id"
}

Common JDBC Drivers

Integration with QueryDatabase Stage

For per-document database enrichment (joining a lookup table for each document mid-pipeline, rather than reading a source table), use the QueryDatabase Stage.

3 - Solr Connector

Source Code

The SolrConnector issues a query against a Solr collection and publishes each result as a Lucille Document. It is useful for cross-index enrichment pipelines — for example, reading documents from one Solr collection, enriching them, and indexing them into a different collection or backend.

Internally, the connector uses cursor-based pagination (cursorMark) to iterate through large result sets without loading the full result into memory. Results are sorted by idField ascending (required for cursor pagination).

Basic Configuration

connectors: [
  {
    name: "solr-source"
    class: "com.kmwllc.lucille.connector.SolrConnector"
    pipeline: "my-pipeline"

    solr {
      url: ["http://localhost:8983/solr"]
      useCloudClient: true
      defaultCollection: "source-collection"
    }

    solrParams {
      q: "*:*"
      fl: "id,title,body,author"
      fq: "status:active"
      rows: 100
    }

    idField: "id"
  }
]

Configuration Parameters

solr Connection Parameters

Query Parameters (solrParams)

The solrParams block accepts any standard Solr query parameters:

solrParams {
  q: "status:active AND type:article"
  fl: "id,title,body,published_date"
  fq: ["category:news", "language:en"]
  rows: 500
}

Common parameters:

• q — query string (default: *:* if omitted)
• fl — field list (fields to return; omit for all fields)
• fq — filter query (can be a single string or a list for multiple filters)
• rows — page size for cursor pagination (default: Solr’s default, typically 10)

The connector always adds sort: {idField} asc automatically (required for cursor pagination) and manages cursorMark internally.

Pre and Post Actions

preActions and postActions are lists of Solr update requests sent before and after the main query, respectively. They support the {runId} placeholder, which is replaced with the current run’s UUID at execution time.

The request format is controlled by useXml:

• useXml: false (default): requests are JSON strings
• useXml: true: requests are XML strings

Example: Mark documents as in-progress before reading, mark as done after

connectors: [
  {
    name: "solr-source"
    class: "com.kmwllc.lucille.connector.SolrConnector"
    pipeline: "process-pipeline"

    solr {
      url: ["http://localhost:8983/solr/my-collection"]
    }

    # JSON update — set run_id on all active documents before reading
    preActions: [
      "{\"add\":{\"doc\":{\"id\":\"marker\",\"run_id\":\"{runId}\"}}}"
    ]

    # JSON update — clean up after run
    postActions: [
      "{\"delete\":{\"query\":\"run_id:{runId}\"}}"
    ]

    solrParams {
      q: "status:active"
      fl: "id,title,body"
    }
  }
]

postActions run only if preActions and the main query both succeed. If the connector throws, postActions are skipped.

Cursor-Based Pagination

The connector iterates through results using Solr’s cursorMark mechanism, which avoids deep pagination performance issues. This requires:

1. Documents to be sorted by a unique field (the idField).
2. The idField to be indexed in Solr as a single-valued, non-analyzed field.

For large collections, tune rows in solrParams to balance memory use and round-trip count.

4 - Kafka Connector

Source Code

The KafkaConnector reads Documents from a Kafka topic and publishes them into a Lucille pipeline. This is distinct from Kafka’s role as the messaging layer in distributed mode — the KafkaConnector is a data source, reading documents produced by an upstream system.

Use Cases

• Streaming ingest from Kafka: An upstream application publishes documents (as JSON) to a Kafka topic, and Lucille reads them for enrichment and indexing.
• Connectorless distributed mode: In this deployment pattern, a third-party publisher puts documents onto a Kafka source topic, and Lucille Workers consume them directly. In this case the KafkaConnector is not used — Workers listen to the source topic directly.

Configuration

All Kafka connection parameters are nested under the kafka key within the connector config block.

connectors: [
  {
    name: "kafka-source"
    class: "com.kmwllc.lucille.connector.KafkaConnector"
    pipeline: "my-pipeline"

    kafka.bootstrapServers: "kafka1:9092,kafka2:9092"
    kafka.topic: "my-source-topic"
    kafka.consumerGroupId: "lucille-kafka-connector"
    kafka.clientId: "lucille-consumer-1"
    kafka.maxPollIntervalSecs: 600
    idField: "article_id"
    maxMessages: 10000
  }
]

Configuration Parameters

Message Format

The KafkaConnector expects each Kafka message value to be a JSON object. Each JSON object becomes a Lucille Document. Field names in the JSON map directly to Document field names.

Example Kafka message:

{
  "article_id": "art-001",
  "title": "Breaking News",
  "body": "Full article text...",
  "published_at": "2025-06-01T12:00:00Z"
}

Security

For Kafka clusters with TLS or SASL authentication, use the top-level kafka {} block (separate from the connector’s inline params) to provide properties files and security settings:

kafka {
  bootstrapServers: "kafka1:9092"
  securityProtocol: "SSL"
  consumerPropertyFile: "/path/to/consumer.properties"
  producerPropertyFile: "/path/to/producer.properties"
  adminPropertyFile: "/path/to/admin.properties"
}

securityProtocol, consumerPropertyFile, producerPropertyFile, and adminPropertyFile are properties of the top-level kafka {} block and apply to all Kafka communication in the process, not just the KafkaConnector.

Kafka as the Messaging Layer vs. as a Source

These are two distinct uses of Kafka in Lucille:

Both can be active simultaneously: a KafkaConnector reads data from one topic while Lucille’s distributed messaging uses separate internal topics.

Related

• Topology: Fully Distributed
• Topology: Connector-less Distributed
• Getting Started: Distributed Mode

5 - Parquet Connector

Source Code

The ParquetConnector reads Apache Parquet files — locally or from Amazon S3 — and publishes each row as a Lucille Document. Parquet is a columnar format commonly used to store pre-computed embeddings, feature vectors, and large datasets.

Maven dependency:

<dependency>
  <groupId>com.kmwllc</groupId>
  <artifactId>lucille-parquet</artifactId>
  <version>${lucille.version}</version>
</dependency>

Configuration

connectors: [
  {
    name: "parquet-source"
    class: "com.kmwllc.lucille.parquet.connector.ParquetConnector"
    pipeline: "my-pipeline"
    pathToStorage: "/data/embeddings.parquet"
    idField: "doc_id"
    fsUri: "file:///"
  }
]

Configuration Parameters

S3 Configuration

For S3, provide the filesystem URI and credentials:

connectors: [
  {
    name: "parquet-s3"
    class: "com.kmwllc.lucille.parquet.connector.ParquetConnector"
    pipeline: "my-pipeline"
    pathToStorage: "/prefix/embeddings"
    idField: "doc_id"
    fsUri: "s3a://my-bucket"
    s3Key: ${AWS_ACCESS_KEY_ID}
    s3Secret: ${AWS_SECRET_ACCESS_KEY}
  }
]

Notes

• The connector uses Hadoop’s filesystem abstraction (FileSystem) for path traversal. Unlike FileConnector, it does not use Lucille’s StorageClient infrastructure.
• Parquet files must have the .parquet extension to be processed.
• When paginating with start/limit, it is recommended to use individual Connectors for each Parquet file rather than a directory path.
• The Parquet format requires random-access reads (not sequential streaming), which is why it is implemented as a standalone Connector rather than a FileConnector FileHandler.

6 - RSS Connector

The RSSConnector

The RSSConnector allows you to publish Documents representing the items found in an RSS feed of your choice. Each Document will (optionally) contain fields from the RSS items, like the author, description, title, etc. By default, the Document IDs will be the item’s guid, which should be a unique identifier for the RSS item.

You can configure the RSSConnector to only publish recent RSS items, based on the pubDate found on the items. Also, it can run incrementally, refreshing the RSS feed after a certain amount of time until you manually stop it. The RSSConnector will avoid publishing Documents for the same RSS item more than once.

The Documents published may have any of the following fields, depending on how the RSS feed is structured:

• author (String)
• categories (List<String>)
• comments (List<String>)
• content (String)
• description (String)
• enclosures (List<JsonNode>). Each JsonNode contains:
  • type (String)
  • url (String)
  • May contain length (Long)
• guid (String)
• isPermaLink (Boolean)
• link (String)
• title (String)
• pubDate (Instant)