mmodzelewski · Oct 29, 2026 · Oct 25, 2026
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/gateways/kafka/Cargo.toml b/gateways/kafka/Cargo.toml
@@ -39,6 +39,9 @@ base64 = { workspace = true }
 bytes = { workspace = true }
 futures = { workspace = true }
 iggy = { workspace = true }
+# `with_binary_deadline`, which hands a Produce request's deadline to the SDK. The prelude does not
+# export it.
+iggy_common = { workspace = true }
 iggy_kafka_wire = { workspace = true }
 # The codec features stay on for the encoder, which the batch tests use to build fixtures.
 kafka-protocol = { workspace = true, features = ["gzip", "lz4", "snappy", "zstd"] }

diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
@@ -477,7 +477,7 @@ One partition, one Iggy send. Each partition answers for itself.
 | `acks` | `0`, `1`, `-1` write the same. Other values: 21, before any per-partition check. |
 | `acks=0` | Writes, answers nothing. Any failed partition closes the connection. |
 | Topics | Creates none. Missing: 3. Bad name: 17. |
-| `timeout_ms` | Honored, max 20 s. Past it: 7. |
+| `timeout_ms` | Honored, max 20 s. The Iggy send stops by then too. Past it: 7. A send that never went out: 6. |
 | Compression | gzip, snappy, lz4. zstd from v7, else 76. |
 | Producer id | Picks the producer's Iggy session. See [Concurrency ceiling](#concurrency-ceiling). |
 | Producer epoch, sequence | Ignored, so a retry writes twice. |
@@ -495,8 +495,8 @@ One partition, one Iggy send. Each partition answers for itself.
 | 10 | Record, send or partition too large, even alone | Java splits multi-record batches. Else fails. |
 | 87 | Record the gateway cannot map. Reason in `error_message` from v8. | Fails. |
 | 35 | Transactional or control batch | Fails. |
-| 6 | Request budget ran out. Nothing written. | Retries. |
-| 7 | Deadline passed: in the send, waiting for decode memory, or behind the producer's own earlier send. Or connection lost mid-send. May be written. | Retries. Can duplicate. |
+| 6 | Request budget ran out. Or the send did not reach Iggy by the deadline, or Iggy refused it: no primary, partition busy or recreated, client shut down. Nothing written. | Retries. |
+| 7 | Deadline passed: after the send went out, waiting for decode memory, or behind the producer's own earlier send. Or connection lost mid-send. Possibly written. | Retries. Can duplicate. |
 
 Partition cap: `max_frame_size` decompressed bytes, `max_frame_size / 64` record slots, 3 headers
 per slot. Past it: 10.
@@ -610,8 +610,10 @@ Iggy becomes unreachable later.
 `send_records` uses the Produce deadline instead:
 
 - One request of a producer runs at a time, on that producer's client.
-- A send past its deadline keeps the producer's turn until it ends. That producer's requests answer
-  7 meanwhile. Other producers send on.
+- The SDK stops retrying the send 50 ms before the deadline. So it sends nothing after the request
+  answered, and the client's retry is the only copy of a send that never went out.
+- A send still running past its deadline keeps the producer's turn until it ends. That producer's
+  requests answer 7 meanwhile. Other producers send on.
 - After 45 s the gateway gives up on the send. It then closes that connection before the turn
   frees, so the send cannot land after a later one. The SDK ends its write within 30 s more,
   unless Iggy stopped reading: the producer then waits until that connection breaks.
@@ -821,6 +823,10 @@ instead of -1, because the Java `AdminClient` fails a partition on any non-retri
 incarnation of the partition, as after a delete and recreate; the retry probes again and answers 3
 if the topic is gone. ListOffsets reads do not get `RequestTooOld` today, so that arm is defensive.
 
+Produce also answers `HistoryUnavailable`, `LifecycleBusy` and `ClientShutdown` with 6 instead of
+-1 (`to_produce_error_code`). The SDK returns them only for a send that wrote nothing. A send that
+possibly landed ends as `OperationUncertain`, which answers 7.
+
 ### Server limits the gateway inherits
 
 These are Iggy server limits, not gateway settings. A Kafka client cannot act on any of them, so

diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md
@@ -84,7 +84,7 @@ file under `tests/` anymore.
 | [`bridge_iggy_integration_tests.rs`](../tests/bridge_iggy_integration_tests.rs) | `IggyBridge` against a real, spawned `iggy-server` — provisioning idempotency, high watermark, credential/connection edge cases, and the bridge dialing a TLS-only server over TLS | No (needs the `iggy-server` binary - see Prerequisites) |
 | [`list_offsets_real_bridge_tests.rs`](../tests/list_offsets_real_bridge_tests.rs) | ListOffsets (key 2) through the whole handler against a real, spawned `iggy-server`: LATEST, EARLIEST, lookups by time and `MAX_TIMESTAMP`, the `-1` answers, code 3, several partitions looked up by time in one request, and a stalled lookup leaving the later partitions their answers | No (needs the `iggy-server` binary - see Prerequisites) |
 | [`fetch_real_bridge_tests.rs`](../tests/fetch_real_bridge_tests.rs) | Fetch (key 1) through the whole handler against a real, spawned `iggy-server` — records go in through the Iggy SDK or the Produce handler and come back through `kafka_protocol`'s client decoder, at every version, with codes 0, 1, 3, 6, -1 and 70, paged polls, and the wait | No (needs the `iggy-server` binary - see Prerequisites) |
-| [`produce_real_bridge_tests.rs`](../tests/produce_real_bridge_tests.rs) | Produce (key 0) through the whole handler against a real, spawned `iggy-server` — records go in as Kafka wire bytes and come back through the Iggy SDK, plus one error code per partition | No (needs the `iggy-server` binary - see Prerequisites) |
+| [`produce_real_bridge_tests.rs`](../tests/produce_real_bridge_tests.rs) | Produce (key 0) through the whole handler against a real, spawned `iggy-server` — records go in as Kafka wire bytes and come back through the Iggy SDK, plus one error code per partition. Through a relay that cuts Iggy off, a send that misses its deadline answers 6, and the client's retry stores the batch once | No (needs the `iggy-server` binary - see Prerequisites) |
 | [`create_partitions_real_bridge_tests.rs`](../tests/create_partitions_real_bridge_tests.rs) | CreatePartitions (key 37) through the whole handler against a real, spawned `iggy-server`: a grow appends indices and shows in Metadata's listing, the grow-only rule (37), a missing topic (3), `validate_only`, duplicate names (42), the manual assignment checks (39), the per-request and topic caps, the `manage_topics` gate (29), a duplicate answered before authorization (42, then 29 for the rest), two overlapping grows to one count (one 0, one 37), a fixed partition topic (44, `validate_only` too), and v0/v1 requests | No (needs the `iggy-server` binary - see Prerequisites) |
 | [`offsets_real_bridge_tests.rs`](../tests/offsets_real_bridge_tests.rs) | OffsetCommit (key 8) and OffsetFetch (key 9) through the whole handler against a real, spawned `iggy-server`: offsets stored as sent, the Iggy external group key, a negative commit as a delete, -1 for nothing committed, a null topic list, and a gateway restart ([#3542](https://github.com/apache/iggy/issues/3542) acceptance) | No (needs the `iggy-server` binary - see Prerequisites) |
 | [`offset_groups_real_bridge_tests.rs`](../tests/offset_groups_real_bridge_tests.rs) | ListGroups (key 16), DescribeGroups (key 15), DeleteGroups (key 42) and OffsetDelete (key 47) through the whole handler against a real, spawned `iggy-server`. A group with offsets and no members is `Empty`, and a live group is listed once. Also the filters, a commit after the last scan, a new gateway, a native Iggy group, and only the live groups once a scan failed with Iggy down. DeleteGroups refuses a group with members, and after they leave deletes it on every topic. Through a relay that holds an Iggy call past its deadline, a commit and a group deletion each keep the other out until the call ends, and past the SDK's 30 s deadline until a replay proves what the call did. A delete that an earlier delete in doubt kept from Iggy lets go of its group. A deletion that ends after its request answered 14 answers 0 to the retry. OffsetDelete keeps the offsets of a topic that a member subscribes to, deletes the others, and finds a group that another gateway committed for since the last scan | No (needs the `iggy-server` binary - see Prerequisites) |

diff --git a/gateways/kafka/src/bridge/error.rs b/gateways/kafka/src/bridge/error.rs
@@ -196,6 +196,22 @@ impl BridgeError {
         }
     }
 
+    /// [`Self::to_kafka_error_code`] for Produce. The SDK returns `HistoryUnavailable`,
+    /// `LifecycleBusy` and `ClientShutdown` only for a send that wrote nothing, because a send
+    /// that may have landed ends as `OperationUncertain`. So these answer 6, which clients retry,
+    /// not -1, which loses the records.
+    #[must_use]
+    pub const fn to_produce_error_code(&self) -> i16 {
+        match self {
+            Self::Iggy(
+                IggyError::HistoryUnavailable
+                | IggyError::LifecycleBusy
+                | IggyError::ClientShutdown,
+            ) => ERROR_NOT_LEADER_OR_FOLLOWER,
+            _ => self.to_kafka_error_code(),
+        }
+    }
+
     /// True when the same call can succeed later: Kafka clients retry its code, 6 or 7. A wrong
     /// password, a missing permission or a bad configuration is not.
     #[must_use]
@@ -421,6 +437,47 @@ mod tests {
         }
     }
 
