[FLINK-39810][Connectors/kinesis] Support upsert changelog streams in Kinesis SQL connector

[fmorillo7694]

Purpose of the change

Adds upsert changelog stream support to the Kinesis SQL connector. When a PRIMARY KEY is defined on the table, the sink accepts the results of aggregations (GROUP BY), deduplication, and streaming joins over append-only inputs, writing the latest row per key to Kinesis Data Streams.

Design decisions

• A mode on the existing kinesis connector, not a separate upsert-kinesis connector. upsert-kafka is a separate connector because compacted topics give it distinct semantics on both sides: the source reads a compacted topic as a changelog (null value = DELETE tombstone), and the sink writes tombstones that log compaction actually applies. Kinesis has neither compaction nor key-based retention, so there is no materialized latest-per-key view for an upsert source to read — the capability is inherently sink-only. A separate connector identifier would duplicate every source/sink option for the sake of one sink behavior; instead, the presence of a PRIMARY KEY activates upsert mode on the existing factory (the same signal upsert-kafka requires), and everything incompatible with it fails validation explicitly.
• Changelog mode is {INSERT, UPDATE_AFTER} — DELETE is intentionally not supported. Kinesis records have no key/value separation, so a delete event cannot carry the deleted key in a format-agnostic way (an empty-payload tombstone is unusable by consumers and breaks JSON deserialization). Delete-producing queries (CDC sources, Top-N) are rejected at planning time with a descriptive planner error instead. A sink.delete-strategy option can be added later if a sound tombstone contract is found.
• Per-key ordering is enforced, not documented away. Upsert consumers materialize the last record per key, so out-of-order updates cause permanently stale state. In upsert mode the connector forces sink.requests.max-inflight = 1 so request batches cannot overtake each other; an explicit conflicting value fails validation. Residual limitation (documented): the Kinesis PutRecords API does not guarantee ordering within one request, so two updates for the same key in a single batch may be sequenced out of order. A key-deduplicating flush buffer (analogous to upsert-kafka's sink.buffer-flush.*) is a natural follow-up that would close this and restore throughput.
• No silent behavior changes. The primary key drives the partition key; a conflicting explicit sink.partitioner, a PARTITIONED BY clause, or sink.requests.max-inflight != 1 all fail with a ValidationException rather than being silently overridden.

Changes

• KinesisDynamicSink: upsertMode flag; restricted ChangelogMode; UpsertSerializationSchemaWrapper normalizes UPDATE_AFTER to INSERT for insert-only formats (RowKind restored via try/finally; DELETE/UPDATE_BEFORE rejected defensively); constructor precondition on maxInFlightRequests.
• KinesisDynamicTableFactory: detects the primary key, enables upsert mode, derives the partition key from PK fields, forces max in-flight to 1, and validates incompatible options.
• Tests: wrapper unit tests (normalization, RowKind restore incl. on serializer failure, DELETE/UPDATE_BEFORE rejection) and factory tests (PK wiring incl. expected-sink equality and changelog-mode restriction, plus all three validation failures).
• Docs (EN + ZH): upsert section rewritten to match — supported events, delete rejection rationale, ordering guarantees and the intra-batch caveat.

Verifying this change

This change added tests and can be verified as follows:

• Added unit tests for UpsertSerializationSchemaWrapper (INSERT pass-through, UPDATE_AFTER normalization + RowKind restore, restore-on-exception, DELETE/UPDATE_BEFORE rejection)
• Added factory tests: PK table produces the expected sink (PK partition key, upsert mode, max-inflight 1), changelog mode excludes DELETE/UPDATE_BEFORE, and sink.partitioner / sink.requests.max-inflight != 1 / PARTITIONED BY each fail validation
• Added a SQL-path integration test (KinesisUpsertTableSinkITCase, Localstack Kinesis): a GROUP BY upsert changelog is written end-to-end with primary-key-derived partition keys and the latest value per key is materializable from the stream; a delete-producing Top-N query fails at planning time with doesn't support consuming delete changes (also empirically confirming that GROUP BY over insert-only input is not inferred to produce DELETE)
• Full flink-connector-aws-kinesis-streams module suite passes, spotless, checkstyle and dependency-convergence clean

Significant changes

• Dependencies have been added or upgraded
• Public API has been changed (Public API is any class annotated with @Public(Evolving))
• Serializers have been changed
• New feature has been introduced
  • If yes, how is this documented? (docs)