ElasticBulkProcessorHandler { public static ElasticBulkProcessorHandler getBulkProcessorHandler(@NotNull ElasticConnection elasticConnection, @NotNull ElasticIndexDefinition indexDefinition, @NotNull NodeBuilder definitionBuilder, CommitInfo commitInfo) { PropertyState async = indexDefinition.getDefinitionNodeState().getProperty("async"); if (async != null) { Iterable<String> opt = async.getValue(Type.STRINGS); if (!commitInfo.getInfo().containsKey(IndexConstants.CHECKPOINT_CREATION_TIME)) { waitForESAcknowledgement = false; } return new ElasticBulkProcessorHandler(elasticConnection, indexDefinition, definitionBuilder); } String syncMode = null; if (commitInfo != null) { syncMode = (String) commitInfo.getInfo().get(SYNC_MODE_PROPERTY); } if (syncMode == null) { PropertyState syncModeProp = indexDefinition.getDefinitionNodeState().getProperty("sync-mode"); if (syncModeProp != null) { syncMode = syncModeProp.getValue(Type.STRING); } } if (SYNC_RT_MODE.equals(syncMode)) { return new RealTimeBulkProcessorHandler(elasticConnection, indexDefinition, definitionBuilder); } return new ElasticBulkProcessorHandler(elasticConnection, indexDefinition, definitionBuilder); } private  ElasticBulkProcessorHandler(@NotNull ElasticConnection elasticConnection,
                                        @NotNull ElasticIndexDefinition indexDefinition,
                                        @NotNull NodeBuilder definitionBuilder); static ElasticBulkProcessorHandler getBulkProcessorHandler(@NotNull ElasticConnection elasticConnection,
                                                                      @NotNull ElasticIndexDefinition indexDefinition,
                                                                      @NotNull NodeBuilder definitionBuilder, CommitInfo commitInfo); void add(DocWriteRequest<?> request); boolean close();  }
@Test public void defaultMode() { when(definitionNodeStateMock.getProperty(eq("async"))).thenReturn(null); ElasticBulkProcessorHandler bulkProcessorHandler = ElasticBulkProcessorHandler .getBulkProcessorHandler(elasticConnectionMock, indexDefinitionMock, definitionBuilder, commitInfo); assertThat(bulkProcessorHandler, instanceOf(ElasticBulkProcessorHandler.class)); }
@Test(expected = IllegalStateException.class) public void multiSyncModes() { when(definitionNodeStateMock.getProperty(eq("async"))).thenReturn(null); when(definitionNodeStateMock.getProperty(eq("sync-mode"))) .thenReturn(new MultiStringPropertyState("sync-mode", Arrays.asList("nrt", "rt"))); ElasticBulkProcessorHandler .getBulkProcessorHandler(elasticConnectionMock, indexDefinitionMock, definitionBuilder, commitInfo); }
@Test public void rtMode() { when(definitionNodeStateMock.getProperty(eq("async"))).thenReturn(null); when(definitionNodeStateMock.getProperty(eq("sync-mode"))) .thenReturn(new StringPropertyState("sync-mode", "rt")); ElasticBulkProcessorHandler bulkProcessorHandler = ElasticBulkProcessorHandler .getBulkProcessorHandler(elasticConnectionMock, indexDefinitionMock, definitionBuilder, commitInfo); assertThat(bulkProcessorHandler, instanceOf(ElasticBulkProcessorHandler.RealTimeBulkProcessorHandler.class)); }
@Test public void defaultModeWithCommitInfoOverride() { when(definitionNodeStateMock.getProperty(eq("async"))).thenReturn(null); when(commitInfo.getInfo()).thenReturn(Collections.singletonMap("sync-mode", "rt")); ElasticBulkProcessorHandler bulkProcessorHandler = ElasticBulkProcessorHandler .getBulkProcessorHandler(elasticConnectionMock, indexDefinitionMock, definitionBuilder, commitInfo); assertThat(bulkProcessorHandler, instanceOf(ElasticBulkProcessorHandler.RealTimeBulkProcessorHandler.class)); }