Saltar a contenido

Exporter System

El sistema de exportación transfiere records del log stream de Zeebe hacia almacenes secundarios (Elasticsearch, OpenSearch, RDBMS). Usa un SPI con semántica at-least-once, tracking de posición por partición, y un mecanismo de backpressure que throttlea writes cuando los exporters no pueden seguir el ritmo. A partir de Camunda 8.8, el CamundaExporter consolida la lógica de importación y archivado que antes vivía en las webapps.

Exporter SPI

La interfaz Exporter define el contrato que toda implementación debe cumplir:

Método Responsabilidad
configure(Context context) Inicialización con la configuración del exporter. Se llama una vez al arrancar. Recibe el Context con access a logger, configuración YAML, y filtros.
open(Controller controller) Abre conexiones al almacén destino (Elasticsearch client, JDBC connection, etc.). El Controller permite reportar posición y programar tareas. Se llama después de configure.
export(Record<?> record) Procesa un record individual del log stream. El exporter decide si lo indexa, transforma, filtra, o ignora. Retorna la posición del record procesado.
close() Libera recursos (cierra conexiones, flush de buffers). Se llama al shutdown del broker o al descartar el exporter.
purge() Limpia los datos exportados del almacén destino, para testeo o migración. Opcional pero recomendado.

Instanciación: un exporter por partición

Cada partición del cluster Zeebe tiene su propia instancia del exporter. Esto significa:

  • Paralelismo natural: exporters de particiones distintas operan independientemente sin coordinación.
  • Aislamiento de fallos: si el exporter de la partición 3 falla, las particiones 1 y 2 siguen exportando.
  • Estado local: cada instancia mantiene su propia posición de exportación y buffers.

Semántica at-least-once

La garantía de entrega es at-least-once, no exactly-once:

  1. El exporter procesa un batch de records.
  2. Una vez que el batch está confirmado en el almacén destino, el exporter llama controller.updateLastExportedRecordPosition(position).
  3. Si el broker se cae o el exporter se reinicia antes del update de posición, los records del batch se re-exportan al arrancar.
  4. Consecuencia: los consumidores downstream (Operate, Tasklist, Optimize) DEBEN ser idempotentes ante records duplicados. Típicamente usan upsert en vez de insert.

Tracking de posición

El Controller provee el método:

void updateLastExportedRecordPosition(long position);
  • La posición es un offset en el log stream de la partición.
  • Se persiste en el state store de Zeebe (RocksDB).
  • Al reiniciar, el exporter reanuda desde la última posición confirmada.
  • Si nunca se llamó updateLastExportedRecordPosition, el exporter empieza desde el inicio del log.

Filtrado de records

El SPI soporta dos niveles de filtrado para que los exporters no procesen records irrelevantes:

Filtrado a nivel de metadata (pre-deserialización)

  • Se evalúa el RecordType y ValueType del record antes de deserializar el payload.
  • Ejemplo: un exporter que solo necesita DEPLOYMENT y PROCESS_INSTANCE puede descartar todos los demás sin costo de deserialización.
  • Se configura en el método configure via Context.getFilter().

Filtrado a nivel de record (post-deserialización)

  • Se evalúa el contenido completo del record después de deserializar.
  • Ejemplo: filtrar solo records con intent = ELEMENT_COMPLETED o solo de un proceso específico.
  • Se implementa en el método export del propio exporter.

Manejo de errores

El comportamiento ante errores es distinto según la fase:

Errores en open()

  • Si open() lanza una excepción, el ExporterDirector programa un reinicio del exporter.
  • El reinicio llama close() y luego open() de nuevo.
  • Se reintenta indefinidamente con backoff.
  • No bloquea la partición: el log stream sigue escribiendo records; se acumula un backlog.

