xai-org/x-algorithm · critical · Exception

Client registration failed: {register_response.error_message

Error message

Client registration failed: {register_response.error_message}

What it means

initialize_clients registers dataloader client IDs with the KafkaDispatcher via InitClientMessage; if the server rejects registration it returns success=False with an error_message, which this code re-raises as a generic Exception. Registration failure means the dispatcher will not assign partitions/feed data to those clients.

Source

Thrown at phoenix/xrex/data/streaming/kafkadispatcherloader.py:126

            for partition, offset in seek_to_offset.items()
        ]
        if seek_to_offset
        else None
    )
    seek_request = SeekRequest()
    if _seek_to_offset is not None:
        seek_request.offsets.CopyFrom(SeekOffsets(offsets=_seek_to_offset))
    elif seek_to_timestamp_ms is not None:
        seek_request.timestampMs = seek_to_timestamp_ms
    elif reset_to_latest:
        seek_request.latest = reset_to_latest
    register_response = await stub.InitClientMessage(
        InitClientRequest(register_client_ids=clients_to_register, seek_request=seek_request)
    )
    rank_logger.info(f"Register response: {register_response}")
    if not register_response.success:
        rank_logger.error(f"Failed to register clients: {register_response.error_message}")
        raise Exception(f"Client registration failed: {register_response.error_message}")


async def consume_messages(
    grpc_host_template: str,
    grpc_port: int,
    num_shards: int,
    shard_index: int,
    topic_name: str,
    seek_to_offset: dict[int, int] | None,
    seek_to_timestamp_ms: int | None,
    reset_to_latest: bool,
    fetch_batch_size: int,
    batch_size: int,
    post_process_fn: Callable[[list[pa.RecordBatch]], RecsysFeaturesBatch],
    example_queue: queue.Queue[tuple[RecsysFeaturesBatch, dict[int, int]]],
    _stop_event: threading.Event | None = None,
    grpc_timeout: float = 30.0,
):

View on GitHub (pinned to 24c60942c5)

Solutions

Read error_message in the exception: it usually states the exact server-side rejection reason; address that (e.g. reset dispatcher state for stale clients).
Restart/reset the KafkaDispatcher service so registrations start clean, then retry the job.
Ensure num_clients and client_ix match the dispatcher's expected shard layout (multiple of num_servers*num_processors).
If jobs are retried, deregister client IDs on shutdown or use unique run-scoped client IDs.

Example fix

# before
clients_to_register = [f"client-{ix}" for ix in range(num_clients)]
await initialize_clients(stub, clients_to_register, seek_request)

# after
run_id = os.environ.get("RUN_ID", str(uuid.uuid4())[:8])
clients_to_register = [f"client-{run_id}-{ix}" for ix in range(num_clients)]
await initialize_clients(stub, clients_to_register, seek_request)

Defensive patterns

Strategy: retry

Validate before calling

probe = await stub.InitClientMessage(InitClientRequest(register_client_ids=[], seek_request=None))
# or verify dispatcher health before registering all clients

Try / catch

try:
    await initialize_clients(stub, client_ids, seek_request)
except Exception as e:
    if "Client registration failed" in str(e):
        # inspect error_message; reset dispatcher state or use fresh IDs, then retry once
        rank_logger.error(f"registration rejected: {e}")
        raise

Prevention

Use run-scoped unique client IDs to avoid stale-registration collisions.
Deregister clients on job shutdown.
Restart dispatcher between topology changes (client count changes).
Match num_clients to num_servers*num_processors multiples.

When it happens

Trigger: Calling consume_messages when the dispatcher rejects InitClientMessage: duplicate/stale client IDs already registered, dispatcher at capacity, shard layout mismatch (client count not divisible by servers*processors triggers the preceding assert, but server-side mismatches surface here), or dispatcher in a bad state.

Common situations: Restarting a training job while the dispatcher still holds the previous registration (stale/duplicate client IDs); num_clients/world_size changed without restarting the dispatcher; dispatcher restarted mid-run losing state inconsistently; version skew changing registration semantics.

Related errors

AI-assisted analysis of xai-org/x-algorithm@24c60942c5 (2026-08-28). Data as JSON: /api/errors/709118226ed859cb. Report an issue: GitHub.