public static Object withCompatibleSchema(final Schema schema, final Object object) {
    if (object == null) {
      return null;
    }
    switch (schema.type()) {
      case ARRAY:
        final List<Object> ksqlArray = new ArrayList<>(((List) object).size());
        ((List) object).forEach(
            e -> ksqlArray.add(withCompatibleSchema(schema.valueSchema(), e)));
        return ksqlArray;

      case MAP:
        final Map<Object, Object> ksqlMap = new HashMap<>();
        ((Map<Object, Object>) object).forEach(
            (key, value) -> ksqlMap.put(
                withCompatibleSchema(schema.keySchema(), key),
                withCompatibleSchema(schema.valueSchema(), value)
            ));
        return ksqlMap;

      case STRUCT:
        return withCompatibleRowSchema((Struct) object, schema);
      case BYTES:
        if (DecimalUtil.isDecimal(schema)) {
          return DecimalUtil.ensureFit((BigDecimal) object, schema);
        } else {
          return object;
        }
      default:
        return object;
    }
  }
@Test
  public void shouldMakeStructSchemaCompatible() {
    // Given:
    final Schema oldSchema = new ConnectSchema(
        Type.STRUCT, false, null, "oldSchema", 1, "");
    final Struct struct = new Struct(oldSchema);
    // When:
    final Schema newSchema = new ConnectSchema(
        Type.STRUCT, false, null, "newSchema", 1, "");;
    final Struct structWithNewSchema = (Struct) ConnectSchemas.withCompatibleSchema(newSchema, struct);

    // Then:
    assertThat(structWithNewSchema.schema(), is(newSchema));
  }