Redpanda output with idempotent_writes makes unordered results

[pserwacinski]

Hi Team

I have unordered results using redpanda input/output. Source topic has been replied 3 times, so it contains the same blocks of messages. For some reason, the 3rd block got disordered as on screenshot below.

My input and output configuration looks like below

output:
    label: kafka_output
    processors:
        - label: remove_build_in_meta_added
          mapping: |-
            let blacklist = ["kafka_topic", "kafka_partition", "kafka_offset",
            "kafka_lag", "kafka_timestamp_unix", "kafka_tombstone_message",
            "backoff_duration", "retry_count", "schema_id"]

            meta = metadata().filter(kv -> !$blacklist.contains(kv.key))
    redpanda:
        seed_brokers:
            - output-kafka-bootstrap.svc.cluster.local:9092
        topic: ${! meta("topic") }
        key: ${! meta("kafka_key") }
        timestamp_ms: ${! meta("kafka_timestamp_ms") }
        partitioner: murmur2_hash
        max_message_bytes: 5MB
        compression: zstd
        metadata:
            include_patterns:
                - .*
        sasl:
            - username: streams
              password: ${sasl_password}
              mechanism: SCRAM-SHA-512
        tls:
            enabled: true
            skip_cert_verify: true
        idempotent_write: true
        max_in_flight: 1

input:
    label: kafka_input
    redpanda:
        seed_brokers:
            - input-kafka-bootstrap.svc.cluster.local:9092
        topics:
            - source-topic
        consumer_group: source-topic
        auto_replay_nacks: true
        sasl:
            - username: streams
              password: ${sasl_password}
              mechanism: SCRAM-SHA-512
        tls:
            enabled: true
            skip_cert_verify: true
        fetch_max_partition_bytes: 5MB
        fetch_max_bytes: 50MB
        transaction_isolation_level: read_committed
        max_yield_batch_bytes: 5MB

I am using red panda connect version v4.61.0

I use idempotent writes and max in flight 1, I understand there might be duplicates as this is not the same as exactly once semantic, but to my understanding this should ensure topic is effectively ordered by looking at given key, meaning that in my example last record should have value 4529.49 and not 2203.57

Can you please tell me if this configuration should ensure data is effectively ordered within the scope of the same key?

[twmb]

Hi @pserwacinski — thanks for the detailed report. This is the documented at-least-once / in-order tradeoff that the redpanda input docs call out, not a bug in idempotent_write itself. The short version of what's happening:

1. idempotent_write: true only dedupes the franz-go client's internal retries of an unacknowledged in-flight batch (same producer ID, same sequence number). It does not dedupe new Produce calls that come from a Connect-level batch replay.
2. max_in_flight: 1 only serializes batches at the Connect layer. Per-partition order is preserved by sequence numbers within a single producer Produce sequence — but not across replays.
3. With auto_replay_nacks: true on the input, when the output writes a batch and some records succeed at the broker but at least one fails (e.g. transient leader change for one of the destination partitions touched by the batch), the whole batch gets nacked and re-delivered. The output then re-Produces every record — the previously-succeeded ones now appear as duplicates, and the originally-earlier-but-failed records get appended after them on the partition. That's the same-key reordering you're seeing on the 3rd block.

So your config is doing what it says on the tin while everything succeeds on the first try; the moment a partial-batch failure occurs, all three knobs you set (idempotent_write, max_in_flight: 1, murmur2_hash) are bypassed by the replay path.

Recommended config for strict ordering

Per the redpanda input ordering docs, switch off auto_replay_nacks and use a fallback output that wraps the redpanda output with an inner retry for backpressure plus a DLQ for poison records:

input:
  redpanda:
    seed_brokers: [ input-kafka-bootstrap.svc.cluster.local:9092 ]
    topics: [ source-topic ]
    consumer_group: source-topic
    auto_replay_nacks: false   # disable the whole-batch replay path
    sasl:
      - username: streams
        password: ${sasl_password}
        mechanism: SCRAM-SHA-512
    tls:
      enabled: true
      skip_cert_verify: true
    fetch_max_partition_bytes: 5MB
    fetch_max_bytes: 50MB
    transaction_isolation_level: read_committed
    max_yield_batch_bytes: 5MB

