Internals

• 1: Messenger Abstraction
• 2: Message Ordering
• 3: Error Handling and Fault Tolerance
• 4: Kafka Integration
• 5: Metrics and Observability

1 - Messenger Abstraction

Overview

The Messenger layer is the architectural seam that makes Lucille deployment-mode independent. The same Connector, Worker, and Indexer code runs identically whether messages flow through in-memory queues, Kafka topics, or a hybrid of both. This is achieved through three interfaces and a factory pattern that selects the right implementation at runtime.

The Three Messenger Interfaces

Each Lucille component has its own messenger interface, exposing only the operations that component needs:

WorkerMessenger

public interface WorkerMessenger {
    Document pollDocToProcess() throws Exception;       // Receive work
    void commitPendingDocOffsets() throws Exception;    // Acknowledge work
    void sendForIndexing(Document document) throws Exception;  // Forward results
    void sendFailed(Document document) throws Exception;       // Dead letter queue
    void sendEvent(Document document, String message, Event.Type type) throws Exception;
    void sendEvent(Event event) throws Exception;       // Lifecycle events
    void close() throws Exception;
}

IndexerMessenger

public interface IndexerMessenger {
    Document pollDocToIndex() throws Exception;         // Receive processed docs
    void sendEvent(Event event) throws Exception;       // Report completion/failure
    void sendEvent(Document document, String message, Event.Type type) throws Exception;
    void close() throws Exception;
    void batchComplete(List<Document> batch) throws Exception;  // Offset feedback
}

PublisherMessenger

public interface PublisherMessenger {
    void initialize(String runId, String pipelineName) throws Exception;
    String getRunId();
    void sendForProcessing(Document document) throws Exception;  // Publish to workers
    Event pollEvent() throws Exception;                          // Receive lifecycle events
    void close();
}

LocalMessenger: One Class, Three Interfaces

LocalMessenger implements all three interfaces using shared LinkedBlockingQueue instances:

public class LocalMessenger implements IndexerMessenger, PublisherMessenger, WorkerMessenger {
    private final BlockingQueue<Event> pipelineEvents = new LinkedBlockingQueue<>();
    private final BlockingQueue<Document> pipelineSource;  // Publisher → Worker
    private final BlockingQueue<Document> pipelineDest;    // Worker → Indexer
}

The data flow:

• sendForProcessing(doc) → puts on pipelineSource
• pollDocToProcess() → takes from pipelineSource
• sendForIndexing(doc) → puts on pipelineDest
• pollDocToIndex() → takes from pipelineDest
• sendEvent(event) → puts on pipelineEvents
• pollEvent() → takes from pipelineEvents

All polls use a 50ms timeout (POLL_TIMEOUT_MS) to allow polling loops to check termination conditions.

Queue capacity is configurable via publisher.queueCapacity (default: 10,000). The put() calls will block if the queue is full, providing natural backpressure.

Key simplifications in local mode:

• commitPendingDocOffsets() is a no-op (no offsets to commit)
• sendFailed() is a no-op (no dead letter queue)
• batchComplete() is a no-op (no offset feedback needed)
• close() is a no-op (queues are garbage collected)

TestMessenger: Recording History

TestMessenger wraps a LocalMessenger and intercepts writes to record message history:

public class TestMessenger implements IndexerMessenger, PublisherMessenger, WorkerMessenger {
    private final LocalMessenger messenger;
    private List<Event> savedEventMessages = Collections.synchronizedList(new ArrayList<>());
    private List<Document> savedSourceMessages = Collections.synchronizedList(new ArrayList<>());
    private List<Document> savedDestMessages = Collections.synchronizedList(new ArrayList<>());
}

The interception points:

• sendForProcessing(doc) → saves to savedSourceMessages, then delegates
• sendForIndexing(doc) → saves to savedDestMessages, then delegates
• sendEvent(event) → saves to savedEventMessages, then delegates

After a test run, you can inspect:

• getDocsSentForProcessing() — documents the connector published
• getDocsSentForIndexing() — documents that completed pipeline processing
• getSentEvents() — all lifecycle events (CREATE, FINISH, FAIL, DROP)

The lists are synchronizedList because Worker and Indexer threads write concurrently.

KafkaWorkerMessenger: Full Kafka Mode

Reads documents from a Kafka source topic, writes processed documents to a dest topic, and sends events to an event topic:

public class KafkaWorkerMessenger implements WorkerMessenger {
    private final Consumer<String, KafkaDocument> sourceConsumer;
    private final KafkaProducer<String, Document> kafkaDocumentProducer;
    private final KafkaProducer<String, String> kafkaEventProducer;
}

Key behaviors:

• pollDocToProcess() — polls the source topic with KafkaUtils.POLL_INTERVAL (2 seconds). Returns at most one record per poll (MAX_POLL_RECORDS_CONFIG = 1). Sets Kafka metadata on the returned KafkaDocument.
• commitPendingDocOffsets() — calls sourceConsumer.commitSync(). Offsets are committed synchronously to minimize reprocessing after crashes.
• sendForIndexing(doc) — produces to the dest topic using the document ID as the Kafka key. Calls .get() to wait for acknowledgment, then flush().
• sendFailed(doc) — produces to the fail topic (dead letter queue).
• sendEvent(event) — serializes the event as JSON string and produces to the event topic.

KafkaIndexerMessenger: Consume and Commit

Reads processed documents from the dest topic:

public class KafkaIndexerMessenger implements IndexerMessenger {
    private final Consumer<String, KafkaDocument> destConsumer;
    private final KafkaProducer<String, String> kafkaEventProducer;
}

Key difference from the Worker: offsets are committed immediately after polling, before the document is indexed:

public Document pollDocToIndex() throws Exception {
    ConsumerRecords<String, KafkaDocument> consumerRecords = destConsumer.poll(KafkaUtils.POLL_INTERVAL);
    if (consumerRecords.count() > 0) {
        destConsumer.commitSync();  // Commit immediately
        // return the document
    }
    return null;
}

This means a document might be indexed twice if the indexer crashes after committing but before sending the FINISH event. This is acceptable because indexing is idempotent (upsert semantics).

HybridWorkerMessenger: Kafka In, Memory Out

The hybrid mode reads from Kafka but writes to an in-memory queue shared with a co-located Indexer:

public class HybridWorkerMessenger implements WorkerMessenger {
    private final Consumer<String, KafkaDocument> sourceConsumer;
    private final KafkaProducer<String, String> kafkaEventProducer;
    private final LinkedBlockingQueue<Document> pipelineDest;           // Shared with Indexer
    private final LinkedBlockingQueue<Map<TopicPartition, OffsetAndMetadata>> offsets;  // From Indexer
}

