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 shouldTranslatePrimitives() {
    final Schema connectSchema = SchemaBuilder
        .struct()
        .field("intField", Schema.INT32_SCHEMA)
        .field("longField", Schema.INT64_SCHEMA)
        .field("doubleField", Schema.FLOAT64_SCHEMA)
        .field("stringField", Schema.STRING_SCHEMA)
        .field("booleanField", Schema.BOOLEAN_SCHEMA)
        .field("bytesField", Schema.BYTES_SCHEMA)
        .build();

    final Schema ksqlSchema = translator.toKsqlSchema(connectSchema);
    assertThat(ksqlSchema.schema().type(), equalTo(Schema.Type.STRUCT));
    assertThat(ksqlSchema.fields().size(), equalTo(connectSchema.fields().size()));
    for (int i = 0; i < ksqlSchema.fields().size(); i++) {
      assertThat(
          ksqlSchema.fields().get(i).name(),
          equalTo(nameTranslator.apply(connectSchema.fields().get(i).name())));
      assertThat(
          ksqlSchema.fields().get(i).schema().type(),
          equalTo(connectSchema.fields().get(i).schema().type()));
      assertThat(ksqlSchema.fields().get(i).schema().isOptional(), is(true));
    }
    // Make sure that regular int32/int64 fields do not get converted to date/time/timestamp
    assertThat(ksqlSchema.field(nameTranslator.apply("longField")).schema(),
        is(Schema.OPTIONAL_INT64_SCHEMA));
    assertThat(ksqlSchema.field(nameTranslator.apply("intField")).schema(),
        is(Schema.OPTIONAL_INT32_SCHEMA));
  }