### Question:
CassandraSinkBase extends RichSinkFunction<IN> implements CheckpointedFunction { @Override public void invoke(IN value) throws Exception { checkAsyncErrors(); tryAcquire(); final ListenableFuture<V> result = send(value); Futures.addCallback(result, callback); } CassandraSinkBase(ClusterBuilder builder, CassandraSinkBaseConfig config, CassandraFailureHandler failureHandler); @Override void open(Configuration configuration); @Override void close(); @Override void initializeState(FunctionInitializationContext context); @Override void snapshotState(FunctionSnapshotContext ctx); @Override void invoke(IN value); abstract ListenableFuture<V> send(IN value);  }### Answer:
@Test(timeout = DEFAULT_TEST_TIMEOUT) public void testIgnoreError() throws Exception { Exception cause = new RuntimeException(); CassandraFailureHandler failureHandler = failure -> Assert.assertEquals(cause, failure); try (TestCassandraSink casSinkFunc = createOpenedTestCassandraSink(failureHandler)) { casSinkFunc.enqueueCompletableFuture(FutureUtils.getFailedFuture(cause)); casSinkFunc.enqueueCompletableFuture(FutureUtils.getFailedFuture(cause)); casSinkFunc.invoke("hello"); casSinkFunc.invoke("world"); } }
@Test(timeout = DEFAULT_TEST_TIMEOUT) public void testThrowErrorOnSnapshot() throws Exception { TestCassandraSink casSinkFunc = new TestCassandraSink(); try (OneInputStreamOperatorTestHarness<String, Object> testHarness = createOpenedTestHarness(casSinkFunc)) { Exception cause = new RuntimeException(); casSinkFunc.enqueueCompletableFuture(FutureUtils.getFailedFuture(cause)); casSinkFunc.invoke("hello"); try { testHarness.snapshot(123L, 123L); Assert.fail(); } catch (Exception e) { Assert.assertTrue(e.getCause() instanceof IOException); } } }
@Test(timeout = DEFAULT_TEST_TIMEOUT) public void testTimeoutExceptionOnInvoke() throws Exception { final CassandraSinkBaseConfig config = CassandraSinkBaseConfig.newBuilder() .setMaxConcurrentRequests(1) .setMaxConcurrentRequestsTimeout(Duration.ofMillis(1)) .build(); try (TestCassandraSink testCassandraSink = createOpenedTestCassandraSink(config)) { CompletableFuture<ResultSet> completableFuture = new CompletableFuture<>(); testCassandraSink.enqueueCompletableFuture(completableFuture); testCassandraSink.enqueueCompletableFuture(completableFuture); testCassandraSink.invoke("Invoke #1"); try { testCassandraSink.invoke("Invoke #2"); Assert.fail("Sending value should have experienced a TimeoutException"); } catch (Exception e) { Assert.assertTrue(e instanceof TimeoutException); } finally { completableFuture.complete(null); } } }