@Test public void shouldThrowExceptionIfFailToFlushBulk() { final int bulkSize = 10; final Record<VariableRecordValue> recordMock = mock(Record.class); when(recordMock.getPartitionId()).thenReturn(1); when(recordMock.getValueType()).thenReturn(ValueType.WORKFLOW_INSTANCE); IntStream.range(0, bulkSize) .forEach( i -> { when(recordMock.getKey()).thenReturn(RECORD_KEY + i); when(recordMock.toJson()).thenReturn("invalid-json-" + i); client.index(recordMock); }); when(recordMock.getKey()).thenReturn(RECORD_KEY + bulkSize); when(recordMock.toJson()).thenReturn("{}"); client.index(recordMock); assertThatThrownBy(client::flush) .isInstanceOf(ElasticsearchExporterException.class) .hasMessage("Failed to flush all items of the bulk"); verify(logSpy) .warn( "Failed to flush {} item(s) of bulk request [type: {}, reason: {}]", bulkSize, "mapper_parsing_exception", "failed to parse"); }
public void index(final Record<?> record) { if (metrics == null) { metrics = new ElasticsearchMetrics(record.getPartitionId()); } checkRecord(record); bulk(newIndexCommand(record), record); }
ElasticsearchClient { public void index(final Record<?> record) { if (metrics == null) { metrics = new ElasticsearchMetrics(record.getPartitionId()); } checkRecord(record); bulk(newIndexCommand(record), record); } }
ElasticsearchClient { public void index(final Record<?> record) { if (metrics == null) { metrics = new ElasticsearchMetrics(record.getPartitionId()); } checkRecord(record); bulk(newIndexCommand(record), record); } ElasticsearchClient(
      final ElasticsearchExporterConfiguration configuration, final Logger log);   ElasticsearchClient(
      final ElasticsearchExporterConfiguration configuration,
      final Logger log,
      final List<String> bulkRequest); }
ElasticsearchClient { public void index(final Record<?> record) { if (metrics == null) { metrics = new ElasticsearchMetrics(record.getPartitionId()); } checkRecord(record); bulk(newIndexCommand(record), record); } ElasticsearchClient(
      final ElasticsearchExporterConfiguration configuration, final Logger log);   ElasticsearchClient(
      final ElasticsearchExporterConfiguration configuration,
      final Logger log,
      final List<String> bulkRequest); void close(); void index(final Record<?> record); void bulk(final Map<String, Object> command, final Record<?> record); void flush(); boolean shouldFlush(); boolean putIndexTemplate(final ValueType valueType); boolean putIndexTemplate(
      final String templateName, final String aliasName, final String filename); }
ElasticsearchClient { public void index(final Record<?> record) { if (metrics == null) { metrics = new ElasticsearchMetrics(record.getPartitionId()); } checkRecord(record); bulk(newIndexCommand(record), record); } ElasticsearchClient(
      final ElasticsearchExporterConfiguration configuration, final Logger log);   ElasticsearchClient(
      final ElasticsearchExporterConfiguration configuration,
      final Logger log,
      final List<String> bulkRequest); void close(); void index(final Record<?> record); void bulk(final Map<String, Object> command, final Record<?> record); void flush(); boolean shouldFlush(); boolean putIndexTemplate(final ValueType valueType); boolean putIndexTemplate(
      final String templateName, final String aliasName, final String filename); static final String INDEX_TEMPLATE_FILENAME_PATTERN; static final String INDEX_DELIMITER; static final String ALIAS_DELIMITER; }