{"record":{"id":"3c4222fccb0b1c13","repo":"xai-org/x-algorithm","slug":"invalid-num-servers-or-num-processors-in-che","errorCode":null,"errorMessage":"Invalid {num_servers=} or {num_processors=} in CheckStateResponse","messagePattern":"Invalid (.+?) or (.+?) in CheckStateResponse","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"phoenix/xrex/data/streaming/kafkadispatcherloader.py","lineNumber":64,"sourceCode":"    (\"grpc.keepalive_timeout_ms\", 10000),\n    (\"grpc.http2.max_pings_without_data\", 0),\n]\n\n\ndef get_grpc_channel(grpc_host: str, grpc_port) -> grpc.aio.Channel:\n    target = f\"{grpc_host}:{grpc_port}\"\n    return grpc.aio.insecure_channel(target, options=OPTIONS)\n\n\nasync def get_service_dimensions(grpc_host_template: str, grpc_port: int) -> tuple[int, int]:\n    master_node_host = grpc_host_template.format(shard_index=\"0\")\n    async with get_grpc_channel(master_node_host, grpc_port) as channel:\n        stub = KafkaDispatcherStub(channel)\n        response = await stub.CheckState(CheckStateRequest())\n        num_servers = response.num_shards\n        num_processors = response.num_processors\n        if num_servers is None or num_processors is None:\n            raise ValueError(f\"Invalid {num_servers=} or {num_processors=} in CheckStateResponse\")\n        return num_servers, num_processors\n\n\nasync def get_server_assignments(\n    grpc_host_template: str,\n    grpc_port: int,\n    num_clients: int,\n    client_ix: int,\n) -> tuple[int, int, list[int]]:\n    num_servers, num_processors = await get_service_dimensions(grpc_host_template, grpc_port)\n    if num_servers is None or num_processors is None:\n        raise ValueError(\"Failed to get num_servers or num_processors from gRPC service\")\n\n    assert num_clients % (num_servers * num_processors) == 0, (\n        f\"Num dataloaders {num_clients} must be a multiple of number of servers {num_servers=} * {num_processors=} = {num_servers * num_processors}\"\n    )\n\n    global_num_processors = num_servers * num_processors","sourceCodeStart":46,"sourceCodeEnd":82,"githubUrl":"https://github.com/xai-org/x-algorithm/blob/24c60942c5c5fdad3a6addffb4c6e6d2f228f04f/phoenix/xrex/data/streaming/kafkadispatcherloader.py#L46-L82","documentation":"get_service_dimensions calls the KafkaDispatcher gRPC service's CheckState and expects both num_shards (servers) and num_processors to be set on the response. If either proto field is unset (proto3 default / None for optional fields), the response is considered invalid for computing the client-to-server sharding layout.","triggerScenarios":"Calling get_server_assignments against a dispatcher service version that does not populate num_shards or num_processors in CheckStateResponse; service still initializing and state not yet computed; connecting to an unexpected/older gRPC endpoint that returns a differently-shaped response.","commonSituations":"Version skew between the Python client and the dispatcher server (new client expects fields the old server never sets); dispatcher pod restarted and reporting empty state; wrong grpc_port hitting a different gRPC service.","solutions":["Check the dispatcher service version/proto definition and upgrade the server so CheckState populates both fields.","Ensure grpc_host_template/grpc_port point at the actual KafkaDispatcher service, not another gRPC endpoint.","If the service may still be booting, retry CheckState with backoff before failing.","Inspect the raw response (log response) to see which field is missing."],"exampleFix":"# before\nresponse = await stub.CheckState(CheckStateRequest())\n\n# after\nresponse = await stub.CheckState(CheckStateRequest())\nrank_logger.info(f\"CheckState response: {response}\")\nfor attempt in range(5):\n    if response.num_shards is not None and response.num_processors is not None:\n        break\n    await asyncio.sleep(2)\n    response = await stub.CheckState(CheckStateRequest())","handlingStrategy":"retry","validationCode":"resp = await stub.CheckState(CheckStateRequest())\nassert resp.num_shards is not None and resp.num_processors is not None","typeGuard":"def check_state_is_valid(resp) -> bool:\n    return resp.num_shards is not None and resp.num_processors is not None","tryCatchPattern":"try:\n    num_servers, num_processors = await get_service_dimensions(host, port)\nexcept ValueError as e:\n    # retry with backoff — service may still be initializing\n    await asyncio.sleep(5)\n    num_servers, num_processors = await get_service_dimensions(host, port)","preventionTips":["Pin client and dispatcher versions together in deploy.","Wait for dispatcher readiness endpoint before consuming.","Log raw gRPC responses during integration debugging."],"tags":["grpc","kafka-dispatcher","version-skew","service-state"],"backgroundTag":"grpc-unset-response-field","analyzedSha":"24c60942c5c5fdad3a6addffb4c6e6d2f228f04f","analyzedAt":"2026-08-28T11:40:14.686Z","schemaVersion":2},"datasetVersion":"2026-08-28T16:17:29.566Z"}