[Bug]: Kafka progress is discarded by a later failed statement although the earlier Kafka INSERT commits

# [Bug]: Kafka progress is discarded by a later failed statement although the earlier Kafka INSERT commits

## Summary

Inside an explicit MatrixOne transaction, a successful `INSERT ... SELECT FROM` an `ENGINE=KAFKA` external table defers its Kafka progress until transaction commit, as designed. If a later unrelated ordinary statement fails but MatrixOne rolls back only that last statement, a subsequent `COMMIT` makes the earlier Kafka-derived rows durable. However, the failed statement clears all deferred Kafka progress. `LAST_KAFKA_MESSAGE_ID()` remains `NULL` and the consumer-group offset is never committed.

This splits durable data from its consumption progress and breaks the advertised exactly-once chaining contract.

## Environment

- MatrixOne SQL endpoint: `127.0.0.1:16001`
- Version string: `8.0.30-MatrixOne-v1.3.0`
- Running binary source commit: `a3027255760850f891e6dad6be95abfe8c7588f8`
- Public `main` checked at: `ae12701a888fac102e9641cc7cbb8706527366d8` (2026-08-25 06:31 UTC)
- Kafka: 4.3.1, local plaintext broker `127.0.0.1:19092`
- External table: `ENGINE=KAFKA`, `format=jsonl`, `autocommit=true`, one partition
- Consumer group, external table, sink table, and error table were globally unique for every repetition.

The relevant frontend source files in public `main` are byte-for-byte identical to the locally tested source files:

| File | SHA-256 |
|---|---|
| `pkg/frontend/mysql_cmd_executor.go` | `8256af13e7ed44dfd670fdc0a4afb78d20be5ed00091d9f2e8380cdb834cd639` |
| `pkg/frontend/session_kafka.go` | `081ee05415451019b6496eaed455ff21158db48345e597a0c8523d0d701e2f33` |
| `pkg/frontend/txn.go` | `a3e2c67e60cc02dab5bc935edc60359fd8c2fc7119f4da8b8ddc93ada54837c7` |

## Minimal prerequisites

Create a Kafka topic with partition 0 containing exactly one committed JSON object compatible with this schema:

```json
{"event_id":"event-0","producer_id":0,"producer_seq":0,"key":"key-0","target_partition":0,"business_group":0,"numeric_value":1,"nullable_value":null,"checksum":1,"batch_id":0,"attempt":"original"}
```

Use a fresh consumer group that has no committed offset.

## Minimal reproduction

```sql
CREATE DATABASE mo_kafka_progress_repro;
USE mo_kafka_progress_repro;

CREATE EXTERNAL TABLE kafka_src (
event_id VARCHAR(96),
producer_id INT,
producer_seq INT,
`key` VARCHAR(64),
target_partition INT,
business_group INT,
numeric_value BIGINT,
nullable_value BIGINT NULL,
checksum BIGINT,
batch_id INT,
attempt VARCHAR(16)
) ENGINE=KAFKA WITH (
'brokers'='127.0.0.1:19092',
'topic'='',
'partition'='0',
'autocommit'='true',
'group'='',
'format'='jsonl'
);

CREATE TABLE sink (
kafka_offset BIGINT PRIMARY KEY,
event_id VARCHAR(96)
);
CREATE TABLE unrelated (id INT PRIMARY KEY);
INSERT INTO unrelated VALUES (1);

BEGIN;
INSERT INTO sink
SELECT __mo_message_id, event_id
FROM kafka_src
WHERE __mo_read_size=1 AND __mo_read_timeout=5;
-- succeeds, affected rows = 1

INSERT INTO unrelated VALUES (1);
-- expected duplicate-key error 1062

COMMIT;
-- succeeds

SELECT * FROM sink;
-- actual: (0, 'event-0') is durable

SELECT LAST_KAFKA_MESSAGE_ID();
-- actual: NULL
```

Then inspect `` with Kafka's read-only consumer-group describe command. The group has no committed offset. The same test without the duplicate-key statement produces `LAST_KAFKA_MESSAGE_ID() = 0` and group offset `1`.

The executable local harness used for the evidence is `kafka_native_progress_loss_min_repro.py`. From the QA workspace:

```bash
cd /Users/ljy/Documents/Codex/2026-08-20/ce-s
MO_PORT=16001 python3 work/kafka_native_progress_loss_min_repro.py \
--database mo_kafka_native_txn_1787643890_dd94c4d8e9 \
--topic mo-kafka-native-txn-1787643890-b6fce530d704 \
--output work/kafka-native-concurrent-txn-runs/1787643890-b6fce530d704/progress-loss-min-repro.json \
--repetitions 3
```

