t8y2/dbx · error · IllegalStateException

Producer is not initialized. Call connect first.

Error message

Producer is not initialized. Call connect first.

What it means

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.

Source

Thrown at agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java:2269

        Map<String, String> headers = new LinkedHashMap<>();
        record.headers().forEach(h ->
            headers.put(h.key(), h.value() == null ? "" : new String(h.value(), StandardCharsets.UTF_8)));
        msg.put("headers", headers);
        if (record.value() != null) {
            msg.put("payloadBase64", Base64.getEncoder().encodeToString(record.value()));
            String text = tryDecodeUtf8(record.value());
            if (text != null) {
                msg.put("payloadText", text);
            }
        } else {
            msg.put("payloadBase64", "");
        }
        return msg;
    }

    private static Object sendMessage(JsonObject params) throws Exception {
        if (producer == null) {
            throw new IllegalStateException("Producer is not initialized. Call connect first.");
        }

        String topic = stringOrEmpty(params, "topic");
        String key = params.has("key") && !params.get("key").isJsonNull()
            ? params.get("key").getAsString() : null;

        // Decode payload from base64
        String payloadBase64 = stringOrEmpty(params, "payloadBase64");
        byte[] value = payloadBase64.isEmpty() ? new byte[0] : Base64.getDecoder().decode(payloadBase64);

        // Build the record
        Integer partition = params.has("partition") && !params.get("partition").isJsonNull()
            ? params.get("partition").getAsInt() : null;

        ProducerRecord<String, byte[]> record;
        if (partition != null) {
            record = new ProducerRecord<>(topic, partition, key, value);
        } else {

View on GitHub (pinned to c0390bff16)

Solutions

1. Call the kafka connect tool with valid bootstrap servers and wait for success before sending.
2. Check the connect result/error to ensure bootstrap servers, auth, and TLS settings are correct.
3. Reconnect if the session was closed, then retry the send.
4. Guard your workflow so send is only invoked after a confirmed connect.

Example fix

// before
agent.call("kafka.send", {"topic": "t", "value": "hi"})
// after
agent.call("kafka.connect", {"bootstrap_servers": "broker:9092"})
agent.call("kafka.send", {"topic": "t", "value": "hi"})

// ensure connect succeeded before sending
if (!agent.isConnected()) {
    agent.call("kafka.connect", cfg); // check result for errors
}

try {
    agent.sendMessage(params);
} catch (IllegalStateException e) {
    if (e.getMessage().contains("Producer is not initialized")) {
        agent.connect(cfg);
        agent.sendMessage(params); // retry once after connect
    } else { throw e; }
}

When it happens

Trigger: Invoking the send-message tool (sendMessage) before connect() succeeded, after a failed connect, or after disconnect/close nulled the producer.

Common situations: 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.

Related errors

• Kafka Agent is not connected
• Not connected. Call connect first.
• Not connected
• Agent session not found: <sessionId>
• ETCD_WATCH_NOT_FOUND

AI-assisted analysis of t8y2/dbx@c0390bff16 (2026-09-05). Data as JSON: /api/errors/0ecc63121475a5e3. Report an issue: GitHub.