@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();
	// setup the next bulk request, and let the whole bulk request fail
	sink.setFailNextBulkRequestCompletely(new Exception("artificial failure for bulk request"));
	testHarness.processElement(new StreamRecord<>("msg"));
	verify(sink.getMockBulkProcessor(), times(1)).add(any(ActionRequest.class));
	// manually execute the next bulk request
	sink.manualBulkRequestWithAllPendingRequests();
	try {
		testHarness.processElement(new StreamRecord<>("next msg"));
	} catch (Exception e) {
		// the invoke should have failed with the bulk request failure
		Assert.assertTrue(e.getCause().getMessage().contains("artificial failure for bulk request"));
		// test succeeded
		return;
	}
	Assert.fail();
}