Key behaviors:

• pollDocToProcess() — reads from Kafka (same as KafkaWorkerMessenger)
• sendForIndexing(doc) — puts on the shared pipelineDest queue (in-memory, not Kafka)
• commitPendingDocOffsets() — drains the offsets queue and commits each batch of offsets to Kafka. These offsets come from the Indexer after it successfully indexes documents.

The offset feedback loop:

1. Worker polls document from Kafka (records topic/partition/offset)
2. Worker processes document, puts on pipelineDest
3. Indexer picks up document, indexes it, puts offset info on offsets queue
4. Worker’s next commitPendingDocOffsets() call commits those offsets to Kafka

This ensures offsets are only committed after documents are actually indexed, providing at-least-once delivery guarantees without requiring the Indexer to have direct Kafka access.

HybridIndexerMessenger: Memory In, Offsets Out

The counterpart to HybridWorkerMessenger:

public class HybridIndexerMessenger implements IndexerMessenger {
    private final LinkedBlockingQueue<Document> pipelineDest;
    private final LinkedBlockingQueue<Map<TopicPartition, OffsetAndMetadata>> offsets;
    private final KafkaProducer<String, String> kafkaEventProducer;
    private final Set idSet;  // Optional: tracks unique indexed doc IDs
}

Key behaviors:

• pollDocToIndex() — takes from the shared pipelineDest queue (50ms timeout)
• sendEvent(event) — produces to the Kafka event topic (same as other Kafka messengers). Also adds the doc ID to idSet if configured.
• batchComplete(batch) — extracts Kafka metadata from each KafkaDocument in the batch, builds an offset map, and puts it on the offsets queue for the Worker to commit:

public void batchComplete(List<Document> batch) throws InterruptedException {
    Map<TopicPartition, OffsetAndMetadata> batchOffsets = new HashMap<>();
    for (Document doc : batch) {
        if (!(doc instanceof KafkaDocument)) continue;
        KafkaDocument kDoc = (KafkaDocument) doc;
        TopicPartition tp = new TopicPartition(kDoc.getTopic(), kDoc.getPartition());
        OffsetAndMetadata offset = new OffsetAndMetadata(kDoc.getOffset() + 1);  // +1 per Kafka convention
        batchOffsets.put(tp, offset);
    }
    if (!batchOffsets.isEmpty()) {
        offsets.put(batchOffsets);
    }
}

The MessengerFactory Pattern

Each messenger type has a factory interface:

public interface WorkerMessengerFactory {
    WorkerMessenger create();

    static WorkerMessengerFactory getConstantFactory(WorkerMessenger messenger) {
        return () -> messenger;  // Always returns the same instance
    }

    static WorkerMessengerFactory getKafkaFactory(Config config, String pipelineName) {
        return () -> new KafkaWorkerMessenger(config, pipelineName);  // New instance each time
    }
}

The getConstantFactory is used for LOCAL/TEST modes where a single messenger instance is shared. The getKafkaFactory creates a new messenger (with its own Kafka consumer) for each Worker thread — necessary because Kafka consumers are not thread-safe.

How the Runner Uses Factories

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

Architectural Significance

The Messenger abstraction is what makes Lucille’s deployment flexibility possible:

• Development/Testing: Use LocalMessenger or TestMessenger — no external dependencies, fast, inspectable
• Single-node production: Use LocalMessenger — all components in one JVM, no Kafka overhead
• Distributed production: Use Kafka messengers — Workers and Indexers can scale independently across machines
• Hybrid: Use Hybrid messengers — read from Kafka for distribution, but avoid Kafka overhead for the Worker→Indexer hop within the same JVM

The components (Worker, Indexer, Publisher) never know which messenger implementation they’re using. They code against the interface, and the Runner wires in the appropriate implementation at startup.

2 - Message Ordering

Why Ordering Matters

Search ingestion often involves sequences of operations on the same document: a create, followed by one or more updates, possibly followed by a delete. If these operations are reordered, the final state of the search index is wrong. For example, if a sequence of create → update → delete is reordered so that the delete is processed before the update, the document survives in the index when it should have been removed. Or if the create arrives after the delete, the document reappears.

Lucille guarantees that operations on the same document ID are processed and indexed in the order they were published, even when multiple Workers and Indexers run concurrently.

The Mechanism: Document ID as Kafka Key

Lucille uses the document ID as the Kafka message key at every stage of the pipeline:

• Publisher → processing topic: new ProducerRecord(sourceTopicName, document.getId(), document)
• Worker → indexing topic: new ProducerRecord(destTopicName, document.getId(), document)
• Worker/Indexer → event topic: new ProducerRecord(eventTopicName, event.getDocumentId(), event)

Kafka’s partitioning guarantee is: messages with the same key are always routed to the same partition, and messages within a partition are consumed in order by a single consumer. This means that all operations for a given document ID land on the same partition at each stage, and are consumed sequentially by whichever component owns that partition.

Ordering in Fully Distributed Mode

In fully distributed mode, Workers and Indexers are separate processes, each consuming from Kafka topics via consumer groups.

Processing topic → Workers: All messages for document ID “D” land on the same partition P of the processing topic. Kafka assigns partition P to exactly one Worker in the consumer group. That Worker consumes messages from P in order, processing them sequentially (one at a time). The Worker will not pick up the second message for “D” until it has finished processing the first.

Indexing topic → Indexers: When the Worker produces processed documents to the indexing topic, it again uses document.getId() as the key. All messages for document “D” land on the same partition Q of the indexing topic. Kafka assigns partition Q to exactly one Indexer in the consumer group. That Indexer consumes messages from Q in order.

The ordering guarantee holds end-to-end because:

1. Same document ID → same partition (at each topic), guaranteed by Kafka’s partitioner.
2. Same partition → same consumer (Worker or Indexer), guaranteed by Kafka’s consumer group protocol.
3. Same consumer → sequential processing, guaranteed by the single-threaded polling loop in each component.

Multiple Indexers do not break ordering. Even with N Indexers running, each Indexer owns a disjoint set of partitions. All messages for document “D” are on partition Q, and only one Indexer consumes from Q. There is no shared queue where multiple Indexers could pick up different parts of a sequence for the same document.

Ordering in WorkerIndexer Mode

In WorkerIndexer mode, a Worker and an Indexer are paired in the same JVM, communicating via an in-memory LinkedBlockingQueue rather than a Kafka topic between them.

Why the 1:1 pairing preserves ordering: Each WorkerIndexer pair owns one or more Kafka partitions of the processing topic exclusively. All operations for a given document ID land on the same partition, so they are consumed by the same Worker, processed in order, and placed on the in-memory queue in order. The paired Indexer reads from that queue (which is FIFO) and indexes them in order.

