private void printSourceDescription(final SourceDescription source) {
    final boolean isTable = source.getType().equalsIgnoreCase("TABLE");

    writer().println(String.format("%-20s : %s", "Name", source.getName()));
    if (!source.isExtended()) {
      printSchema(source.getWindowType(), source.getFields(), isTable);
      writer().println(
          "For runtime statistics and query details run: DESCRIBE <Stream,Table> EXTENDED;");
      return;
    }
    writer().println(String.format("%-20s : %s", "Type", source.getType()));

    printTopicInfo(source);
    writer().println(String.format("%-20s : %s", "Statement", source.getStatement()));
    writer().println("");

    printSchema(source.getWindowType(), source.getFields(), isTable);

    printSourceConstraints(source.getSourceConstraints());

    printQueries(source.getReadQueries(), source.getType(), "read");

    printQueries(source.getWriteQueries(), source.getType(), "write");
    printStatistics(source);

    writer().println(String.format(
        "(%s)",
        "Statistics of the local KSQL server interaction with the Kafka topic "
            + source.getTopic()
    ));
    if (!source.getQueryOffsetSummaries().isEmpty()) {
      writer().println();
      writer().println("Consumer Groups summary:");
      for (QueryOffsetSummary entry : source.getQueryOffsetSummaries()) {
        writer().println();
        writer().println(String.format("%-20s : %s", "Consumer Group", entry.getGroupId()));
        if (entry.getTopicSummaries().isEmpty()) {
          writer().println("<no offsets committed by this group yet>");
        }
        for (QueryTopicOffsetSummary topicSummary : entry.getTopicSummaries()) {
          writer().println();
          writer().println(String.format("%-20s : %s",
              "Kafka topic", topicSummary.getKafkaTopic()));
          writer().println(String.format("%-20s : %s",
              "Max lag", topicSummary.getOffsets().stream()
                  .mapToLong(s -> s.getLogEndOffset() - s.getConsumerOffset())
                  .max()
                  .orElse(0)));
          writer().println("");
          final Table taskTable = new Table.Builder()
              .withColumnHeaders(
                  ImmutableList.of("Partition", "Start Offset", "End Offset", "Offset", "Lag"))
              .withRows(topicSummary.getOffsets()
                  .stream()
                  .map(offset -> ImmutableList.of(
                      String.valueOf(offset.getPartition()),
                      String.valueOf(offset.getLogStartOffset()),
                      String.valueOf(offset.getLogEndOffset()),
                      String.valueOf(offset.getConsumerOffset()),
                      String.valueOf(offset.getLogEndOffset() - offset.getConsumerOffset())
                  )))
              .build();
          taskTable.print(this);
        }
      }
    }
  }
@Test
  public void testPrintSourceDescription() {
    // Given:
    final List<FieldInfo> fields = buildTestSchema(
        SqlTypes.BOOLEAN,
        SqlTypes.INTEGER,
        SqlTypes.BIGINT,
        SqlTypes.DOUBLE,
        SqlTypes.STRING,
        SqlTypes.array(SqlTypes.STRING),
        SqlTypes.map(SqlTypes.STRING, SqlTypes.BIGINT),
        SqlTypes.struct()
            .field("a", SqlTypes.DOUBLE)
            .build()
    );

    final List<RunningQuery> readQueries = ImmutableList.of(
        new RunningQuery("read query", ImmutableSet.of("sink1"), ImmutableSet.of("sink1 topic"), new QueryId("readId"), queryStatusCount, KsqlConstants.KsqlQueryType.PERSISTENT)
    );
    final List<RunningQuery> writeQueries = ImmutableList.of(
        new RunningQuery("write query", ImmutableSet.of("sink2"), ImmutableSet.of("sink2 topic"), new QueryId("writeId"), queryStatusCount, KsqlConstants.KsqlQueryType.PERSISTENT)
    );

    final KsqlEntityList entityList = new KsqlEntityList(ImmutableList.of(
        new SourceDescriptionEntity(
            "some sql",
            buildSourceDescription(readQueries, writeQueries, fields, false),
            Collections.emptyList()
        )
    ));

    // When:
    console.printKsqlEntityList(entityList);

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