public static <OutputT> Coder<OutputT> inferCoder(
      SingleStoreIO.RowMapper<OutputT> rowMapper,
      CoderRegistry registry,
      SchemaRegistry schemaRegistry,
      Logger log) {
    if (rowMapper instanceof SingleStoreIO.RowMapperWithCoder) {
      try {
        return ((SingleStoreIO.RowMapperWithCoder<OutputT>) rowMapper).getCoder();
      } catch (Exception e) {
        log.warn("Unable to infer a coder from RowMapper. Attempting to infer a coder from type.");
      }
    }

    TypeDescriptor<OutputT> outputType =
        TypeDescriptors.extractFromTypeParameters(
            rowMapper,
            SingleStoreIO.RowMapper.class,
            new TypeDescriptors.TypeVariableExtractor<
                SingleStoreIO.RowMapper<OutputT>, OutputT>() {});
    try {
      return schemaRegistry.getSchemaCoder(outputType);
    } catch (NoSuchSchemaException e) {
      log.warn(
          "Unable to infer a schema for type {}. Attempting to infer a coder without a schema.",
          outputType);
    }
    try {
      return registry.getCoder(outputType);
    } catch (CannotProvideCoderException e) {
      throw new IllegalArgumentException(
          String.format("Unable to infer a coder for type %s", outputType));
    }
  }
@Test
  public void testInferCoderFromSchemaRegistry() {
    SchemaRegistry sr = SchemaRegistry.createDefault();
    CoderRegistry cr = CoderRegistry.createDefault();
    Coder<TestRow> c = SerializableCoder.of(TestRow.class);
    cr.registerCoderForClass(TestRow.class, c);

    assertEquals(c, SingleStoreUtil.inferCoder(new TestRowMapper(), cr, sr, LOG));
  }