@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();
	// setup the next bulk request, and let bulk request succeed
	sink.setMockItemFailuresListForNextBulkItemResponses(Collections.singletonList((Exception) null));
	testHarness.processElement(new StreamRecord<>("msg-1"));
	verify(sink.getMockBulkProcessor(), times(1)).add(any(ActionRequest.class));
	// manually execute the next bulk request
	sink.manualBulkRequestWithAllPendingRequests();
	// setup the requests to be flushed in the snapshot
	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();
	// the snapshot should eventually be blocked before snapshot triggers flushing
	while (snapshotThread.getState() != Thread.State.WAITING) {
		Thread.sleep(10);
	}
	// for the snapshot-triggered flush, we let the bulk request fail completely
	sink.setFailNextBulkRequestCompletely(new Exception("artificial failure for bulk request"));
	// let the snapshot-triggered flush continue (bulk request should fail completely)
	sink.continueFlush();
	try {
		snapshotThread.sync();
	} catch (Exception e) {
		// the snapshot should have failed with the bulk request failure
		Assert.assertTrue(e.getCause().getCause().getMessage().contains("artificial failure for bulk request"));
		// test succeeded
		return;
	}
	Assert.fail();
}