Gogo-scc
commented
1 year ago The kafka Connect log：
2022-10-24 12:22:47,644] INFO Connector logminer-source-testapps connected to Oracle Database 11g Enterprise Edition , 11.2.0.4.0
$ORACLE_SID=orcl1, running on orcl1, OS Linux x86 64-bit. (solutions.a2.cdc.oracle.OraCdcLogMinerTask:247)
[2022-10-24 12:22:47,671] INFO No data present in connector's offset storage for orcl1_orcl1:1452920421 (solutions.a2.cdc.oracle.OraCdcLogMinerTask:541)
[2022-10-24 12:22:47,671] INFO oracdc will start from minimum available SCN in V$ARCHIVED_LOG = 0. (solutions.a2.cdc.oracle.OraCdcLogMinerTask:559)
[2022-10-24 12:22:47,680] INFO Initializing oracdc logminer archivelog worker thread (solutions.a2.cdc.oracle.OraCdcLogMinerWorkerThread:138)
[2022-10-24 12:22:47,735] INFO LogMiner will start from SCN 0 (solutions.a2.cdc.oracle.OraCdcV$ArchivedLogImpl:105)
[2022-10-24 12:22:47,736] INFO Mining database orcl1 is in OPEN mode (solutions.a2.cdc.oracle.OraCdcLogMinerWorkerThread:250)
[2022-10-24 12:22:47,736] INFO Same database will be used for dictionary query and mining (solutions.a2.cdc.oracle.OraCdcLogMinerWorkerThread:252)
[2022-10-24 12:22:47,736] INFO RowPrefetch size for accessing V$LOGMNR_CONTENTS set to 32. (solutions.a2.cdc.oracle.OraCdcLogMinerWorkerThread:270)
[2022-10-24 12:22:47,827] INFO Initializing oracdc initial load thread (solutions.a2.cdc.oracle.OraCdcInitialLoadThread:64)
[2022-10-24 12:22:47,828] INFO DB cores available 2, Kafka Cores available 4. (solutions.a2.cdc.oracle.OraCdcInitialLoadThread:74)
[2022-10-24 12:22:47,828] INFO {} parallel loaders for select phase will be used. (solutions.a2.cdc.oracle.OraCdcInitialLoadThread:75)
[2022-10-24 12:22:47,828] INFO BEGIN: OraCdcInitialLoadThread.run() (solutions.a2.cdc.oracle.OraCdcInitialLoadThread:86)
[2022-10-24 12:22:47,831] INFO WorkerSourceTask{id=logminer-source-testapps-0} Source task finished initialization and start (org.apache.kafka.connect.runtime.WorkerSourceTask:215)
[2022-10-24 12:22:47,831] INFO BEGIN: OraCdcLogMinerWorkerThread.run() (solutions.a2.cdc.oracle.OraCdcLogMinerWorkerThread:338)
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access using Lookup on net.openhft.chronicle.core.Jvm (file:/kafka/connect/lib/chronicle-core-2.21.95.jar) to class java.lang.reflect.AccessibleObject
WARNING: Please consider reporting this to the maintainers of net.openhft.chronicle.core.Jvm
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release
[2022-10-24 12:22:47,881] INFO Chronicle core loaded from file:/kafka/connect/lib/chronicle-core-2.21.95.jar (net.openhft.chronicle.core.Jvm:169)
Oct 24, 2022 12:22:47 PM net.openhft.chronicle.core.cleaner.impl.reflect.ReflectionBasedByteBufferCleanerService
averemee-si
HI I encountered the following problem when testing： 1,Useing oracle 11gr2(11.2.0.4) 2,Archival enable alter database add supplemental log data;
alter table ABC.TEST32 add supplemental log data (ALL) columns;
3,Start the kafka connect bin/connect-standalone etc/kafka/connect-standalone.properties etc/kafka/logminer-source-testapps.properties
4, Create and insert table
create table test32( myid int not null primary key, myname varchar(500) )
insert into test32 values (4,'22'); insert into test32 values (5,'23'); insert into test32 values (6,'24'); insert into test32 values (7,'25'); insert into test32 values (8,'26');
5,The kafka topic get the data when I start kafka connect is ok. But can not get data from topic when continuing to insert data into the table.I can get the topic data when I must be restart the connect
[root@kafka2 confluent-5.4.8]# bin/kafka-console-consumer --bootstrap-server 172.18.6.112:9092 --topic ABC-TEST32 --property print.key=true --property print.value=true --from-beginning [2022-10-24 12:22:39,669] WARN [Consumer clientId=consumer-console-consumer-79494-1, groupId=console-consumer-79494] Error while fetching metadata with correlation id 2 : {ABC-TEST32=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient) {"schema":{"type":"struct","fields":[{"type":"bytes","optional":false,"name":"org.apache.kafka.connect.data.Decimal","version":1,"parameters":{"scale":"0"},"field":"MYID"}],"optional":false,"name":"ABC.TEST32.Key","version":1},"payload":{"MYID":"BA=="}} {"schema":{"type":"struct","fields":[{"type":"string","optional":true,"field":"MYNAME"}],"optional":true,"name":"ABC.TEST32.Value","version":1},"payload":{"MYNAME":"22"}} {"schema":{"type":"struct","fields":[{"type":"bytes","optional":false,"name":"org.apache.kafka.connect.data.Decimal","version":1,"parameters":{"scale":"0"},"field":"MYID"}],"optional":false,"name":"ABC.TEST32.Key","version":1},"payload":{"MYID":"BQ=="}} {"schema":{"type":"struct","fields":[{"type":"string","optional":true,"field":"MYNAME"}],"optional":true,"name":"ABC.TEST32.Value","version":1},"payload":{"MYNAME":"23"}}