Confluent Kafka Sink Connector не загружает данные в таблицу Postgres - PullRequest
Новая
       9

Confluent Kafka Sink Connector не загружает данные в таблицу Postgres

0 голосов
/ 23 мая 2018

Я пытаюсь загрузить данные в таблицы (таблицы) Postgres через соединитель Kafka Sink, но получаю следующую ошибку:

Причина: org.apache.kafka.connect.errors.ConnectException: Невозможно ALTER добавить отсутствующее поле SinkRecordField {schema = Schema {STRING}, name = 'A_ABBREV', isPrimaryKey = false}, так как оно не является обязательным и не имеет значения по умолчанию

Таблицав Postgres DB уже есть поле A_ABBREV, но теперь я уверен, почему я получаю пропущенную ошибку поля.

Кто-нибудь сталкивался с подобной проблемой?

Ниже приведена конфигурация моего коннектора Sink: 

connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
table.name.format=AGENCY
connection.password=passcode
topics=AGENCIES
tasks.max=1
batch.size=10000
fields.whitelist=A_ID, A_NAME, A_ABBREV
connection.user=pmmdevuser
name=partner5-jdbcSinkConnector
connection.url=jdbc:postgresql://aws-db.sdfdgfdrwwisc.us-east- 1.rds.amazonaws.com:3306/pmmdevdb?currentSchema=ams
insert.mode=upsert
pk.mode=record_value
pk.fields=A_ID
auto.create=false

Я использую сценарии Liquibase для создания таблиц, а ниже приведен запрос на создание из базы данных postgres, созданной с помощью сценариев Liquibase: 

"CREATE TABLE gds.agency
(
    a_id integer NOT NULL,
    a_name character varying(100) COLLATE pg_catalog."default" NOT NULL,
    a_abbrev character varying(8) COLLATE pg_catalog."default" NOT NULL,
    source character varying(255) COLLATE pg_catalog."default" NOT NULL DEFAULT 'AMS'::character varying,
    CONSTRAINT pk_agency PRIMARY KEY (a_id),
    CONSTRAINT a_abbrev_uk1 UNIQUE (a_abbrev)
)"

1 Ответ

0 голосов
/ 23 августа 2018

Из моего опыта это означает, что определение поля для приемника не соответствует определению поля для исходной таблицы / базы данных.Убедитесь, что определения полей совпадают.Проверьте индивидуальную запись, которую соединитель приемника пытается записать в целевую базу данных.Вы должны увидеть этот оператор вставки в режиме отладки в трассировке стека.возьмите этот запрос и запустите его вручную, чтобы получить более четкое представление об ошибке из базы данных.

...