## Expected result

MatrixOne uses statement-level rollback for the later duplicate-key error and permits `COMMIT`. Therefore the earlier successful Kafka INSERT and its deferred progress must remain paired:

- durable sink rows: `1`
- `LAST_KAFKA_MESSAGE_ID()`: `0`
- Kafka consumer-group offset: `1`

An alternative atomic result would be to abort the whole transaction, leaving both rows and progress absent. Durable data with absent progress is not atomic.

## Actual result

Three independent repetitions, each with a new external table, sink, error table, and consumer group:

| Repetition | Control data / last / group | Candidate data / last / group | Split |
|---:|---|---|---|
| 1 | `1 / 0 / 1` | `1 / NULL / absent` | yes |
| 2 | `1 / 0 / 1` | `1 / NULL / absent` | yes |
| 3 | `1 / 0 / 1` | `1 / NULL / absent` | yes |

The candidate's durable row was exactly Kafka partition 0 offset 0 and had the independently verified event id `mo-kafka-native-txn-1787643890-b6fce530d704-event-5-3` in all repetitions.

A larger four-way audit also reproduced the split three times with 5,000 messages per candidate:

- `BEGIN; Kafka INSERT; ROLLBACK`: correct (`0 / NULL / absent`)
- `BEGIN; Kafka INSERT; COMMIT`: correct (`5000 / 5790 / 5791`)
- two successful Kafka scans followed by `COMMIT`: correct (`5000 / 5790 / 5791`)
- successful Kafka scan, later duplicate error, then `COMMIT`: incorrect (`5000 / NULL / absent`)

An independent non-Kafka prerequisite test confirmed normal MatrixOne semantics: `BEGIN;` successful ordinary INSERT; later duplicate-key error; `COMMIT` succeeded and preserved the first INSERT. Thus this is not based on an assumption that any statement error must abort the full transaction.

## Impact

The next consumer using the same group sees the already-materialized Kafka record again. A retry may fail on a sink uniqueness constraint, require `INSERT IGNORE`, or duplicate downstream effects. The persisted data and the advertised progress marker cannot be treated as one transaction.

## Implementation analysis

Public `main` still contains this sequence:

1. A completed Kafka scan enqueues a transaction-owned progress finalizer.
2. On any later statement error, `executeStmtWithResponse` calls `ses.FinalizeKafkaProgress(false)` (`pkg/frontend/mysql_cmd_executor.go:4793-4802`).
3. `FinalizeKafkaProgress(false)` takes and clears the entire session `kafkaProgressQueue`, not only entries owned by the failing statement (`pkg/frontend/session_kafka.go:61-73`).
4. The transaction rollback path also finalizes the entire queue before deciding whether it will roll back the whole transaction or only call `RollbackLastStatement` (`pkg/frontend/txn.go:927-965`).
5. For the duplicate-key case, only the failed ordinary statement is rolled back. The earlier Kafka-derived rows remain in the transaction and later commit, but their already-cleared progress finalizer is no longer available to the commit terminal.

The queue needs statement/savepoint ownership so that statement-level rollback discards only Kafka progress created by that failed statement, while preserving progress from prior successful statements still present in the enclosing transaction.

## Public issue/PR deduplication

Read-only GitHub searches were performed on 2026-08-25 for Kafka issues/PRs and the exact identifiers/error paths (`LAST_KAFKA_MESSAGE_ID`, `FinalizeKafkaProgress`, `kafkaProgressQueue`, Kafka progress + transaction/rollback/offset). No matching existing bug or fix was found.

- #27518 is the closed feature request defining Kafka external tables and exactly-once chaining.
- PR #27531 implements the feature and fixed downstream statement failure, full transaction rollback, read-committed isolation, and other reviewed cases. Its transaction regression covers `Kafka INSERT; ROLLBACK` and `Kafka INSERT; COMMIT`, but not a later unrelated statement failure followed by commit.
- Public `main` at `ae12701` retains the faulty queue-wide discard path.
- Permanently excluded prior issues #27427, #27432/#27444, #27525/#27544 are unrelated and were not treated as candidates.

After the local QA and deduplication were complete, this report was submitted as #27587 with explicit user authorization and assigned to `Lundomn`. No PR was created or modified.