What would break ordering without the 1:1 pairing: If multiple Worker threads wrote to a single shared in-memory indexing queue, the ordering of messages for the same document ID would still be preserved — because all messages for the same ID come from the same partition, consumed by the same Worker thread, and placed on the shared queue in order. Interleaving of messages for different document IDs is not a problem for ordering.

The real problem arises on the consumption side: if multiple Indexer threads read from that shared queue, different Indexers could consume different portions of an ordered sequence for the same document ID. Indexer A might pick up the create, Indexer B might pick up the delete, and Indexer B might send its batch to the search backend before Indexer A does — resulting in the document surviving when it should have been deleted.

In fully distributed mode, this problem does not arise even with multiple Indexers, because the indexing topic is a Kafka topic with partition-based consumer assignment. All messages for document “D” are on the same partition, consumed by the same Indexer. But in WorkerIndexer mode, the “indexing queue” is an in-memory LinkedBlockingQueue — it has no concept of partitions or consumer assignment. Any thread reading from it can pick up any message. This is why the 1:1 pairing is necessary: with exactly one Indexer reading from the queue, all messages are consumed in FIFO order, preserving the ordering that the Worker established.

Why WorkerIndexer Pairs Each Worker with a Dedicated Indexer

The 1:1 pairing serves two purposes: offset management and message ordering.

Offset Management

The Worker reads from Kafka and should only commit an offset after the document has been indexed — not just processed. The mechanism:

1. The Worker reads a document from Kafka (noting its topic/partition/offset).
2. The Worker processes it and places it on the in-memory queue.
3. The paired Indexer picks it up, batches it, and sends the batch to the search backend.
4. After the batch succeeds, the Indexer places the offsets of the indexed documents on an in-memory offset queue.
5. The Worker reads from that offset queue and commits those offsets back to Kafka.

This closed loop between one Worker and one Indexer is simple because there’s no ambiguity about which Worker should commit which offset. If multiple Workers shared an Indexer, the Indexer would need to route offset information back to the correct Worker — adding coordination complexity.

Message Ordering

As described above, the in-memory queue between Worker and Indexer has no partition semantics. With a single Indexer consuming from it, FIFO order is preserved. With multiple Indexers, ordering would be lost for documents whose operations span multiple batch flushes.

The Tradeoff

The 1:1 pairing means you cannot independently scale Workers and Indexers within a single process — they scale together as pairs (via WorkerIndexerPool). If you need independent scaling (e.g., many Workers but few Indexers, or vice versa), use fully distributed mode with separate Worker and Indexer processes communicating via Kafka topics, where Kafka’s partition-based assignment provides both ordering and independent scaling.

When Could Ordering Break?

Consumer group rebalance. If Kafka triggers a rebalance (e.g., a Worker or Indexer is slow, or a new instance joins the group), partitions may be reassigned. A document that was in-flight on the old consumer might be redelivered to the new consumer. This can cause a duplicate — but not a reordering, because the duplicate appears later in the partition’s log than the original.

Topic partition count changes. Kafka’s default partitioner hashes the key modulo the partition count. If the partition count changes mid-ingest, the same document ID might hash to a different partition than before. Operations published before the change and operations published after could end up on different partitions, breaking the ordering guarantee. In practice, partition counts should not be changed during an active ingest.

Processing time variance does not break ordering. Even if one document takes much longer to process than another, ordering is preserved because the Worker processes messages from a partition sequentially. It does not start the next message until the current one is complete.

Summary

The ordering guarantee holds as long as:

• Document ID is used as the Kafka key (always true in Lucille).
• Partition counts remain stable during an ingest.
• No consumer group rebalance causes partition reassignment mid-sequence (rare, and results in duplicates rather than reordering).

3 - Error Handling and Fault Tolerance

Philosophy

Most search ingestion projects take an optimistic approach to error handling: ingest as much content as possible and continue past errors. A search index is often useful even if it doesn’t contain 100% of the data from the source system. Source systems routinely contain messy data where some records will inevitably cause errors in processing — a malformed PDF, a record with unexpected encoding, a field that exceeds a size limit. Stopping the entire ingest because one document out of millions is problematic would be counterproductive.

At the same time, it is a waste of time and resources to continue with an ingestion process if there is a configuration or initialization error that prevents it from working properly. If the pipeline configuration references a nonexistent class, or a Stage can’t connect to a required external resource during startup, or the Indexer can’t reach the search backend, there is no point in processing documents — they will all fail.

Furthermore, when an ingestion process consists of a sequence of connectors, we want to proceed with the next connector only if the previous one was able to complete successfully. This is because later connectors may depend on data produced by earlier ones (e.g., parent documents must be indexed before child documents that reference them).

This leads to a two-level error handling philosophy:

Continue past per-document errors. If a single document fails during pipeline processing or indexing, log the failure, count it, and move on. The run continues. A Lucille run with a sequence of 5 connectors can complete and report success even if individual documents encountered errors during processing or indexing and failed to reach the search backend. The run succeeded in the sense that all connectors completed their work and all documents reached a terminal state — some of those terminal states just happen to be “failed” rather than “indexed.”

Stop immediately on structural errors. If the configuration is invalid, if a Stage can’t initialize, if the Indexer can’t connect, or if a Connector throws an exception during its lifecycle — stop. Don’t waste time and resources on work that cannot succeed. Abort the run and report the failure clearly.

The guiding principle: get as much data into the search engine as possible, but stop as soon as possible if you’d only be wasting your time.

Error Scenarios

1. Invalid Configuration

Before any run starts, Runner.run() calls runInValidationMode(config) which validates all pipelines (every Stage’s SPEC), all connectors, the indexer config, and other parent configs (publisher, worker, etc.). If any validation errors are found, the run is aborted immediately — no connectors are started, no documents are processed. A RunResult with status=false is returned. All errors are reported at once (not fail-fast on the first one).

Severity: Fatal to the run. Nothing executes.

2. Connector preExecute() Throws

The Runner catches the ConnectorException, logs it, and returns a failing ConnectorResult with message “preExecute failed.” Neither execute() nor postExecute() are called. The run is aborted (subsequent connectors in the sequence are skipped). close() is still called on the connector.

Severity: Fatal to the connector and the run.

3. Connector execute() Throws

The connector runs inside a ConnectorThread. If execute() throws, the exception is captured on the thread (ConnectorThread.exception). The Publisher’s waitForCompletion() detects that the connector thread has died with an exception and returns a failing PublisherResult. postExecute() is NOT called. The run is aborted. close() is still called.

