[
https://issues.apache.org/jira/browse/FLINK-38965?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18053872#comment-18053872
]
Tao Wang commented on FLINK-38965:
----------------------------------
*Solution：* we should add filter in PostgresConnection::doReadTableColumn
!image-2026-01-23-18-24-26-642.png|width=806,height=454!
> Postgres cdc source encounted "java.lang.IllegalStateException: Duplicate
> key Optional"
> ----------------------------------------------------------------------------------------
>
> Key: FLINK-38965
> URL: https://issues.apache.org/jira/browse/FLINK-38965
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Affects Versions: cdc-3.5.0
> Reporter: Tao Wang
> Priority: Major
> Attachments: image-2026-01-23-18-17-50-692.png,
> image-2026-01-23-18-21-17-272.png, image-2026-01-23-18-24-26-642.png
>
>
> connector: Postgress
> description:
> 1、we got two tables with similar name.
> ndi_pg_user_sink_1
> ndi_pg_userbsink_1
> 2、they have same schema.
> 3、pg as cdc source, we encounted the exception
> {code:java}
> java.lang.RuntimeException: SplitFetcher thread 1 received unexpected
> exception while polling the records at
> org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:150)
> ~[flink-connector-base-ne-flink-1.14.0-1.0.13.jar:ne-flink-1.14.0-1.0.13]
> at
> org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:105)
> ~[flink-connector-base-ne-flink-1.14.0-1.0.13.jar:ne-flink-1.14.0-1.0.13]
> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> ~[?:1.8.0_421] at
> java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266)
> ~[?:1.8.0_421] at java.util.concurrent.FutureTask.run(FutureTask.java)
> ~[?:1.8.0_421] at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> ~[?:1.8.0_421] at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> ~[?:1.8.0_421] at java.lang.Thread.run(Thread.java:750)
> ~[?:1.8.0_421]Caused by: org.apache.kafka.connect.errors.ConnectException: An
> exception occurred in the change event producer. This connector will be
> stopped. at
> io.debezium.pipeline.ErrorHandler.setProducerThrowable(ErrorHandler.java:50)
> ~[debezium-core-1.9.7.Final.jar:1.9.7.Final] at
> io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:214)
> ~[flink-connector-postgres-cdc-.jar:1.9.7.Final] at
> com.ververica.cdc.connectors.postgres.source.fetch.PostgresStreamFetchTask$StreamSplitReadTask.execute(PostgresStreamFetchTask.java:211)
> ~[flink-connector-postgres-cdc-.jar:] at
> com.ververica.cdc.connectors.postgres.source.fetch.PostgresStreamFetchTask.execute(PostgresStreamFetchTask.java:95)
> ~[flink-connector-postgres-cdc-.jar:] at
> com.ververica.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher.lambda$submitTask$0(IncrementalSourceStreamFetcher.java:89)
> ~[flink-cdc-base-.jar:] ... 6 moreCaused by:
> java.lang.IllegalStateException: Duplicate key Optional.empty at
> java.util.stream.Collectors.lambda$throwingMerger$0(Collectors.java:133)
> ~[?:1.8.0_421] at java.util.HashMap.merge(HashMap.java:1255)
> ~[?:1.8.0_421] at
> java.util.stream.Collectors.lambda$toMap$58(Collectors.java:1320)
> ~[?:1.8.0_421] at
> java.util.stream.ReduceOps$3ReducingSink.accept(ReduceOps.java:169)
> ~[?:1.8.0_421] at
> java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
> ~[?:1.8.0_421] at
> java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
> ~[?:1.8.0_421] at
> java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
> ~[?:1.8.0_421] at
> java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
> ~[?:1.8.0_421] at
> java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
> ~[?:1.8.0_421] at
> java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
> ~[?:1.8.0_421] at
> java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
> ~[?:1.8.0_421] at
> java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499)
> ~[?:1.8.0_421] at
> io.debezium.connector.postgresql.connection.pgoutput.PgOutputMessageDecoder.handleRelationMessage(PgOutputMessageDecoder.java:319)
> ~[flink-connector-postgres-cdc-.jar:] at
> io.debezium.connector.postgresql.connection.pgoutput.PgOutputMessageDecoder.processNotEmptyMessage(PgOutputMessageDecoder.java:178)
> ~[flink-connector-postgres-cdc-.jar:] at
> io.debezium.connector.postgresql.connection.AbstractMessageDecoder.processMessage(AbstractMessageDecoder.java:33)
> ~[debezium-connector-postgres-1.9.7.Final.jar:] at
> io.debezium.connector.postgresql.connection.PostgresReplicationConnection$1.deserializeMessages(PostgresReplicationConnection.java:604)
> ~[flink-connector-postgres-cdc-.jar:] at
> io.debezium.connector.postgresql.connection.PostgresReplicationConnection$1.readPending(PostgresReplicationConnection.java:594)
> ~[flink-connector-postgres-cdc-.jar:] at
> io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.processMessages(PostgresStreamingChangeEventSource.java:257)
> ~[flink-connector-postgres-cdc-.jar:1.9.7.Final] at
> io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:212)
> ~[flink-connector-postgres-cdc-.jar:1.9.7.Final] at
> com.ververica.cdc.connectors.postgres.source.fetch.PostgresStreamFetchTask$StreamSplitReadTask.execute(PostgresStreamFetchTask.java:211)
> ~[flink-connector-postgres-cdc-.jar:] at
> com.ververica.cdc.connectors.postgres.source.fetch.PostgresStreamFetchTask.execute(PostgresStreamFetchTask.java:95)
> ~[flink-connector-postgres-cdc-.jar:] at
> com.ververica.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher.lambda$submitTask$0(IncrementalSourceStreamFetcher.java:89)
> ~[flink-cdc-base-.jar:] ... 6 more {code}
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
