ElasticsearchSinkBase extends RichSinkFunction<T> implements CheckpointedFunction { @Override public void open(Configuration parameters) throws Exception { client = callBridge.createClient(userConfig); bulkProcessor = buildBulkProcessor(new BulkProcessorListener()); requestIndexer = new BulkProcessorIndexer(bulkProcessor, flushOnCheckpoint, numPendingRequests); } ElasticsearchSinkBase(
		ElasticsearchApiCallBridge callBridge,
		Map<String, String> userConfig,
		ElasticsearchSinkFunction<T> elasticsearchSinkFunction,
		ActionRequestFailureHandler failureHandler); void disableFlushOnCheckpoint(); @Override void open(Configuration parameters); @Override void invoke(T value); @Override void initializeState(FunctionInitializationContext context); @Override void snapshotState(FunctionSnapshotContext context); @Override void close(); static final String CONFIG_KEY_BULK_FLUSH_MAX_ACTIONS; static final String CONFIG_KEY_BULK_FLUSH_MAX_SIZE_MB; static final String CONFIG_KEY_BULK_FLUSH_INTERVAL_MS; static final String CONFIG_KEY_BULK_FLUSH_BACKOFF_ENABLE; static final String CONFIG_KEY_BULK_FLUSH_BACKOFF_TYPE; static final String CONFIG_KEY_BULK_FLUSH_BACKOFF_RETRIES; static final String CONFIG_KEY_BULK_FLUSH_BACKOFF_DELAY; }
