I'm new to Flink, I'm having a problem adding data...
# troubleshootingh
Hiếu trịnh
11/28/2025, 3:32 AMI'm new to Flink, I'm having a problem adding data to jdbc oracle. I read data from kafka confluent with upsert mode - kafka. insert into oracle from the result of joining streaming 2 tables. I can't append only to the oracle table, but only upsert. This is my error: java.lang.IllegalStateException: please declare primary key for sink table when query contains update/delete record.
Here is my code:
CREATE TABLE accounts (
KEY STRING,
NAME STRING,
....
y-MM-dd HHmmss.SSS'),
PRIMARY KEY (RECID) NOT ENFORCED
) WITH (
'connector' = 'upsert-kafka',
'topic' = 'account',
'properties.bootstrap.servers' = '${confluent.cluster}',
'key.format' = 'raw',
'value.format' = 'avro-confluent',
'value.avro-confluent.url' = '${confluent.schema.registry}'
);
CREATE TABLE transaction(
ID STRING,
BALANCE INT,
....
y-MM-dd HHmmss.SSS'),
PRIMARY KEY (RECID) NOT ENFORCED
) WITH (
'connector' = 'upsert-kafka',
'topic' = 'transaction',
'properties.bootstrap.servers' = '${confluent.cluster}',
'key.format' = 'raw',
'value.format' = 'avro-confluent',
'value.avro-confluent.url' = '${confluent.schema.registry}'
);
CREATE TABLE oracle_sink (
...
) WITH (
'connector' = 'jdbc',
'url' = '${oracle.host}',
'table-name' = 'RESULT_TBL',
'username' = '${oracle.ogg.user}',
'password' = '${oracle.ogg.password}',
'driver' = 'oracle.jdbc.driver.OracleDriver'
);
INSERT INTO oracle_sink
SELECT
…
from account acc
inner join transaction tran
....