CassandraSinkBase extends RichSinkFunction<IN> implements CheckpointedFunction { @Override public void open(Configuration configuration) { this.callback = new FutureCallback<V>() { @Override public void onSuccess(V ignored) { int pending = updatesPending.decrementAndGet(); if (pending == 0) { synchronized (updatesPending) { updatesPending.notifyAll(); } } } @Override public void onFailure(Throwable t) { int pending = updatesPending.decrementAndGet(); if (pending == 0) { synchronized (updatesPending) { updatesPending.notifyAll(); } } exception = t; log.error("Error while sending value.", t); } }; this.cluster = builder.getCluster(); this.session = cluster.connect(); } CassandraSinkBase(ClusterBuilder builder); @Override void open(Configuration configuration); @Override void invoke(IN value); abstract ListenableFuture<V> send(IN value); @Override void close(); @Override void initializeState(FunctionInitializationContext context); @Override void snapshotState(FunctionSnapshotContext ctx);  }
@Test(expected = NoHostAvailableException.class) public void testHostNotFoundErrorHandling() throws Exception { CassandraSinkBase base = new CassandraSinkBase(new ClusterBuilder() { @Override protected Cluster buildCluster(Cluster.Builder builder) { return builder .addContactPoint("127.0.0.1") .withoutJMXReporting() .withoutMetrics().build(); } }) { @Override public ListenableFuture send(Object value) { return null; } }; base.open(new Configuration()); }