@Test public void testItemFailureRethrownOnInvoke() throws Throwable { final DummyElasticsearchSink<String> sink = new DummyElasticsearchSink<>( new HashMap<String, String>(), new SimpleSinkFunction<String>(), new NoOpFailureHandler()); final OneInputStreamOperatorTestHarness<String, Object> testHarness = new OneInputStreamOperatorTestHarness<>(new StreamSink<>(sink)); testHarness.open(); sink.setMockItemFailuresListForNextBulkItemResponses(Collections.singletonList(new Exception("artificial failure for record"))); testHarness.processElement(new StreamRecord<>("msg")); verify(sink.getMockBulkProcessor(), times(1)).add(any(ActionRequest.class)); sink.manualBulkRequestWithAllPendingRequests(); try { testHarness.processElement(new StreamRecord<>("next msg")); } catch (Exception e) { Assert.assertTrue(e.getCause().getMessage().contains("artificial failure for record")); return; } Assert.fail(); }
@Test public void testItemFailureRethrownOnCheckpoint() throws Throwable { final DummyElasticsearchSink<String> sink = new DummyElasticsearchSink<>( new HashMap<String, String>(), new SimpleSinkFunction<String>(), new NoOpFailureHandler()); final OneInputStreamOperatorTestHarness<String, Object> testHarness = new OneInputStreamOperatorTestHarness<>(new StreamSink<>(sink)); testHarness.open(); sink.setMockItemFailuresListForNextBulkItemResponses(Collections.singletonList(new Exception("artificial failure for record"))); testHarness.processElement(new StreamRecord<>("msg")); verify(sink.getMockBulkProcessor(), times(1)).add(any(ActionRequest.class)); sink.manualBulkRequestWithAllPendingRequests(); try { testHarness.snapshot(1L, 1000L); } catch (Exception e) { Assert.assertTrue(e.getCause().getCause().getMessage().contains("artificial failure for record")); return; } Assert.fail(); }
@Test(timeout = 5000) public void testItemFailureRethrownOnCheckpointAfterFlush() throws Throwable { final DummyElasticsearchSink<String> sink = new DummyElasticsearchSink<>( new HashMap<String, String>(), new SimpleSinkFunction<String>(), new NoOpFailureHandler()); final OneInputStreamOperatorTestHarness<String, Object> testHarness = new OneInputStreamOperatorTestHarness<>(new StreamSink<>(sink)); testHarness.open(); List<Exception> mockResponsesList = new ArrayList<>(2); mockResponsesList.add(null); mockResponsesList.add(new Exception("artificial failure for record")); sink.setMockItemFailuresListForNextBulkItemResponses(mockResponsesList); testHarness.processElement(new StreamRecord<>("msg-1")); verify(sink.getMockBulkProcessor(), times(1)).add(any(ActionRequest.class)); sink.manualBulkRequestWithAllPendingRequests(); testHarness.processElement(new StreamRecord<>("msg-2")); testHarness.processElement(new StreamRecord<>("msg-3")); verify(sink.getMockBulkProcessor(), times(3)).add(any(ActionRequest.class)); CheckedThread snapshotThread = new CheckedThread() { @Override public void go() throws Exception { testHarness.snapshot(1L, 1000L); } }; snapshotThread.start(); while (snapshotThread.getState() != Thread.State.WAITING) { Thread.sleep(10); } sink.continueFlush(); try { snapshotThread.sync(); } catch (Exception e) { Assert.assertTrue(e.getCause().getCause().getMessage().contains("artificial failure for record")); return; } Assert.fail(); }
@Test public void testBulkFailureRethrownOnInvoke() throws Throwable { final DummyElasticsearchSink<String> sink = new DummyElasticsearchSink<>( new HashMap<String, String>(), new SimpleSinkFunction<String>(), new NoOpFailureHandler()); final OneInputStreamOperatorTestHarness<String, Object> testHarness = new OneInputStreamOperatorTestHarness<>(new StreamSink<>(sink)); testHarness.open(); sink.setFailNextBulkRequestCompletely(new Exception("artificial failure for bulk request")); testHarness.processElement(new StreamRecord<>("msg")); verify(sink.getMockBulkProcessor(), times(1)).add(any(ActionRequest.class)); sink.manualBulkRequestWithAllPendingRequests(); try { testHarness.processElement(new StreamRecord<>("next msg")); } catch (Exception e) { Assert.assertTrue(e.getCause().getMessage().contains("artificial failure for bulk request")); return; } Assert.fail(); }
@Test public void testBulkFailureRethrownOnCheckpoint() throws Throwable { final DummyElasticsearchSink<String> sink = new DummyElasticsearchSink<>( new HashMap<String, String>(), new SimpleSinkFunction<String>(), new NoOpFailureHandler()); final OneInputStreamOperatorTestHarness<String, Object> testHarness = new OneInputStreamOperatorTestHarness<>(new StreamSink<>(sink)); testHarness.open(); sink.setFailNextBulkRequestCompletely(new Exception("artificial failure for bulk request")); testHarness.processElement(new StreamRecord<>("msg")); verify(sink.getMockBulkProcessor(), times(1)).add(any(ActionRequest.class)); sink.manualBulkRequestWithAllPendingRequests(); try { testHarness.snapshot(1L, 1000L); } catch (Exception e) { Assert.assertTrue(e.getCause().getCause().getMessage().contains("artificial failure for bulk request")); return; } Assert.fail(); }
@Test(timeout = 5000) public void testBulkFailureRethrownOnOnCheckpointAfterFlush() throws Throwable { final DummyElasticsearchSink<String> sink = new DummyElasticsearchSink<>( new HashMap<String, String>(), new SimpleSinkFunction<String>(), new NoOpFailureHandler()); final OneInputStreamOperatorTestHarness<String, Object> testHarness = new OneInputStreamOperatorTestHarness<>(new StreamSink<>(sink)); testHarness.open(); sink.setMockItemFailuresListForNextBulkItemResponses(Collections.singletonList((Exception) null)); testHarness.processElement(new StreamRecord<>("msg-1")); verify(sink.getMockBulkProcessor(), times(1)).add(any(ActionRequest.class)); sink.manualBulkRequestWithAllPendingRequests(); testHarness.processElement(new StreamRecord<>("msg-2")); testHarness.processElement(new StreamRecord<>("msg-3")); verify(sink.getMockBulkProcessor(), times(3)).add(any(ActionRequest.class)); CheckedThread snapshotThread = new CheckedThread() { @Override public void go() throws Exception { testHarness.snapshot(1L, 1000L); } }; snapshotThread.start(); while (snapshotThread.getState() != Thread.State.WAITING) { Thread.sleep(10); } sink.setFailNextBulkRequestCompletely(new Exception("artificial failure for bulk request")); sink.continueFlush(); try { snapshotThread.sync(); } catch (Exception e) { Assert.assertTrue(e.getCause().getCause().getMessage().contains("artificial failure for bulk request")); return; } Assert.fail(); }