Соединитель источника Kafka-Connect-JDBC: невозможно прочитать строки из таблицы со столбцом NUMERIC - PullRequest
Новая
       15

Соединитель источника Kafka-Connect-JDBC: невозможно прочитать строки из таблицы со столбцом NUMERIC

0 голосов
/ 04 сентября 2018

enter image description here

Я пытаюсь настроить исходный соединитель для базы данных Adventureworks в postgres, таблица описана на рисунке выше. Конфигурация источника приведена ниже. Когда соединитель работает, он не может обработать столбец с числовым значением и пропустить все строки с ошибочным значением

Предупредить о пропущенной записи 

[2018-09-04 14:48:03,324] WARN Ignoring record due to SQL error: (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier)
org.postgresql.util.PSQLException: Bad value for type byte : 183.9382
    at org.postgresql.jdbc2.AbstractJdbc2ResultSet.getByte(AbstractJdbc2ResultSet.java:2093)
    at io.confluent.connect.jdbc.dialect.GenericDatabaseDialect.lambda$columnConverterFor$18(GenericDatabaseDialect.java:1166)
    at io.confluent.connect.jdbc.source.SchemaMapping$FieldSetter.setField(SchemaMapping.java:160)
    at io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.extractRecord(TimestampIncrementingTableQuerier.java:176)
    at io.confluent.connect.jdbc.source.JdbcSourceTask.poll(JdbcSourceTask.java:297)
    at org.apache.kafka.connect.runtime.WorkerSourceTask.poll(WorkerSourceTask.java:244)
    at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:220)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

Конфигурация разъема 

{
    "name": "jdbc_source_sales.salesorderdetail",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "tasks.max": "1",
        "topics": "sales.salesorderdetail",
        "connection.url": "jdbc:postgresql://172.18.0.1/adventureworks?user=postgres&password=postgres",
        "mode": "timestamp+incrementing",
        "timestamp.column.name": "modifieddate",
        "incrementing.column.name": "salesorderid",
        "topic.prefix": "jdbc_source_sales_",
        "table.whitelist": "sales.salesorderdetail",
        "transforms": "CastUnitPrice, InsertKey, ExtractId, CastLong, AddNamespace",
        "transforms.InsertKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
        "transforms.InsertKey.fields": "salesorderid",
        "transforms.ExtractId.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
        "transforms.ExtractId.field": "salesorderid",
        "transforms.CastLong.type": "org.apache.kafka.connect.transforms.Cast$Key",
        "transforms.CastLong.spec": "int64", 
        "transforms.AddNamespace.type": "de.smava.kafka.connect.transforms.Namespacefy",
        "transforms.AddNamespace.record.namespace": "com.company.data.vault20",
        "transforms.CastUnitPrice.type": "org.apache.kafka.connect.transforms.Cast$Value",
        "transforms.CastUnitPrice.spec": "unitprice:float64",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "key.converter.schemas.enable": "false",
        "value.converter": "io.confluent.connect.avro.AvroConverter",
        "value.converter.schemas.enable": "true",
        "value.converter.schema.registry.url": "http://schema-registry:8081",
        "numeric.mapping": "best_fit"

    }
}

[ EDIT ] Проблема была в масштабе и точности числовых полей, но я все еще не мог понять, почему по умолчанию Scale выдает ошибку

...