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; }
@Test public void shouldNotLogWarningWhenIndexingSmallVariableValue() { final String variableValue = "x".repeat(configuration.index.ignoreVariablesAbove); final Record<VariableRecordValue> recordMock = mock(Record.class); when(recordMock.getPartitionId()).thenReturn(1); when(recordMock.getKey()).thenReturn(RECORD_KEY); when(recordMock.getValueType()).thenReturn(ValueType.VARIABLE); when(recordMock.toJson()).thenReturn("{}"); final VariableRecordValue value = mock(VariableRecordValue.class); when(value.getValue()).thenReturn(variableValue); when(recordMock.getValue()).thenReturn(value); client.index(recordMock); verify(logSpy, never()).warn(anyString(), ArgumentMatchers.<Object[]>any()); }
@Test public void shouldLogWarnWhenIndexingLargeVariableValue() { final String variableName = "varName"; final String variableValue = "x".repeat(configuration.index.ignoreVariablesAbove + 1); final long scopeKey = 1234L; final long workflowInstanceKey = 5678L; final Record<VariableRecordValue> recordMock = mock(Record.class); when(recordMock.getPartitionId()).thenReturn(1); when(recordMock.getKey()).thenReturn(RECORD_KEY); when(recordMock.getValueType()).thenReturn(ValueType.VARIABLE); when(recordMock.toJson()).thenReturn("{}"); final VariableRecordValue value = mock(VariableRecordValue.class); when(value.getName()).thenReturn(variableName); when(value.getValue()).thenReturn(variableValue); when(value.getScopeKey()).thenReturn(scopeKey); when(value.getWorkflowInstanceKey()).thenReturn(workflowInstanceKey); when(recordMock.getValue()).thenReturn(value); client.index(recordMock); final ArgumentCaptor<Object[]> argumentCaptor = ArgumentCaptor.forClass(Object[].class); verify(logSpy).warn(anyString(), argumentCaptor.capture()); final List<Object> args = new ArrayList<>(argumentCaptor.getAllValues()); assertThat(args) .contains( RECORD_KEY, variableName, variableValue.getBytes().length, scopeKey, workflowInstanceKey); }
@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"); }
@Test public void shouldIgnoreRecordIfDuplicateOfLast() { final Record<VariableRecordValue> recordMock = mock(Record.class); when(recordMock.getPartitionId()).thenReturn(1); when(recordMock.getValueType()).thenReturn(ValueType.WORKFLOW_INSTANCE); when(recordMock.getKey()).thenReturn(RECORD_KEY + 1); when(recordMock.toJson()).thenReturn("{}"); client.index(recordMock); assertThat(bulkRequest).hasSize(1); client.index(recordMock); assertThat(bulkRequest).hasSize(1); }