[hly0025]

Description

Related to #3181 and #3182. #3182 makes the producer request a new id after reset_producer_id(); this makes that request survive a request timeout.

Root Cause

TxnRequestHandler.on_complete re-enqueues a request that failed with KafkaConnectionError and treats every other exception as fatal
(L859-L871). A client-side request timeout closes the connection with RequestTimedOutError
(net/connection.py L190-L191), and fail_in_flight_requests
(L245) fails every request in flight on it with that error.

When an in-retry batch expires, the sender calls reset_producer_id() (sender.py L276-L277), which since #3182
returns the manager to UNINITIALIZED, and init_producer_id() sends a new InitProducerId. If the broker is still not answering, a request on the same connection can reach request_timeout_ms while that InitProducerId is in flight.
The connection is closed, the InitProducerId fails with RequestTimedOutError, and on_complete calls fatal_error():
Fatal Error handling request InitProducerIdRequest: ... RequestTimedOutError. The producer stays in FATAL_ERROR without a producer id, nothing is sent afterwards, and only a new producer recovers. The broker never rejected the request.

Observed end to end with kafka-python 3.0.13 in a service against a Kafka 3.8 cluster in Docker, with one in-sync follower frozen (docker pause) for 166 s: the batch expired at 120 s, the new InitProducerId was killed by a request
timeout, and topic offsets stayed flat for 10 minutes after the cluster was healthy again.

The Java client handles this as a disconnect: InFlightRequest.timedOut() is documented "A timed out request is considered disconnected as well" (NetworkClient.java
L1860-L1864), and TxnRequestHandler.onComplete retries on response.wasDisconnected() with the log line
"Disconnected from {}. Will retry." that this code mirrors (TransactionManager.java
L1469-L1473).

Changes

• Re-enqueue RequestTimedOutError in TxnRequestHandler.on_complete exactly like KafkaConnectionError (including the coordinator lookup when the handler needs one). No new retry path: a request timeout already closes the connection,
  and a closed connection is already retried. on_complete is shared by all TxnRequestHandlers, so they get the same treatment.
• Add test_idempotent_producer_reinit_retries_after_request_timeout in test/producer/test_transaction_manager_mock_broker.py. It uses the MockBroker harness and the real client request timeout, with no injected exception: the
  idempotent producer acquires an id, then reset_producer_id() and init_producer_id() run as in the sender; the broker never answers the InitProducerId, so the client's request_timeout_ms (300 ms) fires. The test asserts the manager
  is not fatal and the handler is re-enqueued, then answers the retry and asserts the manager reaches READY with the new producer id.
• _make_client takes an optional request_timeout_ms (default unchanged, 5000).

Test Evidence

• On master (3fa6acf) the new test fails at assert not tm.has_fatal_error() (5 of 5 runs). With this change it passes (20 of 20 runs, about 0.35 s).
• pytest test/producer test/test_mock_broker.py (334 passed)
• repro_request_timeout_reinit.py (attached) runs the same scenario without pytest and needs no cluster. Run it from the root of a checkout: on master it prints RESULT: BUG REPRODUCED and exits 1; with this change it prints RESULT: FIXED and exits 0. The outputs of both runs are attached.
• A build that retries RequestTimedOutError obtained a new producer id and resumed in the end-to-end setup above. That build also retries NodeNotReadyError and KafkaTimeoutError; the narrower change proposed here was validated with
  the test and script, not in that service. I could not trigger the fatal state with a bare KafkaProducer script against a Docker cluster at shortened timeouts: it depends on a request timing out while the new InitProducerId is in
  flight, which is why the test and script control that timing directly.

Diagram and Reproducibility Script

#!/usr/bin/env python3
"""Minimal reproduction: an idempotent producer goes to FATAL_ERROR, permanently, when a request timeout
fails the InitProducerId it sends to re-acquire its producer id.

Needs no Kafka cluster and no Docker: it uses kafka-python's own MockBroker test harness. Run it from the
root of a kafka-python checkout (the script imports the repository's `test` package):

    python /path/to/repro_request_timeout_reinit.py

What it does (the same steps the producer's Sender performs):
  1. An idempotent producer (transactional_id=None, the default configuration) acquires a producer id.
  2. A batch that was retried reaches delivery_timeout_ms, so the Sender calls reset_producer_id()
     (since #3182 the manager returns to UNINITIALIZED) and init_producer_id() requests a new id.
  3. The broker does not answer in time. The client's request timeout fires and closes the connection with
     RequestTimedOutError, which fails every request in flight on it, including this InitProducerId.
     (In a real cluster the trigger is any older request on the same connection reaching its timeout;
     here the InitProducerId request itself is the one that is never answered.)
  4. TxnRequestHandler.on_complete retries only KafkaConnectionError. Any other exception, including
     RequestTimedOutError, ends in fatal_error(): the producer never obtains an id and never sends again.

The client timeout is shortened to 300 ms (the library default is 30000) so the run takes about a second.

Exit code 1: BUG REPRODUCED (the producer is FATAL_ERROR). Exit code 0: FIXED (the request is retried and
the producer obtains a new id).
"""
import logging
import os
import sys
import time

sys.path.insert(0, os.getcwd())

