@Test public void testConvertToAvroStream() throws Exception { ResultSet rs = CassandraQueryTestUtil.createMockResultSet(); ByteArrayOutputStream baos = new ByteArrayOutputStream(); long numberOfRows = QueryCassandra.convertToAvroStream(rs, baos, 0, null); assertEquals(2, numberOfRows); }
public static long convertToAvroStream(final ResultSet rs, final OutputStream outStream, long timeout, TimeUnit timeUnit) throws IOException, InterruptedException, TimeoutException, ExecutionException { final Schema schema = createSchema(rs); final GenericRecord rec = new GenericData.Record(schema); final DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema); try (final DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(datumWriter)) { dataFileWriter.create(schema, outStream); final ColumnDefinitions columnDefinitions = rs.getColumnDefinitions(); long nrOfRows = 0; if (columnDefinitions != null) { do { int rowsAvailableWithoutFetching = rs.getAvailableWithoutFetching(); if (rowsAvailableWithoutFetching == 0) { if (timeout <= 0 || timeUnit == null) { rs.fetchMoreResults().get(); } else { rs.fetchMoreResults().get(timeout, timeUnit); } } for (Row row : rs) { for (int i = 0; i < columnDefinitions.size(); i++) { final DataType dataType = columnDefinitions.getType(i); if (row.isNull(i)) { rec.put(i, null); } else { rec.put(i, getCassandraObject(row, i, dataType)); } } dataFileWriter.append(rec); nrOfRows += 1; } } while (!rs.isFullyFetched()); } return nrOfRows; } }
QueryCassandra extends AbstractCassandraProcessor { public static long convertToAvroStream(final ResultSet rs, final OutputStream outStream, long timeout, TimeUnit timeUnit) throws IOException, InterruptedException, TimeoutException, ExecutionException { final Schema schema = createSchema(rs); final GenericRecord rec = new GenericData.Record(schema); final DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema); try (final DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(datumWriter)) { dataFileWriter.create(schema, outStream); final ColumnDefinitions columnDefinitions = rs.getColumnDefinitions(); long nrOfRows = 0; if (columnDefinitions != null) { do { int rowsAvailableWithoutFetching = rs.getAvailableWithoutFetching(); if (rowsAvailableWithoutFetching == 0) { if (timeout <= 0 || timeUnit == null) { rs.fetchMoreResults().get(); } else { rs.fetchMoreResults().get(timeout, timeUnit); } } for (Row row : rs) { for (int i = 0; i < columnDefinitions.size(); i++) { final DataType dataType = columnDefinitions.getType(i); if (row.isNull(i)) { rec.put(i, null); } else { rec.put(i, getCassandraObject(row, i, dataType)); } } dataFileWriter.append(rec); nrOfRows += 1; } } while (!rs.isFullyFetched()); } return nrOfRows; } } }
QueryCassandra extends AbstractCassandraProcessor { public static long convertToAvroStream(final ResultSet rs, final OutputStream outStream, long timeout, TimeUnit timeUnit) throws IOException, InterruptedException, TimeoutException, ExecutionException { final Schema schema = createSchema(rs); final GenericRecord rec = new GenericData.Record(schema); final DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema); try (final DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(datumWriter)) { dataFileWriter.create(schema, outStream); final ColumnDefinitions columnDefinitions = rs.getColumnDefinitions(); long nrOfRows = 0; if (columnDefinitions != null) { do { int rowsAvailableWithoutFetching = rs.getAvailableWithoutFetching(); if (rowsAvailableWithoutFetching == 0) { if (timeout <= 0 || timeUnit == null) { rs.fetchMoreResults().get(); } else { rs.fetchMoreResults().get(timeout, timeUnit); } } for (Row row : rs) { for (int i = 0; i < columnDefinitions.size(); i++) { final DataType dataType = columnDefinitions.getType(i); if (row.isNull(i)) { rec.put(i, null); } else { rec.put(i, getCassandraObject(row, i, dataType)); } } dataFileWriter.append(rec); nrOfRows += 1; } } while (!rs.isFullyFetched()); } return nrOfRows; } }  }
QueryCassandra extends AbstractCassandraProcessor { public static long convertToAvroStream(final ResultSet rs, final OutputStream outStream, long timeout, TimeUnit timeUnit) throws IOException, InterruptedException, TimeoutException, ExecutionException { final Schema schema = createSchema(rs); final GenericRecord rec = new GenericData.Record(schema); final DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema); try (final DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(datumWriter)) { dataFileWriter.create(schema, outStream); final ColumnDefinitions columnDefinitions = rs.getColumnDefinitions(); long nrOfRows = 0; if (columnDefinitions != null) { do { int rowsAvailableWithoutFetching = rs.getAvailableWithoutFetching(); if (rowsAvailableWithoutFetching == 0) { if (timeout <= 0 || timeUnit == null) { rs.fetchMoreResults().get(); } else { rs.fetchMoreResults().get(timeout, timeUnit); } } for (Row row : rs) { for (int i = 0; i < columnDefinitions.size(); i++) { final DataType dataType = columnDefinitions.getType(i); if (row.isNull(i)) { rec.put(i, null); } else { rec.put(i, getCassandraObject(row, i, dataType)); } } dataFileWriter.append(rec); nrOfRows += 1; } } while (!rs.isFullyFetched()); } return nrOfRows; } }  @Override Set<Relationship> getRelationships(); @Override final List<PropertyDescriptor> getSupportedPropertyDescriptors(); @OnScheduled void onScheduled(final ProcessContext context); @Override void onTrigger(final ProcessContext context, final ProcessSession session); @OnUnscheduled void stop(); @OnShutdown void shutdown(); static long convertToAvroStream(final ResultSet rs, final OutputStream outStream,
                                           long timeout, TimeUnit timeUnit); static long convertToJsonStream(final ResultSet rs, final OutputStream outStream,
                                           Charset charset, long timeout, TimeUnit timeUnit); static Schema createSchema(final ResultSet rs); }