Severity: Fatal to the connector and the run.

4. Connector postExecute() Throws

postExecute() is only called if both preExecute() and execute() succeeded AND all published documents completed (the Publisher reported success). If postExecute() throws, the Runner catches it and returns a failing ConnectorResult with message “postExecute failed.” The run is aborted. close() is still called.

Severity: Fatal to the connector and the run. Note that all documents were already successfully processed and indexed at this point — the data is in the search engine, but the run is reported as failed.

5. Stage start() Throws

Pipeline.startStages() is called during Pipeline.fromConfig(), which is called inside the Worker constructor. If any Stage’s start() throws a StageException, the Worker constructor throws, which propagates up to WorkerPool.start(). The WorkerPool catches the exception, calls stop() on any threads that were already started, and re-throws. The Runner catches this in runConnectorWithComponents() and returns a failing ConnectorResult with message “Error starting workers for pipeline X.” The run is aborted.

Severity: Fatal to the run. No documents are processed.

6. Stage processDocument() Throws for a Given Document

The Worker catches the exception, logs it via the DocLogger ("Document FAILED during pipeline processing: " + doc.getId()), sends a FAIL event to the Publisher via messenger.sendEvent(doc, null, Event.Type.FAIL), commits offsets, and continues processing the next document. The failed document is counted in the Publisher’s numFailed tally. The run does NOT stop.

Severity: Per-document failure. The run continues. Other documents are unaffected.

7. Worker Process Crashes While Processing a Document

This depends on the deployment mode:

• Local mode: The Worker thread dies. Since the Worker is a thread in the Runner’s JVM, the Runner’s waitForCompletion() will eventually time out or detect that work is not completing (pending documents never reach a terminal state). The run fails.

• Kafka-distributed mode: The Worker process dies. Kafka’s consumer group protocol detects the dead consumer and reassigns its partitions to another available Worker instance. The document that was being processed when the crash occurred was not committed (offset not advanced), so it will be redelivered to the new Worker. The document gets reprocessed.

Severity: In local mode, fatal to the run. In distributed mode, recoverable — the document is retried on another Worker.

8. Same Document Causes Workers to Crash Multiple Times (Poison Pill)

When worker.maxRetries is configured, a RetryCounter (backed by ZooKeeper or Redis) tracks how many times each document has been attempted across all Worker instances. Each time a Worker picks up a document, it calls counter.add(doc). If the retry count exceeds the configured maximum, the Worker:

1. Sends the document to a “failed” / dead-letter queue via messenger.sendFailed(doc)
2. Sends a FAIL event to the Publisher
3. Skips processing and moves on

The document is effectively quarantined. The rest of the ingest continues unaffected.

Severity: Per-document failure. The poison pill is isolated. The run continues.

9. Indexer Can’t Connect to the Search Backend

Before the Indexer thread starts processing, validateConnection() is called. If it returns false:

• In the Runner (local/Kafka-local mode): The Runner logs “Indexer could not connect”, calls indexer.closeConnection(), and returns a failing ConnectorResult. The run is aborted.

• In standalone Indexer mode (distributed): The Indexer logs the error and calls System.exit(1).

• In WorkerIndexer mode: The start() method throws an IndexerException("Indexer could not connect"), which prevents the WorkerIndexer from starting.

Severity: Fatal. No documents are indexed.

10. A Batch Fails During Indexing

There are two sub-cases:

10a. The entire batch call throws an exception (batch-level failure): The Indexer catches the exception, logs it, and sends a FAIL event for every document in the batch. If retries are configured (indexer.maxRetries > 0) and the failure’s status code is in indexer.retryableStatusCodes (default [429, 503, -1]), the batch is retried with exponential backoff before being declared failed. After all retries are exhausted (or if the status code is not retryable), all documents in the batch receive FAIL events. The Indexer continues processing subsequent batches. The batchComplete() call in the finally block always fires regardless of outcome.

10b. The bulk API succeeds but reports per-document failures: sendToIndex() returns a set of failed document/reason pairs. The Indexer sends FAIL events for those specific documents and FINISH events for the rest. The Indexer continues normally.

Severity: Per-batch or per-document failure. The run continues. Failed documents are counted in the Publisher’s accounting.

11. Indexer Process Crashes

• Local mode: The Indexer thread dies. Documents accumulate on the indexing queue but are never consumed. The Publisher’s waitForCompletion() will eventually time out because pending documents never reach a terminal state. The run fails.

• Kafka-distributed mode: The Indexer process dies. Kafka’s consumer group protocol reassigns its partitions to another available Indexer instance. Unacknowledged documents (those in the batch that was being processed) are redelivered to the new Indexer. Since search engine upserts are idempotent, re-indexing an already-indexed document produces the same result.

Severity: In local mode, fatal to the run. In distributed mode, recoverable.

12. Runner Process Crashes

The Runner has a signal handler (Signal.handle(new Signal("INT"), ...)) that attempts a clean shutdown: closing the Connector, closing the Publisher, stopping the WorkerPool, and terminating the Indexer. If the Runner crashes hard (e.g., OOM, kill -9):

• Local mode: Everything dies — all components are threads in the same JVM. The run is lost. Documents that were in-flight are lost (they were in in-memory queues).

• Kafka-distributed mode: Only the Connector and Publisher die. Workers and Indexers are separate processes and continue running. However, no new documents will be published, and the Publisher’s accounting is lost. Documents already on Kafka topics will still be processed and indexed by the Workers and Indexers, but there is no completion detection — the run has no coordinator to declare it done.

Severity: Fatal to the run in all modes. In distributed mode, in-flight work is not lost (it’s on Kafka), but run-level accounting is.

13. Additional Error Scenarios

Publisher fails to send an event (messenger.sendEvent throws): The Worker and Indexer both catch this and log it. The consequence is severe: if a FINISH or FAIL event is lost, the Publisher will never learn that the document completed. waitForCompletion() will hang until timeout. The code logs "RUN WILL HANG" in this case. This is a rare edge case — it would require the event queue (Kafka topic or in-memory queue) to be unavailable.

Worker watcher detects a stuck Worker: The WorkerPool runs a watcher thread that checks each Worker’s last poll timestamp. If a Worker hasn’t polled in worker.maxProcessingSecs seconds (default 10 minutes), it logs an error. If worker.exitOnTimeout is configured, it calls System.exit(1). This handles the case where a Stage enters an infinite loop or deadlock on a particular document.

Connector’s close() throws: The Runner catches this and returns a failing ConnectorResult. Since close() is called in a finally block after all other work, the documents may have been successfully processed, but the run is reported as failed.

