@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()); }
@Override public void export(final Record<?> record) { if (!indexTemplatesCreated) { createIndexTemplates(); } client.index(record); lastPosition = record.getPosition(); if (client.shouldFlush()) { flush(); } }
ElasticsearchExporter implements Exporter { @Override public void export(final Record<?> record) { if (!indexTemplatesCreated) { createIndexTemplates(); } client.index(record); lastPosition = record.getPosition(); if (client.shouldFlush()) { flush(); } } }
ElasticsearchExporter implements Exporter { @Override public void export(final Record<?> record) { if (!indexTemplatesCreated) { createIndexTemplates(); } client.index(record); lastPosition = record.getPosition(); if (client.shouldFlush()) { flush(); } }  }
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); }
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; }