+    #[test]
+    fn given_a_send_that_wrote_nothing_when_mapped_for_produce_should_answer_retriable() {
+        let wrote_nothing = [
+            IggyError::HistoryUnavailable,
+            IggyError::LifecycleBusy,
+            IggyError::ClientShutdown,
+        ];
+        for error in wrote_nothing {
+            let err = BridgeError::Iggy(error);
+            assert_eq!(
+                err.to_kafka_error_code(),
+                ERROR_UNKNOWN_SERVER_ERROR,
+                "{err}"
+            );
+            assert_eq!(
+                err.to_produce_error_code(),
+                ERROR_NOT_LEADER_OR_FOLLOWER,
+                "{err}"
+            );
+        }
+    }
+
+    #[test]
+    fn given_a_send_that_may_have_landed_when_mapped_for_produce_should_keep_the_shared_code() {
+        let may_have_landed = [
+            (
+                BridgeError::Iggy(IggyError::OperationUncertain(1)),
+                ERROR_REQUEST_TIMED_OUT,
+            ),
+            (BridgeError::Timeout, ERROR_REQUEST_TIMED_OUT),
+            // A retry after the client was evicted could write it twice.
+            (
+                BridgeError::Iggy(IggyError::RequestTooOld),
+                ERROR_UNKNOWN_SERVER_ERROR,
+            ),
+        ];
+        for (err, code) in may_have_landed {
+            assert_eq!(err.to_produce_error_code(), code, "{err}");
+        }
+    }
+
     #[test]
     fn given_any_other_failure_when_mapped_for_list_offsets_should_keep_the_shared_code() {
         let others = [

diff --git a/gateways/kafka/src/bridge/iggy_bridge/produce.rs b/gateways/kafka/src/bridge/iggy_bridge/produce.rs
@@ -27,6 +27,7 @@ use std::time::Duration;
 
 use futures::future::join_all;
 use iggy::prelude::{IggyClient, IggyError, IggyMessage, MessageClient, Partitioning};
+use iggy_common::{binary_operation_deadline, with_binary_deadline};
 use tokio::runtime::Handle;
 use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
 use tokio::time::{Instant, MissedTickBehavior, interval, sleep_until, timeout, timeout_at};
@@ -39,6 +40,10 @@ use crate::bridge::error::BridgeError;
 /// Frame bytes besides the messages: request header, batch header and ids.
 const SEND_FRAME_ROOM: u64 = 4 * 1024;
 
+/// How much sooner than its request the SDK stops a send. The SDK's own answer then tells whether
+/// the send went out, before the request gives up on it with 7.
+const SDK_STOP_ROOM: Duration = Duration::from_millis(50);
+
 /// Sessions that connect and sign in at once. A sign-in costs Iggy a password hash on a shard
 /// thread, so a burst of new producers must not start them all together.
 const SESSION_OPENS: usize = 4;
@@ -374,6 +379,9 @@ impl IggyBridge {
     /// `Ok(None)`: committed, but the server named no offset (a duplicate request, or a journal
     /// entry that is gone). Do not retry it.
     ///
+    /// The SDK stops the send by `deadline`, so no attempt of it starts after the request
+    /// answered, and a client retry is the only copy of a send that never went out.
+    ///
     /// A send still running at `deadline` keeps `turn` and `kept`, and hands back `None` for
     /// them. So the producer's later requests answer 7 meanwhile and cannot overtake it, and what
     /// `kept` reserves for `messages` stays reserved. Other producers send on.
@@ -386,7 +394,8 @@ impl IggyBridge {
     /// [`BridgeError::Timeout`] if `deadline` passes. The send may have landed.
     /// [`BridgeError::SendLost`] if the connection broke during the send. It may have landed.
     /// [`BridgeError::Iggy`] if the session cannot connect, if the stream, topic or partition is
-    /// missing, or if the send is too large.
+    /// missing, or if the send is too large. Also when the SDK stopped the send at `deadline`:
+    /// `OperationUncertain` if it went out, so it may have landed, and its refusal if it did not.
     pub async fn send_records<K: Send + 'static>(
         &self,
         turn: ProduceTurn,
@@ -408,7 +417,13 @@ impl IggyBridge {
             return (Err(BridgeError::Timeout), Some((turn, kept)));
         }
         let (stream_id, topic_id) = (target.stream_id.clone(), target.topic_id.clone());
-        let recovery_deadline = (Instant::now() + SLOT_LIMIT).into_std();
+        // With less than the room left, the SDK keeps the whole deadline, and the send answers 7.
+        // Never past the SDK's own limit, which `SLOT_LIMIT` counts on.
+        let sdk_deadline = deadline
+            .checked_sub(SDK_STOP_ROOM)
+            .filter(|stop| *stop > Instant::now())
+            .unwrap_or(deadline)
+            .min(binary_operation_deadline());
         let send = async move {
             // Owns the messages, so a send cut off at the limit drops them.
             let send = {
@@ -421,13 +436,15 @@ impl IggyBridge {
                             .await
                         {
                             Err(IggyError::LaneBlocked(operation)) => {
-                                resolve_blocker(&client, operation, recovery_deadline).await?;
+                                resolve_blocker(&client, operation, sdk_deadline.into_std())
+                                    .await?;
                             }
                             result => return result,
                         }
                     }
                 }
             };
+            let send = with_binary_deadline(sdk_deadline, send);
             within_slot_limit(&session, &client, kept, send).await
         };
         // The SDK cannot cancel a send. One that outlives this call keeps the turn in its task,

diff --git a/gateways/kafka/src/protocol/handlers/produce.rs b/gateways/kafka/src/protocol/handlers/produce.rs
@@ -517,10 +517,11 @@ fn request_timeout(timeout_ms: i32) -> Duration {
         .min(REQUEST_DEADLINE)
 }
 
-/// Sends one partition's messages, or answers 7 once the deadline has passed. Hands `held` back,
+/// Sends one partition's messages, and gives up on them at the deadline. Hands `held` back,
 /// unless the send still runs and keeps it.
 ///
-/// 7 means nothing was written, or the send may still land and a retry duplicate it.
+/// 7 means nothing was written, or the send may still land and a retry duplicate it. 6 means that
+/// it never went out.
 async fn send_by(
     bridge: &IggyBridge,
     (turn, admission): Held,
@@ -535,22 +536,31 @@ async fn send_by(
         .await;
     let sent = match sent {
         Ok(base_offset) => Ok(base_offset),
-        Err(BridgeError::Timeout) => {
+        // The SDK retries a send until the deadline, and then ends it with one of these two: not
+        // sent, or in doubt.
+        Err(
+            error @ (BridgeError::Timeout
+            | BridgeError::Iggy(
+                IggyError::TransientNotAccepted | IggyError::OperationUncertain(_),
+            )),
+        ) => {
             // The client picks `timeout_ms`, so its expiry is no signal to the operator.
             if deadline.ceiling {
                 tracing::warn!(
+                    %error,
                     kafka_topic,
                     partition_id,
                     "Produce ran past the gateway deadline"
                 );
             } else {
                 tracing::debug!(
+                    %error,
                     kafka_topic,
                     partition_id,
                     "Produce ran past the client deadline"
                 );
             }
-            Err(ERROR_REQUEST_TIMED_OUT)
+            Err(error.to_produce_error_code())
         }
         Err(error) => Err(refusal_code(kafka_topic, &error)),
     };
@@ -559,7 +569,7 @@ async fn send_by(
 
 /// The Kafka code for a bridge refusal, logged at a level that matches who can fix it.
 fn refusal_code(kafka_topic: &str, error: &BridgeError) -> i16 {
-    let code = error.to_kafka_error_code();
+    let code = error.to_produce_error_code();
     if error.is_bridge_login_rejected() {
         tracing::error!(%error, kafka_topic, "Iggy rejected the bridge's credentials");
     } else if matches!(code, ERROR_UNKNOWN_SERVER_ERROR) {