@Override
  public void write(final OutputStream out) {
    try {
      final PrintStream print = new PrintStream(out, true, "UTF8");
      final RecordFormatter formatter = new RecordFormatter(schemaRegistryClient, topicName);

      final FormatsTracker formatsTracker = new FormatsTracker(print);
      while (!connectionClosed && !print.checkError() && !limitReached.test(messagesWritten)) {
        final ConsumerRecords<Bytes, Bytes> records = topicConsumer.poll(disconnectCheckInterval);
        if (records.isEmpty()) {
          print.println();
          continue;
        }

        final List<String> values = formatter.format(records.records(topicName));
        if (values.isEmpty()) {
          continue;
        }

        final List<String> toOutput = new ArrayList<>();
        for (final String value : values) {
          if (messagesPolled++ % interval == 0) {
            messagesWritten++;
            toOutput.add(value);
          }

          if (limitReached.test(messagesWritten)) {
            break;
          }
        }

        formatsTracker.update(formatter);

        toOutput.forEach(print::println);
      }
    } catch (final Exception exception) {
      log.error("Exception encountered while writing to output stream", exception);
      outputException(out, exception);
    } finally {
      close();
    }
  }
@Test
  public void shouldIntervalTwoAndLimitTwo() {
    // Given:
    final TopicStreamWriter writer = new TopicStreamWriter(
        schemaRegistry,
        kafkaConsumer,
        "topic",
        2,
        Duration.ZERO,
        OptionalInt.of(2),
        new CompletableFuture<>()
    );

    // When:
    final ValidatingOutputStream out = new ValidatingOutputStream();
    writer.write(out);

    // Then:
    final List<String> expected = ImmutableList.of(
        "Key format: ",
        "KAFKA_STRING",
        System.lineSeparator(),
        "Value format: ",
        "KAFKA_STRING",
        System.lineSeparator(),
        "rowtime: N/A, key: key-0, value: value-0, partition: 0",
        System.lineSeparator(),
        "rowtime: N/A, key: key-2, value: value-2, partition: 0",
        System.lineSeparator()
    );
    out.assertWrites(expected);
  }