CassandraIO { public static <T> Write<T> write() { return Write.<T>builder(MutationType.WRITE).build(); } private  CassandraIO(); static Read<T> read(); static Write<T> write(); static Write<T> delete();  }
@Test public void testWrite() { ArrayList<ScientistWrite> data = new ArrayList<>(); for (int i = 0; i < NUM_ROWS; i++) { ScientistWrite scientist = new ScientistWrite(); scientist.id = i; scientist.name = "Name " + i; data.add(scientist); } pipeline .apply(Create.of(data)) .apply( CassandraIO.<ScientistWrite>write() .withHosts(Collections.singletonList(CASSANDRA_HOST)) .withPort(cassandraPort) .withKeyspace(CASSANDRA_KEYSPACE) .withEntity(ScientistWrite.class)); pipeline.run(); List<Row> results = getRows(CASSANDRA_TABLE_WRITE); assertEquals(NUM_ROWS, results.size()); for (Row row : results) { assertTrue(row.getString("person_name").matches("Name (\\d*)")); } }
@Test public void testCustomMapperImplWrite() throws Exception { counter.set(0); SerializableFunction<Session, Mapper> factory = new NOOPMapperFactory(); pipeline .apply(Create.of("")) .apply( CassandraIO.<String>write() .withHosts(Collections.singletonList(CASSANDRA_HOST)) .withPort(cassandraPort) .withKeyspace(CASSANDRA_KEYSPACE) .withMapperFactoryFn(factory) .withEntity(String.class)); pipeline.run(); assertEquals(1, counter.intValue()); }