CassandraIO { public static <T> Write<T> delete() { return Write.<T>builder(MutationType.DELETE).build(); } private  CassandraIO(); static Read<T> read(); static Write<T> write(); static Write<T> delete();  }
@Test public void testCustomMapperImplDelete() { counter.set(0); SerializableFunction<Session, Mapper> factory = new NOOPMapperFactory(); pipeline .apply(Create.of("")) .apply( CassandraIO.<String>delete() .withHosts(Collections.singletonList(CASSANDRA_HOST)) .withPort(cassandraPort) .withKeyspace(CASSANDRA_KEYSPACE) .withMapperFactoryFn(factory) .withEntity(String.class)); pipeline.run(); assertEquals(1, counter.intValue()); }
@Test public void testDelete() throws Exception { List<Row> results = getRows(CASSANDRA_TABLE); assertEquals(NUM_ROWS, results.size()); Scientist einstein = new Scientist(); einstein.id = 0; einstein.name = "Einstein"; pipeline .apply(Create.of(einstein)) .apply( CassandraIO.<Scientist>delete() .withHosts(Collections.singletonList(CASSANDRA_HOST)) .withPort(cassandraPort) .withKeyspace(CASSANDRA_KEYSPACE) .withEntity(Scientist.class)); pipeline.run(); results = getRows(CASSANDRA_TABLE); assertEquals(NUM_ROWS - 1, results.size()); session.execute( String.format( "INSERT INTO %s.%s(person_id, person_name) values(" + einstein.id + ", '" + einstein.name + "');", CASSANDRA_KEYSPACE, CASSANDRA_TABLE)); }