### Question:
CassandraSink extends AbstractSink implements Configurable { @Override public synchronized void stop() { if (session != null && !session.isClosed()) { try { session.close(); } catch (RuntimeException ex) { log.error("Error while closing session", ex); } } if (cluster != null && !cluster.isClosed()) { try { cluster.close(); } catch (RuntimeException ex) { log.error("Error while closing cluster", ex); } } this.sinkCounter.stop(); super.stop(); } CassandraSink(); @Override void configure(Context context); @Override synchronized void start(); @Override synchronized void stop(); @Override Status process();  }### Answer:
@Test public void stop() { final CassandraSink sink = new CassandraSink(); final Channel channel = mock(Channel.class); final Session session = mock(Session.class); final Cluster cluster = mock(Cluster.class); final Context ctx = new Context(); ctx.put("tables", "keyspace.table"); sink.configure(ctx); sink.setChannel(channel); sink.session = session; sink.cluster = cluster; sink.stop(); verify(session).isClosed(); verify(session).close(); verifyNoMoreInteractions(session); verify(cluster).isClosed(); verify(cluster).close(); verifyNoMoreInteractions(cluster); }