QueryCassandra extends AbstractCassandraProcessor { public static long convertToAvroStream(final ResultSet rs, final OutputStream outStream, long timeout, TimeUnit timeUnit) throws IOException, InterruptedException, TimeoutException, ExecutionException { final Schema schema = createSchema(rs); final GenericRecord rec = new GenericData.Record(schema); final DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(schema); try (final DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(datumWriter)) { dataFileWriter.create(schema, outStream); final ColumnDefinitions columnDefinitions = rs.getColumnDefinitions(); long nrOfRows = 0; if (columnDefinitions != null) { do { int rowsAvailableWithoutFetching = rs.getAvailableWithoutFetching(); if (rowsAvailableWithoutFetching == 0) { if (timeout <= 0 || timeUnit == null) { rs.fetchMoreResults().get(); } else { rs.fetchMoreResults().get(timeout, timeUnit); } } for (Row row : rs) { for (int i = 0; i < columnDefinitions.size(); i++) { final DataType dataType = columnDefinitions.getType(i); if (row.isNull(i)) { rec.put(i, null); } else { rec.put(i, getCassandraObject(row, i, dataType)); } } dataFileWriter.append(rec); nrOfRows += 1; } } while (!rs.isFullyFetched()); } return nrOfRows; } }  @Override Set<Relationship> getRelationships(); @Override final List<PropertyDescriptor> getSupportedPropertyDescriptors(); @OnScheduled void onScheduled(final ProcessContext context); @Override void onTrigger(final ProcessContext context, final ProcessSession session); @OnUnscheduled void stop(); @OnShutdown void shutdown(); static long convertToAvroStream(final ResultSet rs, final OutputStream outStream,
                                           long timeout, TimeUnit timeUnit); static long convertToJsonStream(final ResultSet rs, final OutputStream outStream,
                                           Charset charset, long timeout, TimeUnit timeUnit); static Schema createSchema(final ResultSet rs); static final String AVRO_FORMAT; static final String JSON_FORMAT; static final String RESULT_ROW_COUNT; static final PropertyDescriptor CQL_SELECT_QUERY; static final PropertyDescriptor QUERY_TIMEOUT; static final PropertyDescriptor FETCH_SIZE; static final PropertyDescriptor OUTPUT_FORMAT; static final Relationship REL_SUCCESS; static final Relationship REL_FAILURE; static final Relationship REL_RETRY; }