{"record":{"id":"709118226ed859cb","repo":"xai-org/x-algorithm","slug":"client-registration-failed-register-response-err","errorCode":null,"errorMessage":"Client registration failed: {register_response.error_message}","messagePattern":"Client registration failed: (.+?)","errorType":"exception","errorClass":"Exception","httpStatus":null,"severity":"critical","filePath":"phoenix/xrex/data/streaming/kafkadispatcherloader.py","lineNumber":126,"sourceCode":"            for partition, offset in seek_to_offset.items()\n        ]\n        if seek_to_offset\n        else None\n    )\n    seek_request = SeekRequest()\n    if _seek_to_offset is not None:\n        seek_request.offsets.CopyFrom(SeekOffsets(offsets=_seek_to_offset))\n    elif seek_to_timestamp_ms is not None:\n        seek_request.timestampMs = seek_to_timestamp_ms\n    elif reset_to_latest:\n        seek_request.latest = reset_to_latest\n    register_response = await stub.InitClientMessage(\n        InitClientRequest(register_client_ids=clients_to_register, seek_request=seek_request)\n    )\n    rank_logger.info(f\"Register response: {register_response}\")\n    if not register_response.success:\n        rank_logger.error(f\"Failed to register clients: {register_response.error_message}\")\n        raise Exception(f\"Client registration failed: {register_response.error_message}\")\n\n\nasync def consume_messages(\n    grpc_host_template: str,\n    grpc_port: int,\n    num_shards: int,\n    shard_index: int,\n    topic_name: str,\n    seek_to_offset: dict[int, int] | None,\n    seek_to_timestamp_ms: int | None,\n    reset_to_latest: bool,\n    fetch_batch_size: int,\n    batch_size: int,\n    post_process_fn: Callable[[list[pa.RecordBatch]], RecsysFeaturesBatch],\n    example_queue: queue.Queue[tuple[RecsysFeaturesBatch, dict[int, int]]],\n    _stop_event: threading.Event | None = None,\n    grpc_timeout: float = 30.0,\n):","sourceCodeStart":108,"sourceCodeEnd":144,"githubUrl":"https://github.com/xai-org/x-algorithm/blob/24c60942c5c5fdad3a6addffb4c6e6d2f228f04f/phoenix/xrex/data/streaming/kafkadispatcherloader.py#L108-L144","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"# before\nclients_to_register = [f\"client-{ix}\" for ix in range(num_clients)]\nawait initialize_clients(stub, clients_to_register, seek_request)\n\n# after\nrun_id = os.environ.get(\"RUN_ID\", str(uuid.uuid4())[:8])\nclients_to_register = [f\"client-{run_id}-{ix}\" for ix in range(num_clients)]\nawait initialize_clients(stub, clients_to_register, seek_request)","handlingStrategy":"retry","validationCode":"probe = await stub.InitClientMessage(InitClientRequest(register_client_ids=[], seek_request=None))\n# or verify dispatcher health before registering all clients","typeGuard":null,"tryCatchPattern":"try:\n    await initialize_clients(stub, client_ids, seek_request)\nexcept Exception as e:\n    if \"Client registration failed\" in str(e):\n        # inspect error_message; reset dispatcher state or use fresh IDs, then retry once\n        rank_logger.error(f\"registration rejected: {e}\")\n        raise","preventionTips":["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."],"tags":["grpc","kafka-dispatcher","registration","distributed"],"backgroundTag":"grpc-service-registration-rejected","analyzedSha":"24c60942c5c5fdad3a6addffb4c6e6d2f228f04f","analyzedAt":"2026-08-28T11:40:14.686Z","schemaVersion":2},"datasetVersion":"2026-08-28T16:17:29.566Z"}