I have a SQL dynamoDB sink job working locally. But there is one snag I cant seem to get past. Delete requests to Dynamo dont use a proper partition key causing deletes to error because it uses all columns of the table rather than the partition key. Something in the connector javacode + SQL i think. Im using a partitioned flink table as well. Create and PUT events use the proper partition key. IDK if this rings any bells for anyone, or could be specific to the dynamo connector the sink table

CREATE TABLE property_features_dynamo_sink (
	`key` VARCHAR NOT NULL,
	`property_id` BIGINT NOT NULL,
	`value` VARCHAR NOT NULL,
	`value_override` VARCHAR NOT NULL,
    PRIMARY KEY (`key`, `property_id`) NOT ENFORCED
) PARTITIONED BY ( `key`, property_id )
WITH (
    'connector' = 'dynamodb'
)

the insert into sink is

INSERT INTO property_features_dynamo_sink
select
    `key`,
    `property_id`,
    `value`,
    `value_override`
from sr.topics.`cdc.property_features` property_features
where
    property_features.`key` = 'cb_amplify_dashboard'
    and (
        property_features.`value` = '1'
        or property_features.value_override = '1'
    );

@Brandon Dougherty I have also same issue but, I just want your help in one topic, By chance did you implemented dynamic rule i.e. You are using SQL rule(SQL query) to apply on log, Did you implemented static/fix number of query or Dynamic number of query?

@Keyur Makwana they query is on a kafka changelog topic (CDC)

I understand but is that query fix or you can give input while job is running?

yes you can give input, its a kafka topic at the source

If possible can you give pseudo code or any help?

This is the code its executed using SQL api

INSERT INTO property_features_dynamo_sink
select
    `key`,
    `property_id`,
    `value`,
    `value_override`
from sr.topics.`cdc.property_features` property_features
where
    property_features.`key` = 'cb_amplify_dashboard'
    and (
        property_features.`value` = '1'
        or property_features.value_override = '1'
    );

This is the only SQL that you are executing correct? or this SQL coming from Kafka?

@Keyur Makwana did you get this resolved?

No brother not yet

@Brandon Dougherty @Keyur Makwana - I believe I’m hitting the same problem with using a SQL DynamoDB Sink. When I start a stream processing job, the first few records get inserted into the DyanmoDB sink successfully, but then I hit the error:

Caused by: org.apache.flink.connector.dynamodb.shaded.software.amazon.awssdk.services.dynamodb.model.DynamoDbException: The provided key element does not match the schema (Service: DynamoDb, Status Code: 400

@Brandon Dougherty how’d you get the full API request body to confirm that the issue was with a DELETE operation? Was that via CloudTrail?