{"record":{"id":"0ecc63121475a5e3","repo":"t8y2/dbx","slug":"producer-is-not-initialized-call-connect-first","errorCode":null,"errorMessage":"Producer is not initialized. Call connect first.","messagePattern":"Producer is not initialized\\. Call connect first\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java","lineNumber":2269,"sourceCode":"        Map<String, String> headers = new LinkedHashMap<>();\n        record.headers().forEach(h ->\n            headers.put(h.key(), h.value() == null ? \"\" : new String(h.value(), StandardCharsets.UTF_8)));\n        msg.put(\"headers\", headers);\n        if (record.value() != null) {\n            msg.put(\"payloadBase64\", Base64.getEncoder().encodeToString(record.value()));\n            String text = tryDecodeUtf8(record.value());\n            if (text != null) {\n                msg.put(\"payloadText\", text);\n            }\n        } else {\n            msg.put(\"payloadBase64\", \"\");\n        }\n        return msg;\n    }\n\n    private static Object sendMessage(JsonObject params) throws Exception {\n        if (producer == null) {\n            throw new IllegalStateException(\"Producer is not initialized. Call connect first.\");\n        }\n\n        String topic = stringOrEmpty(params, \"topic\");\n        String key = params.has(\"key\") && !params.get(\"key\").isJsonNull()\n            ? params.get(\"key\").getAsString() : null;\n\n        // Decode payload from base64\n        String payloadBase64 = stringOrEmpty(params, \"payloadBase64\");\n        byte[] value = payloadBase64.isEmpty() ? new byte[0] : Base64.getDecoder().decode(payloadBase64);\n\n        // Build the record\n        Integer partition = params.has(\"partition\") && !params.get(\"partition\").isJsonNull()\n            ? params.get(\"partition\").getAsInt() : null;\n\n        ProducerRecord<String, byte[]> record;\n        if (partition != null) {\n            record = new ProducerRecord<>(topic, partition, key, value);\n        } else {","sourceCodeStart":2251,"sourceCodeEnd":2287,"githubUrl":"https://github.com/t8y2/dbx/blob/c0390bff16418b651f4728520d99adf8ce48829a/agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java#L2251-L2287","documentation":"sendMessage requires the agent's static Kafka producer, which is only created by connect(). If producer is null the agent was never connected (or was closed), so sending is impossible and the agent throws IllegalStateException.","triggerScenarios":"Invoking the send-message tool (sendMessage) before connect() succeeded, after a failed connect, or after disconnect/close nulled the producer.","commonSituations":"Calling send before the connect step in a tool sequence; connect failed silently on bad bootstrap servers; agent restarted/reconnected elsewhere and the producer was reset; race where two calls race disconnect and send.","solutions":["Call the kafka connect tool with valid bootstrap servers and wait for success before sending.","Check the connect result/error to ensure bootstrap servers, auth, and TLS settings are correct.","Reconnect if the session was closed, then retry the send.","Guard your workflow so send is only invoked after a confirmed connect."],"exampleFix":"// before\nagent.call(\"kafka.send\", {\"topic\": \"t\", \"value\": \"hi\"})\n// after\nagent.call(\"kafka.connect\", {\"bootstrap_servers\": \"broker:9092\"})\nagent.call(\"kafka.send\", {\"topic\": \"t\", \"value\": \"hi\"})","handlingStrategy":"try-catch","validationCode":"// ensure connect succeeded before sending\nif (!agent.isConnected()) {\n    agent.call(\"kafka.connect\", cfg); // check result for errors\n}","typeGuard":null,"tryCatchPattern":"try {\n    agent.sendMessage(params);\n} catch (IllegalStateException e) {\n    if (e.getMessage().contains(\"Producer is not initialized\")) {\n        agent.connect(cfg);\n        agent.sendMessage(params); // retry once after connect\n    } else { throw e; }\n}","preventionTips":["Always run connect before any send in your workflow.","Verify connect's result for auth/bootstrap errors that leave the producer uninitialized.","Treat disconnect/reconnect as invalidating outstanding send plans.","Check isConnected-style state before each send in long-lived sessions."],"tags":["kafka","lifecycle","not-initialized","state"],"backgroundTag":"not-connected-client","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"}