output:
  fallback:
    - retry:                   # backpressure on transient errors
        output:
          redpanda:
            seed_brokers: [ output-kafka-bootstrap.svc.cluster.local:9092 ]
            topic: ${! meta("topic") }
            key: ${! meta("kafka_key") }
            timestamp_ms: ${! meta("kafka_timestamp_ms") }
            partitioner: murmur2_hash
            max_message_bytes: 5MB
            compression: zstd
            metadata:
              include_patterns: [ ".*" ]
            sasl:
              - username: streams
                password: ${sasl_password}
                mechanism: SCRAM-SHA-512
            tls:
              enabled: true
              skip_cert_verify: true
            idempotent_write: true
            max_in_flight: 1
            processors:
              - mapping: |
                  let blacklist = ["kafka_topic", "kafka_partition", "kafka_offset",
                    "kafka_lag", "kafka_timestamp_unix", "kafka_tombstone_message",
                    "backoff_duration", "retry_count", "schema_id"]
                  meta = metadata().filter(kv -> !$blacklist.contains(kv.key))
    - redpanda:                # DLQ for unrecoverable errors
        seed_brokers: [ output-kafka-bootstrap.svc.cluster.local:9092 ]
        topic: ${! meta("topic") }_dlq
        key: ${! meta("kafka_key") }
        sasl:
          - username: streams
            password: ${sasl_password}
            mechanism: SCRAM-SHA-512
        tls:
          enabled: true
          skip_cert_verify: true

With this shape, transient errors apply backpressure via the inner retry (so franz-go's idempotent retry, with the same sequence numbers, is the only retry mechanism that runs — and order is preserved), and unrecoverable errors land in the DLQ instead of triggering a whole-batch replay.

A small caveat: today the redpanda output reports an error for the whole batch even when only one record failed, so the DLQ would currently receive duplicates of every record in a partially-failed batch rather than only the failed one(s). I'm planning to send a follow-up PR to make the output emit per-record errors (same approach as the legacy kafka output's service.NewBatchError) so the DLQ pattern works cleanly. I'll link the PR here.

[twmb]

PR linked: #4400

[pserwacinski]

Thank you for very detailed explanation of the root cause.

With auto_replay_nacks: true on the input, when the output writes a batch and some records succeed at the broker but at least one fails (e.g. transient leader change for one of the destination partitions touched by the batch), the whole batch gets nacked and re-delivered. The output then re-Produces every record — the previously-succeeded ones now appear as duplicates, and the originally-earlier-but-failed records get appended after them on the partition. That's the same-key reordering you're seeing on the 3rd block.

I understand why key reordering can happen in a scenario where the replay happens at the Connect level. However, as you mentioned, if the entire batch gets replayed in that case, why do I still not see the correct sequence of messages eventually?

In my example, when the yellow and red messages are present in the output without the blue and brown messages between them, I would assume that the batch containing them was nacked, and therefore the entire batch should have been replayed. If so, shouldn’t I eventually observe the correct sequence like on hypothetical scenario below?

[twmb]

Alright, modifying this to the current state of the theory. #4400 wont address your issue but doesn't hurt.

@pserwacinski - The actual (supposedly) real cause is a real bug in benthos's AsyncReader, not in kgo or connect's redpanda output:

When auto_replay_nacks: true is set, a nacked batch is supposed to be re-delivered before any subsequent fresh batch. The autoretry list (internal/autoretry/auto_retry_list.go) prioritises retries over fresh reads, which is correct in isolation. But AsyncReader.loop reads ahead — after pushing batch N to its unbuffered transactions channel, it spawns an ack-handler goroutine and immediately calls ReadBatch for batch N+1. By the time batch N errors and the ack-handler queues the retry, the next ReadBatch has already returned a fresh batch and pushed it onto the channel. The retry ends up scheduled behind the read-ahead. With same-key records concentrated on a single source partition, that surfaces at the destination as exactly the kind of intra-block reordering you screenshotted.

I've prepared a fix in benthos that splices queued retries ahead of any fresh batch the AsyncReader has read but not yet handed off downstream. A regression test (TestBatchAutoRetryReadAheadOrdering) reproduces the reordering on the unfixed code and passes with the fix.

Once that lands and a connect release picks it up, your config should preserve order without further changes. Until then, the workaround I posted earlier - auto_replay_nacks: false + fallback output with inner retry + DLQ - still applies; it sidesteps the autoretry path entirely so the read-ahead race never matters.

[pserwacinski]

Hi and thank you again for the diagnosis.

When do you think that PR will make its way to the master?

Focusing on the workround now, I have following questions the the proposed solution:

1. When using the fallback with retry (as in the example configuration provided earlier, where retries are unlimited), I would expect that if a batch cannot be successfully processed for any reason, it will be retried indefinitely. In that case, why is a DLQ needed as a secondary output?
2. If a batch is being retried on the output side and auto_replay_nacks is set to false on the input, does that mean that all subsequent batches from the input are blocked until the currently retried batch succeeds?
3. More generally, my requirement is that the output should contain the complete list of records from the input. Duplicates are allowed, but the effective order of values for a given key must be preserved, as in the hypothetical scenario below.

Does this workround satisfies my requirements and if not, can it be tuned to satisfy them?