GeezerVison

What I ran (PR head be944ac, debug build)

• Gate failed: flaky tests (high). Zero tolerance, so the fenced-producer assertion needs fixing before merge.
• Commit quality (medium). 2a4c15e ("Replay the transaction…") doesn't compile alone: kafka::metrics and SendError arrive only in the later commit be944ac. That breaks bisect on a linear history. Reorder or squash.
• Missing coverage (medium).
  • Nothing exercises the spill-to-file path through the endpoint. The 64 MiB SPOOL_MEMORY_LIMIT isn't injectable, so the spool's unit tests are the only coverage.
  • No test covers a replay that itself hits NeedsAbort (a broker that flaps during the replay), or the Terminal path followed by the next batch_start.
  • No test runs an outage longer than transaction.timeout.ms against a real broker, to check whether a coordinator-side abort comes back as abortable or as fenced/fatal.

QA test I added locally:

#[test]
fn qa_consecutive_outages_recover_independently() {
    let cluster = cluster();
    let (mut endpoint, callbacks, _shutdown) = endpoint(&cluster);
    let mut total = 0;
    for txn in 0..2u64 {
        endpoint.batch_start(txn, OutputBatchType::Delta).unwrap();
        cluster.broker_down(1).unwrap();
        for i in 0..3 {
            endpoint.push_buffer(format!("t{txn} m{i}").as_bytes()).unwrap();
        }
        total += 3;
        let before = callbacks.metric("kafka_output_delivery_errors_total");
        wait(|| callbacks.metric("kafka_output_delivery_errors_total") > before, 30_000).unwrap();
        let worker = thread::spawn(move || (endpoint.batch_end(), endpoint));
        thread::sleep(Duration::from_millis(1500));
        cluster.broker_up(1).unwrap();
        let (result, e) = worker.join().unwrap();
        result.unwrap();
        endpoint = e;
        assert_eq!(callbacks.metric("kafka_output_replay_spool_bytes"), 0.0);
    }
    assert!(callbacks.metric("kafka_output_transaction_aborts_total") >= 2.0);
    let expected: Vec<String> = (0..2).flat_map(|t| (0..3).map(move |i| format!("t{t} m{i}"))).collect();
    assert_eq!(read_topic(&cluster, total).into_iter().map(|(_, p)| p).collect::<Vec<_>>(), expected);
}

What I ran

Gates: tests, docs, changelog and breaking-change notes are present. No unsafe, no new dependencies. Per-type SQL coverage does not apply: the spool replays encoded bytes and does not depend on the column type.

Missing tests (medium)

• A spool I/O failure (temp dir full or read-only) should reach fail_fatally without committing. Nothing covers this today.
• A begin_transaction error on the Begin { replay: false } path (an abortable error there leads to an abort with no open transaction).
• A refused record that is replayed more than once (see the inline comment).

Flakiness (low): the mock tests depend on wall-clock timing: the 10 s API timeout, plus elapsed < 1s and < 15s assertions. They were stable here, but under a loaded CI runner shutdown_ends_a_stalled_commit has only a 5 s margin.

Overall the design looks sound. Please address the double-reporting bug before merge.

My first thought in looking at this was that it's odd to record and replay transactions instead of just retrying them forever. Looking closer, I found out that this doesn't work because librdkafka puts a limit on the length of a transaction, and even if we made that very long, Kafka brokers also cap the length of a transaction (at 15 minutes by deafult). So we have to record and replay.

I feel like this should be mentioned somewhere, probably both in comments and in the commit message, because it's surprising.

These notes are from Claude (an AI assistant), posted at the request of blp, who is also writing a separate review. Line numbers refer to crates/connectors/src/transport/kafka/ft/output.rs at 71463b1.

The earlier bot findings (double report on replay, spill directory, SendError duplicate cause, flaky errors()[0], commit that did not build alone) look fixed. These are new:

1. A record that the broker refuses at delivery time makes the sink loop forever (high)

send_record (L205-221) only classifies refusals that librdkafka returns from produce(). The broker can also refuse a single record at delivery time, for example:

• a record larger than the topic's max.message.bytes, which can be smaller than the client's message.max.bytes;
• INVALID_RECORD (compaction requirements, broker-side schema validation);
• CORRUPT_MESSAGE.

That failure goes to delivery() (L1018-1029), which only counts it and logs at debug level. The transaction then needs an abort. The sink aborts, replays the same record, the broker refuses it again, and this repeats every 10 s with no end. The only signal is the unhealthy status, with no connector error. Before this change, the sink stopped after 5 retries. The docs say "a permanently refused record ... is dropped and reported", which is not true for this case.

Possible fix: set DeliveryOpaque to the spool index, so that the delivery callback can sort out per-record permanent codes and add the index to refused before the next replay.

2. A transaction that cannot be sent within transaction.timeout.ms never commits (medium)

If sending the spool plus the commit takes longer than transaction.timeout.ms (60 s by default), every replay expires on the broker. The sink aborts and replays without end. The first output after a backfill or a snapshot is the likely case, and the docs use that same case for spool sizing. Nothing detects this. As a minimum, after N consecutive aborts in drive (L476-491), log a warning that names transaction.timeout.ms, and document this limit.

3. Hot-path cost (medium, request for numbers)

push_key (L709-723) now copies every payload and clones the configured headers for each record. Above 64 MiB it also serializes each record with rmp to disk. This happens on the normal path, when there is no failure. A before/after throughput number for a large output would show if this is significant.

4. Smaller items

• L553-577 is not formatted. The long string on L566 stops rustfmt in the closure. Move the message out and rustfmt can format the match.
• The error text on L583 says "temporary directory" and L758 says "spool directory". The file can now be in the storage directory, so use one term.
• TopicAuthorizationFailed and UnknownTopicOrPartition go to NeedsAbort (L217-219). That is correct for a short metadata glitch. But for a deleted topic or a revoked ACL, the sink aborts and replays forever with only an unhealthy status. Consider a connector error after some time.
• blp noted earlier that record-and-replay is necessary because librdkafka and the broker limit how long a transaction can stay open. The docs now say this, but a code comment near ReplaySpool or drive, and a sentence in the commit message, would help the next reader.