import kafka  # noqa: E402
import kafka.errors as Errors  # noqa: E402
from kafka.net.compat import KafkaNetClient  # noqa: E402
from kafka.producer.transaction_manager import TransactionManager, TransactionState  # noqa: E402
from kafka.protocol.producer import InitProducerIdRequest, InitProducerIdResponse  # noqa: E402
from test.mock_broker import MockBroker  # noqa: E402

REQUEST_TIMEOUT_MS = 300


def say(msg):
    print(msg, flush=True)


def send_next_request(client, tm):
    """Send the next queued transactional request, as Sender._maybe_send_transactional_request does."""
    handler = tm.next_request_handler(has_incomplete_batches=False)
    client.await_ready(0, timeout_ms=5000)
    correlation_id = tm.next_in_flight_request_correlation_id()
    future = client.send(0, handler.request)
    future.add_both(handler.on_complete, correlation_id)
    return handler, future


def wait_for(client, future, timeout_s=5):
    deadline = time.monotonic() + timeout_s
    while not future.is_done and time.monotonic() < deadline:
        client.poll(future=future, timeout_ms=200)


def main():
    logging.basicConfig(level=logging.WARNING, format='    [%(name)s] %(message)s', force=True)
    say('kafka-python %s from %s' % (kafka.__version__, os.path.dirname(kafka.__file__)))

    broker = MockBroker(broker_version=(4, 2))
    client = KafkaNetClient(bootstrap_servers='%s:%d' % (broker.host, broker.port), api_version=broker.broker_version,
                            request_timeout_ms=REQUEST_TIMEOUT_MS, metadata_max_age_ms=300000)
    broker.attach(client._net)
    client.check_version(timeout_ms=5000)
    try:
        tm = TransactionManager(transactional_id=None, api_version=(2, 5), metadata=client.cluster)

        say('step 1: the idempotent producer acquires a producer id')
        tm.init_producer_id()
        broker.respond(InitProducerIdResponse, InitProducerIdResponse(
            throttle_time_ms=0, error_code=0, producer_id=1234, producer_epoch=5))
        _, future = send_next_request(client, tm)
        wait_for(client, future)
        say('         state=%s producer_id=%s' % (tm._current_state.name, tm.producer_id_and_epoch.producer_id))

        say('step 2: a retried batch expires, so the Sender calls reset_producer_id(); a new id is requested')
        tm.reset_producer_id()
        tm.init_producer_id()
        say('         state=%s producer_id=%s' % (tm._current_state.name, tm.producer_id_and_epoch.producer_id))

        async def never_answers(api_key, api_version, correlation_id, request_bytes):
            await client._net.sleep(60)

        say('step 3: the broker does not answer InitProducerId; the client request timeout (%d ms) fires' % REQUEST_TIMEOUT_MS)
        broker.respond_fn(InitProducerIdRequest, never_answers)
        handler, future = send_next_request(client, tm)
        wait_for(client, future)
        say('         InitProducerId failed with: %r' % (future.exception,))

        fatal = tm.has_fatal_error()
        say('step 4: state=%s fatal_error=%s request_requeued=%s' % (
            tm._current_state.name, fatal, any(h is handler for _, _, h in tm._pending_requests)))
        if not isinstance(future.exception, Errors.RequestTimedOutError):
            say('RESULT: INCONCLUSIVE: the request did not fail with RequestTimedOutError')
            return 2
        if fatal:
            say('RESULT: BUG REPRODUCED: the producer is in %s and will never obtain a producer id' % tm._current_state.name)
            return 1

        say('step 5: the request was retried; the broker answers the retry')
        broker.respond(InitProducerIdResponse, InitProducerIdResponse(
            throttle_time_ms=0, error_code=0, producer_id=1235, producer_epoch=0))
        _, future = send_next_request(client, tm)
        wait_for(client, future)
        say('         state=%s producer_id=%s' % (tm._current_state.name, tm.producer_id_and_epoch.producer_id))
        if tm._current_state == TransactionState.READY and tm.producer_id_and_epoch.producer_id == 1235:
            say('RESULT: FIXED: the request timeout was retried and the producer obtained a new producer id')
            return 0
        say('RESULT: UNEXPECTED: state=%s' % tm._current_state.name)
        return 2
    finally:
        logging.disable(logging.CRITICAL)  # silence the mock network loop's teardown messages
        client.close()


if __name__ == '__main__':
    code = main()
    sys.stdout.flush()
    os._exit(code)  # skip interpreter teardown noise from the mock network loop

Not changed (open question, not needed for this fix)

Other exceptions believed to be able to reach on_complete, such as NodeNotReadyError and KafkaTimeoutError, still end in fatal_error(). I left them out because I have no reproduction for either with default settings, and I wanted this change to cover only the failure I could show. Cancelled is likely better left fatal, since it can arrive while the producer is closing. If you prefer parity with Java's broader wasDisconnected() handling, I am happy to extend the change in a follow-up or made by the maintainer(s)

Prepared with AI assistance (Claude), reviewed for correctness before submission

[dpkp]

We might also consider making RequestTimedOutError a subclass of KafkaConnectionError.

[dpkp]

Thanks for the find!