Worker

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

What the Worker Does

When a Worker starts, it:

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

Per-Thread Pipeline Isolation

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

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

Multiple Workers

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

worker {
  threads: 4
}

Each thread runs its own Pipeline instance concurrently.

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

Configuration

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

  # Maximum time (seconds) between Kafka polls before the worker shuts down
  # (only relevant in Kafka mode; requires exitOnTimeout: true)
  maxProcessingSecs: 600

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

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

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

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

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

Error Handling

Per-Document Failures

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

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

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

Poison Pills

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

Metrics

Each Worker reports Codahale metrics to the shared registry:

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

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

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

Lifecycle Events

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

Running a Worker Standalone

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

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

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

WorkerIndexer

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

How it differs from a standalone Worker:

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

Start a WorkerIndexer:

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

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

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

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

Practical Guide

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

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