public Schema toKsqlSchema(final Schema schema) {
    try {
      final Schema rowSchema = toKsqlFieldSchema(schema);
      if (rowSchema.type() != Schema.Type.STRUCT) {
        throw new KsqlException("KSQL stream/table schema must be structured");
      }

      if (rowSchema.fields().isEmpty()) {
        throw new KsqlException("Schema does not include any columns with "
            + "types that ksqlDB supports."
            + System.lineSeparator()
            + "schema: " + FORMATTER.format(schema));
      }

      return rowSchema;
    } catch (final UnsupportedTypeException e) {
      throw new KsqlException("Unsupported type at root of schema: " + e.getMessage(), e);
    }
  }
@Test
  public void shouldTranslateTimeTypes() {
    final Schema connectSchema = SchemaBuilder
        .struct()
        .field("timefield", Time.SCHEMA)
        .field("datefield", Date.SCHEMA)
        .field("timestampfield", Timestamp.SCHEMA)
        .build();

    final Schema ksqlSchema = translator.toKsqlSchema(connectSchema);

    assertThat(ksqlSchema.field(nameTranslator.apply("timefield")).schema(),
        equalTo(OPTIONAL_TIME_SCHEMA));
    assertThat(ksqlSchema.field(nameTranslator.apply("datefield")).schema(),
        equalTo(OPTIONAL_DATE_SCHEMA));
    assertThat(ksqlSchema.field(nameTranslator.apply("timestampfield")).schema(),
        equalTo(OPTIONAL_TIMESTAMP_SCHEMA));
  }