我在处理 PostgreSQL 数据库的快照时遇到 Debezium 版本 2.3.1 的问题。该错误发生在快照阶段，似乎与解析日期/时间值有关。具体错误信息为：
Caused by: org.apache.kafka.connect.errors.ConnectException:
java.time.format.DateTimeParseException: Text 'f' could not be parsed at index 0
at io.debezium.connector.postgresql.connection.DateTimeFormat$ISODateTimeFormat.format(DateTimeFormat.java:166)
at io.debezium.connector.postgresql.connection.DateTimeFormat$ISODateTimeFormat.timestampToInstant(DateTimeFormat.java:172)
at io.debezium.connector.postgresql.connection.AbstractColumnValue.asInstant(AbstractColumnValue.java:81)
at io.debezium.connector.postgresql.connection.ReplicationMessageColumnValueResolver.resolveValue(ReplicationMessageColumnValueResolver.java:110)
at io.debezium.connector.postgresql.connection.pgoutput.PgOutputReplicationMessage.getValue(PgOutputReplicationMessage.java:92)
at io.debezium.connector.postgresql.connection.pgoutput.PgOutputMessageDecoder$1.getValue(PgOutputMessageDecoder.java:748)
at io.debezium.connector.postgresql.PostgresChangeRecordEmitter.columnValues(PostgresChangeRecordEmitter.java:179)
at io.debezium.connector.postgresql.PostgresChangeRecordEmitter.getNewColumnValues(PostgresChangeRecordEmitter.java:125)
at io.debezium.relational.RelationalChangeRecordEmitter.emitCreateRecord(RelationalChangeRecordEmitter.java:69)
at io.debezium.relational.RelationalChangeRecordEmitter.emitChangeRecords(RelationalChangeRecordEmitter.java:47)
at io.debezium.connector.postgresql.PostgresChangeRecordEmitter.emitChangeRecords(PostgresChangeRecordEmitter.java:94)
at io.debezium.pipeline.EventDispatcher.dispatchDataChangeEvent(EventDispatcher.java:296)
... 17 more
Caused by: java.time.format.DateTimeParseException: Text 'f' could not be parsed at index 0
at java.base/java.time.format.DateTimeFormatter.parseResolved0(DateTimeFormatter.java:2046)
at java.base/java.time.format.DateTimeFormatter.parse(DateTimeFormatter.java:1874)
at io.debezium.connector.postgresql.connection.DateTimeFormat$ISODateTimeFormat.lambda$timestampToInstant$3(DateTimeFormat.java:172)
at io.debezium.connector.postgresql.connection.DateTimeFormat$ISODateTimeFormat.format(DateTimeFormat.java:162)
使用为增量快照配置的表时会出现此问题。这是我正在使用的连接器配置的示例：
*
*{
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"connector.displayName": "PostgreSQL",
"database.user": "docker",
"database.dbname": "exampledb",
"transforms": "unwrap",
"database.server.name": "localhost",
"heartbeat.interval.ms": "60000",
"database.port": "5432",
"plugin.name": "pgoutput",
"slot.max.retries": "10",
"schema.include.list": "public",
"slot.retry.delay.ms": "15000",
"heartbeat.action.query": "INSERT INTO public.debezium_heartbeat VALUES ('debezium', now())",
"decimal.handling.mode": "string",
"database.hostname": "postgres",
"database.password": "docker",
"transforms.unwrap.drop.tombstones": "false",
"signal.data.collection": "public.debezium_signal",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"table.include.list": "public.table, public.debezium_signal",
"max.batche.size": "65000",
"max.queue.size": "275000",
"incremental.snapshot.chunk.size": "6500",
"connector.id": "postgres",
"topic.prefix": "db-prod"
}**
该错误似乎与快照数据中的日期/时间字段相关，如带有文本“f”的 DateTimeParseException 所示，表明日期/时间格式或解析存在潜在问题。
我的表的 DDL：
CREATE TABLE IF NOT EXISTS public.table
(
id bigint NOT NULL DEFAULT nextval('table_id_seq'::regclass),
federal_tax_id character varying(255) COLLATE pg_catalog."default",
external_id character varying(255) COLLATE pg_catalog."default",
first_name character varying(255) COLLATE pg_catalog."default",
last_name character varying(255) COLLATE pg_catalog."default",
email character varying(255) COLLATE pg_catalog."default",
phone character varying(255) COLLATE pg_catalog."default",
cellphone character varying(255) COLLATE pg_catalog."default",
address_street character varying(255) COLLATE pg_catalog."default",
address_street_number character varying(255) COLLATE pg_catalog."default",
address_complement character varying(255) COLLATE pg_catalog."default",
address_city_district character varying(255) COLLATE pg_catalog."default",
address_post_code character varying(255) COLLATE pg_catalog."default",
address_city character varying(255) COLLATE pg_catalog."default",
address_city_code character varying(255) COLLATE pg_catalog."default",
address_state_code character varying(255) COLLATE pg_catalog."default",
address_country character varying(255) COLLATE pg_catalog."default",
address_latitude numeric,
address_longitude numeric,
address_geo geometry(Point,4326) GENERATED ALWAYS AS (st_setsrid(st_makepoint((address_longitude)::double precision, (address_latitude)::double precision), 4326)) STORED,
created_at timestamp without time zone NOT NULL,
updated_at timestamp without time zone NOT NULL,
is_company boolean NOT NULL DEFAULT false,
state_tax_id character varying(255) COLLATE pg_catalog."default",
official_name character varying(255) COLLATE pg_catalog."default",
CONSTRAINT table_pkey PRIMARY KEY (id)
)
TABLESPACE pg_default;
应用程序版本：
Debezium：2.3.1 PostgreSQL：13.15
我尝试使用相同版本的 Debezium 重现错误，当我拍摄快照然后使用以下查询执行更新时发生错误：
UPDATE public.table
SET updated_at = current_timestamp
WHERE id = 28948709;
example snapshot incremental:
INSERT INTO debezium_signal (id, type, data)
VALUES (
'sdadsada-sdasdas-9b45-12608f88b2a6',
'execute-snapshot',
'{"data-collections": ["public.table"]}'
);
我可以将 cdc 与流媒体集成，但我需要历史记录。
更新： 我在 Debezium 中启用了日志并发现了这个：
024-08-10 22:09:43,120 ERROR Postgres|company-db-prod|streaming Failed to properly convert data value for 'public.table.is_company' of type bool [io.debezium.relational.TableSchemaBuilder]
2024-08-10 22:09:43,121 ERROR Postgres|company-db-prod|streaming Failed to properly convert data value for 'public.table.is_company' of type bool [io.debezium.relational.TableSchemaBuilder]
2024-08-10 22:09:43,121 ERROR Postgres|company-db-prod|streaming Failed to properly convert data value for 'public.table.is_company' of type bool [io.debezium.relational.TableSchemaBuilder]
2024-08-10 22:09:43,122 ERROR Postgres|company-db-prod|streaming Failed to properly convert data value for 'public.table.is_company' of type bool [io.debezium.relational.TableSchemaBuilder]
2024-08-10 22:09:43,123 ERROR Postgres|company-db-prod|streaming Failed to properly convert data value for 'public.table.is_company' of type bool [io.debezium.relational.TableSchemaBuilder]
2024-08-10 22:09:43,123 ERROR Postgres|company-db-prod|streaming Failed to properly convert data value for 'public.table.is_company' of type bool [io.debezium.relational.TableSchemaBuilder]
2024-08-10 22:09:43,123 ERROR Postgres|company-db-prod|streaming Failed to properly convert data value for 'public.table.is_company' of type bool [io.debezium.relational.TableSchemaBuilder]
2024-08-10 22:09:43,124 ERROR Postgres|company-db-prod|streaming Failed to properly convert data value for 'public.table.is_company' of type bool [io.debezium.relational.TableSchemaBuilder]
2024-08-10 22:09:43,124 ERROR Postgres|company-db-prod|streaming Failed to properly convert data value for 'public.table.is_company' of type bool [io.debezium.relational.TableSchemaBuilder]
2024-08-10 22:09:43,125 ERROR Postgres|company-db-prod|streaming Failed to properly convert data value for 'public.table.is_company' of type bool [io.debezium.relational.TableSchemaBuilder]
2024-08-10 22:09:43,126 ERROR Postgres|company-db-prod|streaming Failed to properly convert data value for 'public.table.is_company' of type bool [io.debezium.relational.TableSchemaBuilder]
2024-08-10 22:09:43,126 ERROR Postgres|company-db-prod|streaming Failed to properly convert data value for 'public.table.is_company' of type bool [io.debezium.relational.TableSchemaBuilder]
2024-08-10 22:09:43,127 ERROR Postgres|company-db-prod|streaming Failed to properly convert data value for 'public.table.is_company' of type bool [io.debezium.relational.TableSchemaBuilder]
java.lang.IllegalArgumentException: Column 'address_geo' not found in result set 'id, federal_tax_id, external_id, first_name, last_name, email, phone, cellphone, address_street, address_street_number, address_complement, address_city_district, address_post_code, address_city, address_city_code, address_state_code, address_country, address_latitude, address_longitude, address_geo, created_at, updated_at, is_company, state_tax_id, official_name' for table 'public.table', columns: {
address_geo 不在模式注册表中。这会引起问题吗？
我发现了问题的原因，这只发生在pgouput插件中，在decodebufs中我没有找到相同的场景。我设法跟踪它并找到问题，建议我在 Jira 中打开一个错误： https://issues.redhat.com/browse/DBZ-8150
要查看案例讨论，请查看此链接：