private void printConnectorDescription(final ConnectorDescription description) {
    final ConnectorStateInfo status = description.getStatus();
    writer().println(String.format("%-20s : %s", "Name", status.name()));
    writer().println(String.format("%-20s : %s", "Class", description.getConnectorClass()));
    writer().println(String.format("%-20s : %s", "Type", description.getStatus().type()));
    writer().println(String.format("%-20s : %s", "State", status.connector().state()));
    writer().println(String.format("%-20s : %s", "WorkerId", status.connector().workerId()));
    if (!ObjectUtils.defaultIfNull(status.connector().trace(), "").isEmpty()) {
      writer().println(String.format("%-20s : %s", "Trace", status.connector().trace()));
    }

    if (!status.tasks().isEmpty()) {
      writer().println();
      final Table taskTable = new Table.Builder()
          .withColumnHeaders(ImmutableList.of("Task ID", "State", "Error Trace"))
          .withRows(status.tasks()
              .stream()
              .map(task -> ImmutableList.of(
                  String.valueOf(task.id()),
                  task.state(),
                  ObjectUtils.defaultIfNull(task.trace(), ""))))
          .build();
      taskTable.print(this);
    }

    if (!description.getSources().isEmpty()) {
      writer().println();
      final Table sourceTable = new Table.Builder()
          .withColumnHeaders("KSQL Source Name", "Kafka Topic", "Type")
          .withRows(description.getSources()
              .stream()
              .map(source -> ImmutableList
                  .of(source.getName(), source.getTopic(), source.getType())))
          .build();
      sourceTable.print(this);
    }

    if (!description.getTopics().isEmpty()) {
      writer().println();
      final Table topicTable = new Table.Builder()
          .withColumnHeaders("Related Topics")
          .withRows(description.getTopics().stream().map(ImmutableList::of))
          .build();
      topicTable.print(this);
    }
  }
@Test
  public void testPrintConnectorDescription() {
    // Given:
    final KsqlEntityList entityList = new KsqlEntityList(ImmutableList.of(
        new ConnectorDescription(
            "STATEMENT",
            "io.confluent.Connector",
            new ConnectorStateInfo(
                "name",
                new ConnectorState("state", "worker", "msg"),
                ImmutableList.of(
                    new TaskState(0, "task", "worker", "task_msg")
                ),
                ConnectorType.SOURCE),
            ImmutableList.of(sourceDescription),
            ImmutableList.of("a-jdbc-topic"),
            ImmutableList.of()
        )
    ));

    // When:
    console.printKsqlEntityList(entityList);

    // Then:
    final String output = terminal.getOutputString();
    Approvals.verify(output, approvalOptions);
  }