flinkcdc2.1 oracle 请问这个错误怎么解决

[Tongyaoyu]

我的数据库是oracle11g centos7虚拟机
测试代码是

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SourceFunction<String> sourceFunction = OracleSource.<String>builder()
                .hostname("192.168.10.135")
                .port(1521)
                .database("ORCL") 
                .schemaList("TEST")
                .tableList("TEST.PATIENT") 
                .username("test")
                .password("test")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();

        DataStreamSource<String> stringDataStreamSource = env.addSource(sourceFunction);
        stringDataStreamSource.print();

        env.execute("JOB");

报错是

{"before":null,"after":{"ID":"1","NAME":"张三","ADDR":"成都"},"source":{"version":"1.5.4.Final","connector":"oracle","name":"oracle_logminer","ts_ms":1638950893065,"snapshot":"last","db":"ORCL","sequence":null,"schema":"TEST","table":"PATIENT","txId":null,"scn":"1028510","commit_scn":null,"lcr_position":null},"op":"r","ts_ms":1638950893067,"transaction":null}
Exception in thread "main" org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
       ...省略
	at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
	at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
	at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:138)
	at org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:82)
	...省略
	at akka.dispatch.Mailbox.run(Mailbox.scala:225)
	at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
	... 4 more
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:42)
	at io.debezium.connector.oracle.logminer.LogMinerStreamingChangeEventSource.execute(LogMinerStreamingChangeEventSource.java:208)
	...省略
	at java.lang.Thread.run(Thread.java:748)
Caused by: io.debezium.DebeziumException: Supplemental logging not configured for table ORCL.TEST.patient.  Use command: ALTER TABLE TEST.patient ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS
	at io.debezium.connector.oracle.logminer.LogMinerHelper.checkSupplementalLogging(LogMinerHelper.java:407)
	at io.debezium.connector.oracle.logminer.LogMinerStreamingChangeEventSource.execute(LogMinerStreamingChangeEventSource.java:132)
	... 7 more

Process finished with exit code 1

他能读出来我插入的第一条数据，但是报错就断开连接
但是ALTER TABLE TEST.patient ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS我已经执行过了，还是报这个错
折腾3天了，有没有大佬帮帮看一下怎么解决😥

[Tongyaoyu]

自己debug找到了
我的表名是大写PATIENT，在源码中

可以看到tableId 自动把大写表名变成小写
进入isTableSupplementalLogDataAll方法

再进入tableSupplementalLoggingCheckQuery方法

static String tableSupplementalLoggingCheckQuery(TableId tableId) {
        return String.format("SELECT 'KEY', LOG_GROUP_TYPE FROM %s WHERE OWNER = '%s' AND TABLE_NAME = '%s'", "ALL_LOG_GROUPS", tableId.schema(), tableId.table());
    }

他会执行这句sql，但是此时的TABLE_NAME 是小写，于是我在navicat中执行了sql

select * from ALL_LOG_GROUPS

可以看到PATIENT表名为大写，源码中的sql执行出来是查不到的所以出错了
当我把表名改成小写，就可以了
但要是万一表名就是大写该怎么办，有点困惑，有大佬知道的话指点一下

[Tongyaoyu]

而且当我增删改某一条数据时，对应的日志隔了30多s才会打印，我也不知道为啥这么慢

[gtk96]

For Oracle version 11, debezium will set tableidcasesensitive to true by default, resulting in the table name being updated to lowercase. Therefore, the table completion log setting cannot be queried in Oracle, resulting in the false alarm of "supplementary logging not configured for table error".

If it is the datastream API, add the configuration item of debezium 'database.tablename.case.insensitive' = 'false'. If the SQL method is used, add the configuration item 'debezium.database.tablename.case.insensitive' = 'false' in the option of the table

对于 oracle11 版本，debezium 会默认把 tableIdCaseInsensitive 设置为true, 导致表名被更新为小写，因此在oracle中查询不到 这个表补全日志设置，导致误报这个Supplemental logging not configured for table 错误”。 添加 debezium 的配置项 'database.tablename.case.insensitive' = 'false'， 如果使用 SQL 的方式，则在表的 option 中添加配置项 'debezium.database.tablename.case.insensitive' = 'false'

[Tongyaoyu]

请问一下大佬oracleCDC的数据想sink到Kafka中，出了如下的错误，该如何解决，我试了固定的txt数据源Kafka就可以消费，oracleCDC就出这个错，调试我也没看出来😑

[Tongyaoyu]

解决了，版本冲突问题，要引入

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>2.7.0</version>
        </dependency>

指定kafka-client的版本，否则和2.4冲突

[17554262006]

大佬 您这边在同步oracle数据的时候 ,数据库字段类型是number,这种情况下同步出来的数据格式还是数字类型的嘛 会不会变成{"scale":0,"value":"A+g="} 这种格式?

[lizhe6132]

大佬 您这边在同步oracle数据的时候 ,数据库字段类型是number,这种情况下同步出来的数据格式还是数字类型的嘛 会不会变成{"scale":0,"value":"A+g="} 这种格式?

properties.setProperty("decimal.handling.mode","STRING");

[Karan2007]

properties.setProperty("decimal.handling.mode", "double"); //所有的number 转成double了。
或者你自己处理deserialize SourceRecord时：
Map<String, Object> dataMap = new HashMap<>();
Schema schema = struct.schema();
List fieldList = schema.fields();
for (Field field : fieldList) {
//handle oracle number type
if (struct.get(field) instanceof Struct fs){
byte[] bytes = fs.getBytes("value");
String v = Byte.toString(bytes[0]);
dataMap.put(field.name(), v);
Integer scale = fs.getInt32("scale");
BigDecimal val = BigDecimal.valueOf(Long.parseLong(v),scale);
dataMap.put(field.name(), val);
} else {
dataMap.put(field.name(), struct.get(field));
}
}