Publisher’s waitForCompletion() times out: If the configured runner.connectorTimeout (default 24 hours) is exceeded, waitForCompletion() returns a failing PublisherResult. The run is reported as failed. This catches scenarios where documents are stuck (lost events, hung Workers, etc.).

Kafka offset commit fails: The Worker calls messenger.commitPendingDocOffsets() after processing each document. If this throws, the Worker logs the error and continues. The consequence is that if the Worker later crashes, the document may be redelivered and reprocessed (at-least-once semantics). This is safe because pipeline processing is expected to be idempotent.

Document publish fails (publisher.publish() throws): If the Publisher’s sendForProcessing() throws (e.g., the processing queue is broken), the exception propagates up to the Connector’s execute() method. Unless the Connector catches it, this becomes scenario #3 (connector execute throws) and the run fails.

Fault Tolerance

Single-JVM Mode Is Not Fault-Tolerant

In local mode, all components — Connector, Workers, Indexer, Publisher — run as threads inside a single JVM. If the JVM crashes or the server goes down, everything is lost: in-flight documents were in in-memory queues and are gone. There is no recovery mechanism. The run must be restarted from the beginning.

This is an acceptable tradeoff for development, testing, and small production jobs where a restart is cheap. For workloads where fault tolerance matters, Lucille’s distributed mode with Kafka provides the answer.

Fault Tolerance Is Inherited from Kafka

Lucille does not implement its own fault-tolerance logic (no custom write-ahead logs, no checkpointing to disk, no replication protocol). Instead, it inherits fault tolerance from Kafka by treating Kafka topics as durable queues and relying on Kafka’s consumer group protocol for work reassignment when a process dies.

The key to making this work is proper offset management. Lucille’s contract is:

A Worker only commits a Kafka offset once the current document has been fully processed and placed on the destination queue.

If the Worker crashes while processing a document, that Kafka message’s offset has not been committed. When Kafka detects the dead consumer (via missed heartbeats), it reassigns that partition to another Worker in the consumer group. The new Worker begins consuming from the last committed offset — which is before the document that caused the crash. That document is redelivered and reprocessed.

Two Levels of Offset Commitment

Lucille has two deployment patterns with different offset semantics:

Fully distributed mode (separate Worker and Indexer processes): The Worker commits offsets via KafkaWorkerMessenger.commitPendingDocOffsets() after processing each document and placing it on the indexing topic. The guarantee is: if the Worker crashes, unprocessed documents are redelivered. However, documents that were processed and placed on the indexing topic but not yet indexed are safe — they’re on Kafka and will be picked up by an Indexer.

WorkerIndexer mode (paired Worker + Indexer in one process): The offset flow is more sophisticated. The Indexer, after successfully sending a batch to the search backend, places the offsets of the indexed documents on an in-memory queue. The Worker reads from this queue during its commitPendingDocOffsets() call and commits them back to Kafka. The guarantee is stronger: offsets are only committed after documents are indexed, not just processed. If the WorkerIndexer crashes, documents that were processed but not yet indexed are redelivered and re-indexed. Since search engine upserts are idempotent, this produces correct results.

What Fault Tolerance Guarantees

• No data loss. A document that enters the system (is placed on a Kafka topic) will eventually be processed and indexed, or will exhaust its retry count and be routed to a dead-letter queue. It will not silently disappear.
• At-least-once processing. A document may be processed more than once if a crash occurs after processing but before offset commit. Pipeline stages and indexing operations should be idempotent (search engine upserts naturally are).
• Automatic recovery. No manual intervention is required when a Worker or Indexer crashes. Kafka’s consumer group protocol handles partition reassignment automatically. New instances can be added at any time.

What Fault Tolerance Does Not Guarantee

• Exactly-once processing. There is a window between completing work and committing the offset where a crash causes redelivery. This is the standard Kafka at-least-once pattern.
• Run-level accounting survives a Runner crash. The Publisher (which tracks completion) runs alongside the Connector in the Runner process. If the Runner dies, accounting is lost. Documents on Kafka will still be processed, but no component knows when “the run” is done.
• Order preservation after redelivery. A redelivered document may be processed after documents that were originally behind it in the queue. For most search ingestion workloads this is acceptable — the final state of the index is the same regardless of processing order.

Summary Table

4 - Kafka Integration

Overview

Kafka is Lucille’s distributed messaging backbone. When running in KAFKA_LOCAL or KAFKA_DISTRIBUTED mode, all inter-component communication flows through Kafka topics. This enables horizontal scaling: multiple Worker processes can consume from the same source topic, and multiple Indexer processes can consume from the same dest topic.

Topic Naming Conventions

Lucille uses four topics per pipeline, named by convention:

public static String getSourceTopicName(String pipelineName, Config config) {
    // Override: kafka.sourceTopic
    // Default: {pipelineName}_source
    return pipelineName + "_source";
}

public static String getDestTopicName(String pipelineName) {
    return pipelineName + "_dest";
}

public static String getFailTopicName(String pipelineName) {
    return pipelineName + "_fail";
}

public static String getEventTopicName(Config config, String pipelineName, String runId) {
    // Override: kafka.eventTopic
    // Default: {pipelineName}_event_{runId}
    return pipelineName + "_event_" + runId;
}

The event topic is per-run (includes the runId) because each run needs its own isolated event stream. The source, dest, and fail topics are per-pipeline and persist across runs.

The source topic name is validated to contain only safe characters ([A-Za-z\d._-]+) since it may be used as a regex pattern for consumer subscription.

KafkaDocument: Carrying Kafka Metadata

KafkaDocument extends JsonDocument to carry partition/offset/key metadata alongside document fields:

public class KafkaDocument extends JsonDocument {
    private String topic;
    private int partition;
    private long offset;
    private String key;

    public void setKafkaMetadata(ConsumerRecord<String, ?> record) {
        this.topic = record.topic();
        this.partition = record.partition();
        this.offset = record.offset();
        this.key = record.key();
    }
}

This metadata travels with the document through the pipeline. It’s essential for the Hybrid mode where the Indexer needs to report back which offsets have been successfully processed.

Plain Document objects are written to Kafka. When deserialized, they come back as KafkaDocument instances with the Kafka metadata attached from the ConsumerRecord.

Serializer/Deserializer

Documents are serialized as JSON using Jackson:

public class KafkaDocumentSerializer implements Serializer<Document> {
    private static final ObjectMapper MAPPER = new ObjectMapper();

    @Override
    public byte[] serialize(String topic, Document doc) {
        if (doc == null) return null;
        return MAPPER.writeValueAsBytes(doc);
    }
}

public class KafkaDocumentDeserializer implements Deserializer<Document> {
    private static final ObjectMapper MAPPER = new ObjectMapper();

