@Override
  public Object toConnectRow(final Object ksqlData) {
    /*
     * Reconstruct ksqlData struct with given schema and try to put original data in it.
     * Schema may have more fields than ksqlData, don't put those field by default. If needed by
     * some format like Avro, create new subclass to handle
     */
    if (ksqlData instanceof Struct) {
      final Schema schema = getSchema();
      validate(((Struct) ksqlData).schema(), schema);

      final Struct struct = new Struct(schema);
      final Struct source = (Struct) ksqlData;

      for (final Field sourceField : source.schema().fields()) {
        final Object value = source.get(sourceField);
        struct.put(sourceField.name(), value);
      }

      return struct;
    }

    return ksqlData;
  }
@Test
  public void shouldThrowIfConvertInvalidValue() {
    // Given:
    final Schema schema = SchemaBuilder.struct()
        .field("f1", SchemaBuilder.STRING_SCHEMA)
        .field("f2", SchemaBuilder.OPTIONAL_INT32_SCHEMA)
        .field("f3", SchemaBuilder.OPTIONAL_INT64_SCHEMA)
        .build();
    final Struct struct = new Struct(ORIGINAL_SCHEMA)
        .put("f1", null)
        .put("f2", 12);

    // When:
    final Exception e = assertThrows(
        DataException.class,
        () -> new ConnectSRSchemaDataTranslator(schema).toConnectRow(struct)
    );

    // Then:
    assertThat(e.getMessage(), is("Invalid value: null used for required field: \"f1\", "
        + "schema type: STRING"));
  }