### Question:
QueryCassandra extends AbstractCassandraProcessor { public static long convertToJsonStream(final ResultSet rs, final OutputStream outStream, Charset charset, long timeout, TimeUnit timeUnit) throws IOException, InterruptedException, TimeoutException, ExecutionException { try { outStream.write("{\"results\":[".getBytes(charset)); 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) { if (nrOfRows != 0) { outStream.write(",".getBytes(charset)); } outStream.write("{".getBytes(charset)); for (int i = 0; i < columnDefinitions.size(); i++) { final DataType dataType = columnDefinitions.getType(i); final String colName = columnDefinitions.getName(i); if (i != 0) { outStream.write(",".getBytes(charset)); } if (row.isNull(i)) { outStream.write(("\"" + colName + "\"" + ":null").getBytes(charset)); } else { Object value = getCassandraObject(row, i, dataType); String valueString; if (value instanceof List || value instanceof Set) { boolean first = true; StringBuilder sb = new StringBuilder("["); for (Object element : ((Collection) value)) { if (!first) { sb.append(","); } sb.append(getJsonElement(element)); first = false; } sb.append("]"); valueString = sb.toString(); } else if (value instanceof Map) { boolean first = true; StringBuilder sb = new StringBuilder("{"); for (Object element : ((Map) value).entrySet()) { Map.Entry entry = (Map.Entry) element; Object mapKey = entry.getKey(); Object mapValue = entry.getValue(); if (!first) { sb.append(","); } sb.append(getJsonElement(mapKey)); sb.append(":"); sb.append(getJsonElement(mapValue)); first = false; } sb.append("}"); valueString = sb.toString(); } else { valueString = getJsonElement(value); } outStream.write(("\"" + colName + "\":" + valueString + "").getBytes(charset)); } } nrOfRows += 1; outStream.write("}".getBytes(charset)); } } while (!rs.isFullyFetched()); } return nrOfRows; } finally { outStream.write("]}".getBytes()); } }  @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; }### Answer:
@Test public void testConvertToJSONStream() throws Exception { ResultSet rs = CassandraQueryTestUtil.createMockResultSet(); ByteArrayOutputStream baos = new ByteArrayOutputStream(); long numberOfRows = QueryCassandra.convertToJsonStream(rs, baos, StandardCharsets.UTF_8, 0, null); assertEquals(2, numberOfRows); }