    @Override
    public Document deserialize(String topic, byte[] data) {
        if (data == null) return null;
        return new KafkaDocument((ObjectNode) MAPPER.readTree(data));
    }
}

The deserializer always produces a KafkaDocument (even though the return type is Document). The Kafka metadata is set separately after deserialization via setKafkaMetadata().

Custom serializers/deserializers can be specified via config:

kafka.documentSerializer = "com.example.MySerializer"
kafka.documentDeserializer = "com.example.MyDeserializer"

Document ID as Kafka Message Key

Documents are produced with their ID as the Kafka key:

// In KafkaPublisherMessenger:
kafkaProducer.send(new ProducerRecord(sourceTopicName, document.getId(), document));

// In KafkaWorkerMessenger:
kafkaDocumentProducer.send(new ProducerRecord<>(destTopicName, document.getId(), document));

This provides ordering guarantees: all messages with the same key go to the same partition, ensuring that a document and its children are processed in order within a single partition. It also means that if the same document ID is published multiple times, all versions land on the same partition.

Consumer Group Management

Workers and Indexers join consumer groups to enable parallel consumption:

consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, config.getString("kafka.consumerGroupId"));

All Workers for a pipeline share the same consumer group. Kafka distributes partitions among group members, so adding more Workers increases parallelism (up to the number of partitions).

Each consumer gets a unique client ID to avoid Kafka warnings:

String kafkaClientId = "com.kmwllc.lucille-worker-" + pipelineName + "-" + RandomStringUtils.randomAlphanumeric(8);

Key consumer settings:

consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);          // One doc at a time
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");  // Manual commits

The maxPollIntervalSecs Setting

consumerProps.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 
    1000 * config.getInt("kafka.maxPollIntervalSecs"));

This is the maximum time between poll() calls before Kafka considers the consumer dead and triggers a rebalance. It must be set higher than the longest expected document processing time. If a document takes longer to process than this interval, the consumer will be kicked from the group and the document will be reprocessed by another consumer.

Topic Creation

The event topic is created explicitly with exactly one partition:

public static boolean createEventTopic(Config config, String pipelineName, String runId) {
    String eventTopicName = KafkaUtils.getEventTopicName(config, pipelineName, runId);

    // Single partition is critical for ordering
    NewTopic eventTopic = new NewTopic(eventTopicName, 1, (short) 1);

    try (Admin kafkaAdminClient = Admin.create(props)) {
        CreateTopicsResult result = kafkaAdminClient.createTopics(List.of(eventTopic));
        result.all().get();
    } catch (ExecutionException e) {
        if (e.getCause() instanceof TopicExistsException) {
            return false;  // Already exists, that's fine
        }
        throw e;
    }
    return true;
}

Why single partition for events? Multiple partitions could cause events to arrive out of order. If a child’s FINISH event arrives before its CREATE event (because they’re on different partitions), the Publisher’s accounting logic would be corrupted. A single partition guarantees FIFO ordering.

The source and dest topics are NOT explicitly created by Lucille — they’re expected to exist already or be auto-created by Kafka’s broker configuration.

Per-Run Event Topics in Batch Mode

In batch mode, each run gets its own event topic (e.g., my_pipeline_event_c1d9413a-8191-4f4a-92bb-0fc42b5499e3). Lucille creates this topic via the Kafka Admin API at the start of each run. This means:

• Kafka Admin API access is required. Lucille creates the event topic explicitly via the Admin API to guarantee it has exactly 1 partition (required for FIFO event ordering). Kafka’s auto.create.topics.enable is NOT sufficient — auto-created topics use the broker’s default partition count, which would create a multi-partition topic and break the Publisher’s accounting logic. If your Kafka cluster requires separate admin credentials, provide them via kafka.adminPropertyFile.
• Each batch ingest produces a new topic. Over time, event topics accumulate. Consider a retention policy or periodic cleanup of old event topics.
• Each Publisher requires its own dedicated event topic. There is no support for multiple Publishers sharing a single event topic and filtering events by run_id. The per-run topic IS the isolation mechanism.

Do NOT set kafka.eventTopic in batch mode. Setting a fixed event topic name would cause all runs to share one topic. If two runs overlap (e.g., launched via the RunnerManager API), their events would be interleaved on the same topic. Each Publisher would see events from the other run, corrupting its accounting and potentially declaring completion prematurely.

kafka.eventTopic is safe only in streaming mode, where there is no Publisher performing completion accounting.

This topic-creation requirement does not apply in streaming mode — when events are disabled (kafka.events: false) no event topic is needed, and when a fixed kafka.eventTopic is set the topic can be pre-created once and reused indefinitely.

The Event Topic: Lifecycle Events

Events flow from Workers and Indexers back to the Publisher:

• Worker → Event Topic: CREATE (child document generated), FAIL (processing error), DROP (document dropped by stage)
• Indexer → Event Topic: FINISH (successfully indexed), FAIL (indexing error)

Events are serialized as JSON strings (not using the document serializer):

// Producer uses StringSerializer for events
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

// Sending an event
kafkaEventProducer.send(
    new ProducerRecord<>(confirmationTopicName, event.getDocumentId(), event.toString()));

The event consumer uses auto-commit for throughput:

consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");

This is acceptable because event loss is not catastrophic — the worst case is the Publisher waits longer or times out. Duplicate events are handled gracefully by the Publisher’s Bag-based accounting.

Events can be disabled entirely:

kafka.events = false

When disabled, createEventProducer() returns null and all sendEvent() calls become no-ops. This is useful for Workers/Indexers that run independently without a Publisher waiting for completion.

Event Topic Naming in Streaming Mode

In streaming mode (no Runner, no Publisher), documents arrive from an external system without a run_id field. Since the default event topic name is {pipeline}_event_{runId}, a null run_id produces a topic named {pipeline}_event_null. This is functional but awkward.

To avoid this, use kafka.eventTopic to set a fixed topic name:

kafka {
  events: true
  eventTopic: "lucille_events"  # fixed name, independent of run_id
}

When kafka.eventTopic is set, all events go to that single topic regardless of the document’s run_id. This is safe in streaming mode because there is no Publisher performing per-run completion accounting. In batch mode with a Runner, do NOT set kafka.eventTopic — the per-run topic isolation is essential for correct completion detection.

See Streaming Mode Configuration for the full streaming setup.

The Fail Topic (Dead Letter Queue)

Documents that exceed retry limits are sent to the fail topic:

