@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); } } }
@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);  }