@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"); } }
@Override public void invoke(IN value) throws Exception { checkAsyncErrors(); tryAcquire(); final ListenableFuture<V> result = send(value); Futures.addCallback(result, callback); }
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 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); }
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); }
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);  }