Hi Team When using toChanlogStream with JDBC postgres the fl Apache Flink #troubleshooting

Hi Team, When using toChanlogStream with JDBC post...

Dheeraj Panangat

03/13/2023, 1:26 PM

Hi Team, When using toChanlogStream with JDBC postgres, the flink completes execution after reading the table and does not wait for CDC events. What I understood from the documentation is that toChangelogStream should be able to capture the CDC events. Is my understanding incorrect ? Also do we have a sample for consuming CDC or changelog from Postgres as a stream? Appreciate any help. Thanks

Martijn Visser

03/13/2023, 1:27 PM

Which connector are you using?

Dheeraj Panangat

03/13/2023, 1:27 PM

jdbc

Martijn Visser

03/13/2023, 1:27 PM

That's not a CDC connector 🙂 You can find one at https://github.com/ververica/flink-cdc-connectors

Dheeraj Panangat

03/13/2023, 1:31 PM

Tried this. Followed the steps as is, but keep running into different issues for different approaches. With SQL Approach it gives error as

Copy code

Exception in thread "main" org.apache.flink.table.api.ValidationException: Could not execute CreateTable in path `default_catalog`.`*****`.`****`

And with PostgreSQLSource

Copy code

Caused by: java.lang.NoSuchFieldError: DEFAULT
	at org.apache.kafka.connect.runtime.WorkerConfig.baseConfigDef(WorkerConfig.java:284)

Code Sample :

Copy code

SourceFunction<String> sourceFunction = PostgreSQLSource.<String>builder()
                .hostname("hostname")
                .port(5432)
                .database("postgres")
                .schemaList("schemaname")
                .tableList("schemaname.tablename")
                .username("postgres")
                .password("***pwd***")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();
        DataStreamSource<String> flowStream = env.addSource(sourceFunction);
        flowStream.addSink(new PrintSinkFunction<>()).name("Print to System Output");

Martijn Visser

03/13/2023, 1:37 PM

I don't understand how you can run into a Kafka Connect error with Flink

Martijn Visser

03/13/2023, 1:37 PM

There's https://github.com/immerok/recipes/blob/main/change-data-capture/src/main/java/com/immerok/cookbook/ReadCDC.java for a working example

Dheeraj Panangat

03/13/2023, 1:50 PM

Thanks @Martijn Visser, The only difference I see when compared with the above sample is, my dependency I am excluding jetty as the compilation fails if I dont

Copy code

<dependency>
			<groupId>com.ververica</groupId>
			<artifactId>flink-connector-postgres-cdc</artifactId>
			<version>2.3.0</version>
			<exclusions>
				<exclusion>
					<groupId>org.eclipse.jetty</groupId>
					<artifactId>*</artifactId>
				</exclusion>
			</exclusions>
		</dependency>

The complete stack trace : I copied the above code as is :

Copy code

Caused by: java.lang.NoSuchFieldError: DEFAULT
	at org.apache.kafka.connect.runtime.WorkerConfig.baseConfigDef(WorkerConfig.java:284)
	at io.debezium.embedded.EmbeddedEngine$EmbeddedConfig.<clinit>(EmbeddedEngine.java:1101)
	at io.debezium.embedded.EmbeddedEngine.<init>(EmbeddedEngine.java:600)
	at io.debezium.embedded.EmbeddedEngine.<init>(EmbeddedEngine.java:81)
	at io.debezium.embedded.EmbeddedEngine$BuilderImpl.build(EmbeddedEngine.java:302)
	at io.debezium.embedded.EmbeddedEngine$BuilderImpl.build(EmbeddedEngine.java:218)
	at io.debezium.embedded.ConvertingEngineBuilder.build(ConvertingEngineBuilder.java:151)
	at com.ververica.cdc.debezium.DebeziumSourceFunction.run(DebeziumSourceFunction.java:420)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:110)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:67)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:333)

Process finished with exit code 1

Dheeraj Panangat

03/14/2023, 12:25 PM

Hi @Martijn Visser, Thanks for your response. Was able to find the root cause of the issues. Posting here, if it helps anyone 1. The

kafka-clients

dependency coming from

flink-connector-kafka

and

flink-connector-postgres-cdc

can conflict. Had to exlude the dependency from flink-connector-kafka 2. If using partitions for table the table name has to include suffix _[0-n]+.* 3. Difference in PostgresSqlSource and TableEnvironment : • Table Environment : REPLICA IDENTITY has to be FULL to use the table API • The tableList has to include schemaName.tableName for SourceFunction, but for TableAPI it needs only the tableName 4. Also the slot.name has to be different for each CDC events

Martijn Visser

03/14/2023, 1:57 PM

Thanks. For the next time, please don't repost to the channel

✅ 1

5 Views

Open in Slack

Previous Next