Errores en export()

  • Si export() lanza una excepción, el ExporterDirector reintenta el mismo batch indefinidamente.
  • No avanza la posición hasta que el batch sea procesado exitosamente.
  • SÍ bloquea la partición: nuevos records se siguen escribiendo al log, pero el exporter no avanza su posición. Si el gap entre write position y export position crece demasiado, se activa backpressure.

Consecuencia de un exporter bloqueado

Un exporter que falla persistentemente en export() eventualmente causa que la partición entera se throttlee via backpressure, afectando el throughput de toda la partición.

Backpressure: FlowControl

FlowControl es el mecanismo que previene que el log stream crezca sin límite cuando los exporters no pueden seguir el ritmo:

  1. Gap tracking: se mide la diferencia entre la posición de escritura más reciente y la posición de exportación más atrasada.
  2. Threshold: si el gap supera un umbral configurable, se activa backpressure.
  3. Throttling: el engine rechaza o retrasa nuevos writes al log stream (commands entrantes).
  4. Efecto: los clientes gRPC reciben errores de backpressure (gRPC status RESOURCE_EXHAUSTED) y deben reintentar.
  5. Recuperación: cuando el exporter se pone al día y el gap disminuye por debajo del threshold, se levanta la restricción.

Diseño deliberado: es preferible ralentizar todo el sistema a perder datos de exportación, porque las webapps (Operate, Tasklist) dependen de datos completos para funcionar correctamente.

Implementaciones principales

ElasticsearchExporter

El exporter original y más maduro. Indexa records en Elasticsearch usando la Bulk API.

Configuración de batching: - Batch size: 1000 records por bulk request (default). - Batch memory: 10 MB máximo por bulk request. - Flush interval: 5 segundos — flush del batch acumulado aunque no se alcance el size o memory limit.

El batch se envía cuando se cumple cualquiera de las tres condiciones (el primero que se cumpla).

Índices: crea índices separados por tipo de record (process-instance, job, variable, etc.) con templates y mappings predefinidos.

CamundaExporter (8.8+)

Exporter que a partir de Camunda 8.8 consolida la lógica que antes estaba distribuida entre los exporters y los importers de las webapps:

  • 60+ handlers: cada handler procesa un tipo específico de record y lo transforma al modelo de datos de las webapps.
  • Import logic: la lógica de importación que antes vivía en Operate/Tasklist ahora vive en el exporter, ejecutándose más cerca del log stream.
  • Archiving logic: la lógica de archivado (mover instancias completadas a índices archive) también se mueve al exporter.
  • Beneficio: reduce la complejidad de las webapps, mejora la latencia de datos (menos hops), y simplifica el deployment.

OpenSearchExporter

Variante del ElasticsearchExporter para clusters que usan Amazon OpenSearch. Funcionalidad equivalente, adaptada a las diferencias de API entre Elasticsearch y OpenSearch.

RDBMSExporter

Exporter experimental para bases de datos relacionales. Permite usar PostgreSQL u otra RDBMS en vez de Elasticsearch/OpenSearch como almacén secundario.

Pipeline de datos completo

Engine (command processing)
    ↓ (writes)
LogStream (append-only log por partición)
    ↓ (reads)
ExporterDirector (orquesta exporters)
    ↓ (dispatch)
Exporters (ES/OS/RDBMS/Camunda)
    ↓ (bulk indexing)
Elasticsearch / OpenSearch
    ↓ (queries)
Search infrastructure (indices, templates)
    ↓ (REST queries)
REST API layer
    ↓ (HTTP)
Webapps: Operate, Tasklist, Optimize

Latencia típica del pipeline: entre 1-10 segundos desde que un record se escribe en el log hasta que es visible en las webapps. El bottleneck principal es el flush interval del exporter (5s default) más el refresh interval de Elasticsearch (1s default).

Implicación para las webapps: Operate y Tasklist muestran datos eventualmente consistentes. Una instancia de proceso puede completarse en el engine pero seguir apareciendo como "activa" en Operate por unos segundos. Este es un trade-off aceptado por el diseño del sistema.