// In KafkaWorkerMessenger:
public void sendFailed(Document document) throws Exception {
    ProducerRecord<String, Document> producerRecord =
        new ProducerRecord<>(KafkaUtils.getFailTopicName(pipelineName), document.getId(), document);
    kafkaDocumentProducer.send(producerRecord).get();
    kafkaDocumentProducer.flush();
}

The fail topic ({pipeline}_fail) acts as a dead letter queue. Documents here can be inspected, fixed, and replayed. The Worker sends a document to the fail topic when its retry count (tracked in ZooKeeper) exceeds worker.maxRetries.

Kafka Configuration Options

All Kafka settings live under the kafka config prefix:

Partitions and Parallelism

The relationship between Kafka partitions and Lucille parallelism:

• Source topic partitions determine max Worker parallelism. With 8 partitions, at most 8 Workers can consume concurrently (within the same consumer group).
• Dest topic partitions determine max Indexer parallelism. Same principle.
• Event topic always has 1 partition (ordering requirement).

To scale Workers: increase source topic partitions and add more Worker processes/threads.

Lucille polls one record at a time (MAX_POLL_RECORDS_CONFIG = 1) to ensure fine-grained offset control and prevent one slow document from blocking a batch.

Important: Source Topic Partition Count and Worker Threads

Lucille does not automatically create the source or dest topics. It only explicitly creates the event topic (with 1 partition for ordering). The source and dest topics are either:

• Pre-created by an administrator with the desired partition count, or
• Auto-created by Kafka when the Publisher first writes to them (if auto.create.topics.enable=true on the broker)

The gotcha: If Kafka auto-creates the source topic, the partition count is determined by the broker’s num.partitions setting (default: 1). This means that even if you configure worker.threads: 8, only 1 thread will receive documents — the other 7 will join the consumer group but sit idle because there’s only 1 partition to assign.

Lucille does not validate at startup that the source topic has enough partitions for the configured number of worker threads. Excess consumers are simply idle — they don’t error out.

How to avoid this:

• Pre-create the source topic with the desired partition count before running Lucille:

  kafka-topics.sh --create --topic my-pipeline_source \
    --partitions 8 --replication-factor 1 \
    --bootstrap-server kafka:9092
  

• Or configure the Kafka broker’s num.partitions to a higher default (applies to all auto-created topics).
• Rule of thumb: Set the source topic partition count to at least the maximum total number of Worker threads you expect to run across all processes.

External Property Files for Advanced Configuration

For complex Kafka setups (SASL, SSL, custom partitioners), external property files can be specified:

kafka.consumerPropertyFile = "/path/to/consumer.properties"
kafka.producerPropertyFile = "/path/to/producer.properties"
kafka.adminPropertyFile = "/path/to/admin.properties"

When a property file is specified, it completely replaces the programmatic configuration (except for CLIENT_ID_CONFIG which is always set). The file is loaded via FileContentFetcher which supports local files and cloud storage (S3, Azure, GCP).

private static Properties loadExternalProps(String filename, Config config) {
    try (Reader propertiesReader = FileContentFetcher.getOneTimeReader(filename, StandardCharsets.UTF_8.name(), config)) {
        Properties consumerProps = new Properties();
        consumerProps.load(propertiesReader);
        return consumerProps;
    }
}

Offset Commit Strategies

Different components use different commit strategies:

The Worker’s synchronous commit ensures that if a Worker crashes, the document it was processing will be redelivered to another Worker. Without this, the document could be lost (committed but not processed).

Producer Behavior

All producers use synchronous sends (.get() after send()):

kafkaDocumentProducer.send(new ProducerRecord<>(...)).get();
kafkaDocumentProducer.flush();

This ensures the message is acknowledged by the broker before proceeding. Combined with MAX_POLL_RECORDS_CONFIG = 1, this creates a strict one-at-a-time processing model that prioritizes correctness over throughput.

Producer settings:

producerProps.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, config.getInt("kafka.maxRequestSize"));
producerProps.put(ProducerConfig.BUFFER_MEMORY_CONFIG, config.getInt("kafka.maxRequestSize"));

Both maxRequestSize and bufferMemory are set to the same value, ensuring the producer can always send a single maximum-sized message.

5 - Metrics and Observability

Overview

Lucille uses the Codahale (Dropwizard) Metrics library for runtime observability. Every component — Publisher, Worker, Pipeline, Stage, and Indexer — reports metrics through a shared registry. These metrics are logged periodically during execution and summarized at the end of each run.

The Shared MetricRegistry

All components share a single registry:

MetricRegistry metrics = SharedMetricRegistries.getOrCreate(LogUtils.METRICS_REG);

Where LogUtils.METRICS_REG is the constant "default". Using SharedMetricRegistries means any component in the JVM can access the same registry without passing references around.

The Naming Convention

Metrics are namespaced using a metricsPrefix that encodes the run context:

String metricsPrefix = runId + "." + connector.getName() + "." + connector.getPipelineName();

This ensures metrics are collected separately for each connector/pipeline pair within a run. Individual components append their own suffix:

Stages use their own naming scheme within the Pipeline (see below).

Publisher Metrics

The Publisher tracks the rate and cadence of document publishing:

this.timer = SharedMetricRegistries.getOrCreate(LogUtils.METRICS_REG)
    .timer(metricsPrefix + ".timeBetweenPublishCalls");

In publish(), the timer measures the gap between consecutive calls:

if (timerContext.get() != null) {
    timerContext.get().stop();  // Stop timing the gap since last publish
}
try {
    publishInternal(document);
} finally {
    timerContext.set(timer.time());  // Start timing gap to next publish
}

This gives two insights:

• Mean rate — how fast the connector produces documents (docs/sec)
• Mean duration — average time between publish calls (ms/doc), indicating connector latency

The timer is ThreadLocal so multiple publishing threads don’t interfere with each other.

Periodic logging in waitForCompletion():

log.info(String.format(
    "%d docs published. One minute rate: %.2f docs/sec. Mean connector latency: %.2f ms/doc. Waiting on %d docs.",
    numReceived.get(), timer.getOneMinuteRate(), timer.getSnapshot().getMean() / 1000000, numPending()));

Worker Metrics

The Worker measures document processing time (pipeline latency):

Timer timer = metrics.timer(metricsPrefix + Worker.METRICS_SUFFIX);
// METRICS_SUFFIX = ".worker.docProcessingTme"

// In the processing loop:
Timer.Context context = timer.time();
Iterator<Document> results = pipeline.processDocument(doc);
// ... emit results ...
context.stop();

This timer captures the full pipeline execution time per document, including all stages.

Stage Metrics

Each Stage tracks its own processing time and counts. Metrics are initialized when the stage is added to a pipeline via stage.initialize(position, metricsPrefix):

The Stage base class provides:

