ElasticsearchExporter implements Exporter { @Override public void export(final Record<?> record) { if (!indexTemplatesCreated) { createIndexTemplates(); } client.index(record); lastPosition = record.getPosition(); if (client.shouldFlush()) { flush(); } }  @Override void configure(final Context context); @Override void open(final Controller controller); @Override void close(); @Override void export(final Record<?> record); static final String ZEEBE_RECORD_TEMPLATE_JSON; }
@Test public void shouldCreateIndexTemplates() { config.index.prefix = "foo-bar"; config.index.createTemplate = true; config.index.deployment = true; config.index.error = true; config.index.incident = true; config.index.job = true; config.index.jobBatch = true; config.index.message = true; config.index.messageSubscription = true; config.index.variable = true; config.index.variableDocument = true; config.index.workflowInstance = true; config.index.workflowInstanceCreation = true; config.index.workflowInstanceSubscription = true; createAndOpenExporter(); testHarness.export(); verify(esClient).putIndexTemplate("foo-bar", "foo-bar", ZEEBE_RECORD_TEMPLATE_JSON); verify(esClient).putIndexTemplate(ValueType.DEPLOYMENT); verify(esClient).putIndexTemplate(ValueType.ERROR); verify(esClient).putIndexTemplate(ValueType.INCIDENT); verify(esClient).putIndexTemplate(ValueType.JOB); verify(esClient).putIndexTemplate(ValueType.JOB_BATCH); verify(esClient).putIndexTemplate(ValueType.MESSAGE); verify(esClient).putIndexTemplate(ValueType.MESSAGE_SUBSCRIPTION); verify(esClient).putIndexTemplate(ValueType.VARIABLE); verify(esClient).putIndexTemplate(ValueType.VARIABLE_DOCUMENT); verify(esClient).putIndexTemplate(ValueType.WORKFLOW_INSTANCE); verify(esClient).putIndexTemplate(ValueType.WORKFLOW_INSTANCE_CREATION); verify(esClient).putIndexTemplate(ValueType.WORKFLOW_INSTANCE_SUBSCRIPTION); }
@Test public void shouldUpdateLastPositionOnFlush() { when(esClient.shouldFlush()).thenReturn(true); createAndOpenExporter(); final Record record = testHarness.export( r -> r.getMetadata() .setValueType(ValueType.WORKFLOW_INSTANCE) .setRecordType(RecordType.EVENT)); assertThat(testHarness.getController().getPosition()).isEqualTo(record.getPosition()); }
@Test public void shouldUpdatePositionAfterDelay() { config.index.event = true; createAndOpenExporter(); final List<Record> exported = testHarness.stream( r -> r.getMetadata() .setValueType(ValueType.WORKFLOW_INSTANCE) .setRecordType(RecordType.EVENT)) .export(4); testHarness.getController().runScheduledTasks(Duration.ofSeconds(config.bulk.delay)); verify(esClient, times(4)).index(any()); assertThat(testHarness.getController().getPosition()).isEqualTo(exported.get(3).getPosition()); }