{"record":{"id":"045fa759ebd5465c","repo":"quickwit-oss/quickwit","slug":"position-of-a-kafka-partition-should-never-be-eof","errorCode":null,"errorMessage":"position of a Kafka partition should never be EOF","messagePattern":"position of a Kafka partition should never be EOF","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"quickwit/quickwit-indexing/src/source/kafka_source.rs","lineNumber":372,"sourceCode":"\n            self.state\n                .assigned_partitions\n                .insert(partition, partition_id.clone());\n\n            let Some(current_position) = checkpoint.position_for_partition(&partition_id).cloned()\n            else {\n                continue;\n            };\n            let next_offset = match &current_position {\n                Position::Beginning => Offset::Beginning,\n                Position::Offset(offset) => {\n                    let offset = offset\n                        .as_i64()\n                        .expect(\"Kafka offset should be stored as i64\");\n                    Offset::Offset(offset + 1)\n                }\n                Position::Eof(_) => {\n                    panic!(\"position of a Kafka partition should never be EOF\")\n                }\n            };\n            self.state\n                .current_positions\n                .insert(partition, current_position);\n            next_offsets.push((partition, next_offset));\n        }\n        info!(\n            index_id=%self.source_runtime.index_id(),\n            source_id=%self.source_runtime.source_id(),\n            topic=%self.topic,\n            group_id=%self.group_id,\n            partitions=?partitions,\n            \"new partition assignment after rebalance\",\n        );\n        assignment_tx\n            .send(next_offsets)\n            .context(\"Kafka consumer context was dropped\")?;","sourceCodeStart":354,"sourceCodeEnd":390,"githubUrl":"https://github.com/quickwit-oss/quickwit/blob/a39730c5cdcd1a4fe798403737ae293999ea21f8/quickwit/quickwit-indexing/src/source/kafka_source.rs#L354-L390","documentation":"build_ingester_service wires the ingester service client. When the target node is the local node, the code expects the locally initialized ingester service to have been passed in via `ingester_opt`; `.expect` fires if it is None. This is an internal wiring invariant: on a self-node the service must already have been created during node startup.","triggerScenarios":"Calling build_ingester_insert_change / build_ingester_service for a self-node while the ingester service was never initialized (e.g. the node config does not enable the ingester role but some code path still asks for the local ingester client).","commonSituations":"Running a node whose config omits the ingester role while an index/ingest flow still routes to the local node; a code path or test that builds the service registry before the ingester actor is set up; misconfigured peer routing where a request thought to be local is resolved locally.","solutions":["Add the ingester role to the node configuration so the ingester service is initialized at startup.","Fix the routing/placement logic so ingest requests for a node without an ingester go over gRPC to a remote ingester instead of the local shortcut.","Check node_config / peer seed setup: `is_self_node()` must only be true for nodes that actually run the ingester.","Verify cluster membership data (Chitchat) is not stale, which can make a remote node appear as self."],"exampleFix":"// before\nlet ingester = ingester_opt.expect(\"ingester service should be initialized\");\n// after\nlet ingester = ingester_opt.ok_or_else(|| {\n    anyhow::anyhow!(\"ingester service is not initialized on this node; is the ingester role enabled?\")\n})?;","handlingStrategy":"validation","validationCode":"// before bootstrapping ingest flows, verify the node runs the ingester role\nlet roles = &config.node_config.roles();\nif !roles.contains(Role::Ingest) {\n    return Err(anyhow!(\"node does not run the ingester; cannot use local ingester client\"));\n}","typeGuard":null,"tryCatchPattern":"// callers of build_ingester_insert_change should map panics to startup failure:\nlet client = std::panic::catch_unwind(|| build_ingester_insert_change(...));","preventionTips":["Enable the ingester role on any node expected to serve ingest locally.","Only route ingest requests to nodes advertising the ingester role in Chitchat.","Review placement/config so `is_self_node()` is only true for ingester nodes.","Test multi-role and single-role node configurations in CI."],"tags":["rust","grpc","ingest","service-initialization","internal-invariant"],"backgroundTag":"missing-required-config","analyzedSha":"a39730c5cdcd1a4fe798403737ae293999ea21f8","analyzedAt":"2026-09-08T13:19:37.784Z","contentChangedAt":"2026-09-08T13:19:37.784Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}