• A Timer for per-document processing time
• An error Counter for documents that fail in this stage
• A child Counter for child documents generated by this stage

The logMetrics() method on Pipeline iterates all stages and logs their individual metrics.

Indexer Metrics

The Indexer tracks throughput and backend latency:

this.meter = metrics.meter(metricsPrefix + ".indexer.docsIndexed");
this.histogram = metrics.histogram(metricsPrefix + ".indexer.batchTimeOverSize");

After each batch is sent to the search engine:

stopWatch.reset();
stopWatch.start();
Set<Pair<Document, String>> failedDocPairs = sendToIndex(batchedDocs);
stopWatch.stop();

histogram.update(stopWatch.getNanoTime() / batchedDocs.size());  // Per-doc latency
meter.mark(batchedDocs.size());                                   // Throughput

The histogram records time per document (total batch time / batch size), giving a normalized view of backend performance regardless of batch size.

Periodic logging:

log.info(String.format(
    "%d docs indexed. One minute rate: %.2f docs/sec. Mean backend latency: %.2f ms/doc.",
    meter.getCount(), meter.getOneMinuteRate(), histogram.getSnapshot().getMean() / 1000000));

The WorkerPool Watcher Thread

The WorkerPool starts a scheduled watcher that runs every 500ms:

private ScheduledExecutorService startWatcher(List<Worker> workers, int maxProcessingSecs) {
    TimerTask watcher = new TimerTask() {
        private final Timer timer = metrics.timer(metricsPrefix + Worker.METRICS_SUFFIX);
        private Instant lastLogInstant = null;

        public void run() {
            // Periodic stats logging
            if (Duration.between(lastLogInstant, Instant.now()).getSeconds() >= logSeconds) {
                log.info(String.format(
                    "%d docs processed. One minute rate: %.2f docs/sec. Mean pipeline latency: %.2f ms/doc.",
                    timer.getCount(), timer.getOneMinuteRate(), timer.getSnapshot().getMean() / 1000000));

                // Heartbeat
                if (enableHeartbeat) {
                    heartbeatLog.info("Issuing heartbeat");
                }
            }

            // Stuck worker detection
            for (Worker worker : workers) {
                if (Duration.between(worker.getPreviousPollInstant().get(), Instant.now()).getSeconds() > maxProcessingSecs) {
                    log.error("Worker has not polled in " + maxProcessingSecs + " seconds.");
                    if (exitOnTimeout) {
                        System.exit(1);
                    }
                }
            }
        }
    };
}

The watcher serves three purposes:

1. Periodic Statistics Logging

Every logSeconds (default: 30, configurable via log.seconds), it logs pipeline throughput and latency. This gives operators real-time visibility into processing speed.

2. Stuck Worker Detection

Each Worker updates an AtomicReference<Instant> every time it polls for a new document:

// In Worker.run():
pollInstant.set(Instant.now());
doc = messenger.pollDocToProcess();

The watcher checks if any worker hasn’t polled within maxProcessingSecs (default: 600 seconds / 10 minutes, configurable via worker.maxProcessingSecs). A worker that hasn’t polled is likely stuck processing a single document.

3. The exitOnTimeout Mechanism

When worker.exitOnTimeout is true and a stuck worker is detected, the JVM exits immediately with System.exit(1). This is designed for containerized deployments where a hard restart (via Kubernetes pod restart) is preferable to a hung process.

The Heartbeat Mechanism

public static final String HEARTBEAT_LOG_NAME = "com.kmwllc.lucille.core.Heartbeat";
private static final Logger heartbeatLog = LoggerFactory.getLogger(HEARTBEAT_LOG_NAME);

if (enableHeartbeat) {
    heartbeatLog.info("Issuing heartbeat");
    if (heartbeatLog.isDebugEnabled()) {
        heartbeatLog.debug("Thread Dump:\n{}", 
            Arrays.toString(ManagementFactory.getThreadMXBean().dumpAllThreads(true, true)));
    }
}

When worker.enableHeartbeat is true, the watcher writes to a dedicated heartbeat logger. This can be configured (via logback) to write to a specific file that a Kubernetes liveness probe checks. If the file stops being updated, the probe fails and the pod is restarted.

At DEBUG level, it also dumps all thread stacks — useful for diagnosing what a stuck worker is doing.

End-of-Run Metrics Reporting

After all connectors complete, the Runner logs all collected metrics via Slf4jReporter:

Slf4jReporter.forRegistry(SharedMetricRegistries.getOrCreate(LogUtils.METRICS_REG))
    .outputTo(log)
    .withLoggingLevel(getMetricsLoggingLevel(config))
    .build()
    .report();

The logging level is configurable via runner.metricsLoggingLevel (default: DEBUG). This dumps every timer, meter, histogram, and counter in the registry.

The log.seconds Configuration

Controls how frequently periodic stats are logged:

this.logSeconds = ConfigUtils.getOrDefault(config, "log.seconds", LogUtils.DEFAULT_LOG_SECONDS);
// DEFAULT_LOG_SECONDS = 30

Used by:

• Publisher (in waitForCompletion)
• WorkerPool watcher
• Indexer (in sendToIndexWithAccounting)

Setting this lower gives more frequent visibility; setting it higher reduces log noise.

MDC (Mapped Diagnostic Context) Usage

Lucille uses SLF4J’s MDC to attach contextual information to every log line:

run_id

Set at the start of each thread’s work:

MDC.put("run_id", runId);  // In ConnectorThread
MDC.put(RUNID_FIELD, localRunId);  // In Worker
MDC.pushByKey(RUNID_FIELD, localRunId);  // In Indexer (stack-based for multi-run)

This allows log aggregation tools to filter all log lines for a specific run.

doc_id

Set when processing a specific document:

MDC.put(Document.ID_FIELD, document.getId());  // In Publisher.publish()

try (MDC.MDCCloseable docIdMDC = MDC.putCloseable(ID_FIELD, doc.getId())) {
    // In Worker and Indexer — auto-removed when block exits
    docLogger.info("Worker is processing document {}.", doc.getId());
}

The DocLogger (logger name com.kmwllc.lucille.core.DocLogger) is a dedicated logger for document lifecycle events. Combined with MDC, you can trace a single document’s journey through the entire system.

Indexer MDC Stack

The Indexer uses pushByKey/popByKey for run_id because in Kafka distributed mode, documents from different runs might be interleaved:

if (d.getRunId() != null) {
    MDC.pushByKey(RUNID_FIELD, d.getRunId());
}
// ... send event ...
if (d.getRunId() != null) {
    MDC.popByKey(RUNID_FIELD);
}

Summary of Configurable Observability Settings