@Override
  public void write(final OutputStream out) {
    // CHECKSTYLE_RULES.ON: CyclomaticComplexity
    try {
      out.write("[".getBytes(StandardCharsets.UTF_8));
      write(out, buildHeader());

      final BlockingRowQueue rowQueue = queryMetadata.getRowQueue();

      while (!connectionClosed && queryMetadata.isRunning() && !limitReached && !complete) {
        final KeyValueMetadata<List<?>, GenericRow> row = rowQueue.poll(
            disconnectCheckInterval,
            TimeUnit.MILLISECONDS
        );
        if (row != null) {
          write(out, buildRow(row));
        } else {
          // If no new rows have been written, the user may have terminated the connection without
          // us knowing. Check by trying to write a single newline.
          out.write("\n".getBytes(StandardCharsets.UTF_8));
          out.flush();
        }
        drainAndThrowOnError(out);
      }

      if (connectionClosed) {
        return;
      }

      drain(out);

      if (limitReached) {
        objectMapper.writeValue(out, StreamedRow.finalMessage("Limit Reached"));
      } else if (complete) {
        objectMapper.writeValue(out, StreamedRow.finalMessage("Query Completed"));
      }

      out.write("]\n".getBytes(StandardCharsets.UTF_8));
      out.flush();
    } catch (final EOFException exception) {
      // The user has terminated the connection; we can stop writing
      log.warn("Query terminated due to exception:" + exception.toString());
    } catch (final InterruptedException exception) {
      // The most likely cause of this is the server shutting down. Should just try to close
      // gracefully, without writing any more to the connection stream.
      log.warn("Interrupted while writing to connection stream");
    } catch (final Exception exception) {
      log.error("Exception occurred while writing to connection stream: ", exception);
      outputException(out, exception);
    } finally {
      close();
    }
  }
@Test
  public void shouldHandleTableRows() {
    // Given:
    when(queryMetadata.getResultType()).thenReturn(ResultType.TABLE);

    doAnswer(tableRows("key1", "Row1", "key2", null, "key3", "Row3"))
        .when(rowQueue).drainTo(any());

    createWriter();

    forceWriterToNotBlock();

    // When:
    writer.write(out);

    // Then:
    final List<String> lines = getOutput(out);
    assertThat(lines, hasItems(
        containsString("{\"row\":{\"columns\":[\"Row1\"]}}"),
        containsString("{\"row\":{\"columns\":[null],\"tombstone\":true}}"),
        containsString("{\"row\":{\"columns\":[\"Row3\"]}}")
    ));
  }