{"record":{"id":"7e5ee7dc7b4dd65c","repo":"t8y2/dbx","slug":"kafka-agent-is-not-connected","errorCode":null,"errorMessage":"Kafka Agent is not connected","messagePattern":"Kafka Agent is not connected","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java","lineNumber":809,"sourceCode":"            List<PartitionInfo> partitions = topics.get(name);\n            List<TopicPartition> topicPartitions = partitions.stream()\n                .map(partition -> new TopicPartition(name, partition.partition()))\n                .collect(Collectors.toList());\n            Map<TopicPartition, Long> beginningOffsets = consumer.beginningOffsets(topicPartitions, requestTimeout);\n            Map<TopicPartition, Long> endOffsets = consumer.endOffsets(topicPartitions, requestTimeout);\n            return legacyTopicStatsResult(name, partitions, beginningOffsets, endOffsets);\n        }\n    }\n\n    static void requireExistingTopic(Collection<String> topicNames, String name) {\n        if (!topicNames.contains(name)) {\n            throw new UnknownTopicOrPartitionException(\"Kafka topic does not exist: \" + name);\n        }\n    }\n\n    static Properties topicStatsConsumerProperties(JsonObject conn) {\n        if (conn == null) {\n            throw new IllegalStateException(\"Kafka Agent is not connected\");\n        }\n        Properties props = peekConsumerProperties(conn, 1);\n        // The fallback discovers the topic through all-topics metadata, which cannot create\n        // a topic, before issuing requests for its concrete partitions.\n        props.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, \"true\");\n        return props;\n    }\n\n    static Object legacyTopicStatsResult(\n        String name,\n        List<PartitionInfo> partitions,\n        Map<TopicPartition, Long> beginningOffsets,\n        Map<TopicPartition, Long> endOffsets\n    ) {\n        long totalMessages = 0;\n        List<Map<String, Object>> partitionStats = new ArrayList<>();\n        for (PartitionInfo partition : partitions) {\n            TopicPartition topicPartition = new TopicPartition(name, partition.partition());","sourceCodeStart":791,"sourceCodeEnd":827,"githubUrl":"https://github.com/t8y2/dbx/blob/c0390bff16418b651f4728520d99adf8ce48829a/agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java#L791-L827","documentation":"topicStatsConsumerProperties builds the consumer Properties used for topic stats and requires a non-null connection JsonObject; a null conn means the Kafka Agent has no active connection configured. The agent throws IllegalStateException with this message instead of producing an NPE downstream, telling the caller to connect (or pass a valid connection config) before using topic stats.","triggerScenarios":"Calling a topic stats API that internally resolves the agent's connection, when the connection JsonObject is null — i.e. the agent was never connected, connect() failed earlier, or the connection was cleared/reset before this call.","commonSituations":"Forgetting to call the agent's connect step before issuing stats requests; connect failed silently at startup and the app continued; connection config removed by a reconfigure/reload path; sharing the agent across modules where one module expects another to have connected.","solutions":["Call the agent's connect operation with valid Kafka connection config before requesting topic stats.","Verify the connect call succeeded (check for swallowed errors at startup) and that the connection object was actually stored on the agent.","Confirm the connection config source (env vars, config file) is populated in the running environment.","If the connection can be lost, re-establish it or guard calls with an is-connected check before invoking stats operations."],"exampleFix":"// before\nMap<String, Object> stats = agent.topicStats(\"events\", 5000); // conn == null\n// after\nagent.connect(kafkaConfig); // ensure successful connect first\nMap<String, Object> stats = agent.topicStats(\"events\", 5000);","handlingStrategy":"validation","validationCode":null,"typeGuard":null,"tryCatchPattern":null,"preventionTips":[],"tags":["kafka","connection","lifecycle"],"backgroundTag":"agent-not-connected","analyzedSha":"c0390bff16418b651f4728520d99adf8ce48829a","analyzedAt":"2026-09-05T23:05:10.900Z","contentChangedAt":"2026-09-05T23:05:10.900Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}