@SuppressWarnings("ResultOfMethodCallIgnored")
@Test(timeout = 10000)
public void testAsyncErrorRethrownAfterFlush() throws Throwable {

	final DummyFlinkKinesisProducer<String> producer = new DummyFlinkKinesisProducer<>(new SimpleStringSchema());
	OneInputStreamOperatorTestHarness<String, Object> testHarness =
		new OneInputStreamOperatorTestHarness<>(new StreamSink<>(producer));
	testHarness.open();
	testHarness.processElement(new StreamRecord<>("msg-1"));
	testHarness.processElement(new StreamRecord<>("msg-2"));
	testHarness.processElement(new StreamRecord<>("msg-3"));
	// only let the first record succeed for now
	UserRecordResult result = mock(UserRecordResult.class);
	when(result.isSuccessful()).thenReturn(true);
	producer.getPendingRecordFutures().get(0).set(result);
	CheckedThread snapshotThread = new CheckedThread() {
		@Override
		public void go() throws Exception {
			// this should block at first, since there are still two pending records that needs to be flushed
			testHarness.snapshot(123L, 123L);
		}
	};
	snapshotThread.start();
	// let the 2nd message fail with an async exception
	producer.getPendingRecordFutures().get(1).setException(new Exception("artificial async failure for 2nd message"));
	producer.getPendingRecordFutures().get(2).set(mock(UserRecordResult.class));
	try {
		snapshotThread.sync();
	} catch (Exception e) {
		// after the flush, the async exception should have been rethrown
		Assert.assertTrue(ExceptionUtils.findThrowableWithMessage(e, "artificial async failure for 2nd message").isPresent());
		// test succeeded
		return;
	}
	Assert.fail();
}