t8y2/dbx · error · IllegalStateException

Kafka Agent is not connected

Error message

Kafka Agent is not connected

What it means

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.

Source

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

            List<PartitionInfo> partitions = topics.get(name);
            List<TopicPartition> topicPartitions = partitions.stream()
                .map(partition -> new TopicPartition(name, partition.partition()))
                .collect(Collectors.toList());
            Map<TopicPartition, Long> beginningOffsets = consumer.beginningOffsets(topicPartitions, requestTimeout);
            Map<TopicPartition, Long> endOffsets = consumer.endOffsets(topicPartitions, requestTimeout);
            return legacyTopicStatsResult(name, partitions, beginningOffsets, endOffsets);
        }
    }

    static void requireExistingTopic(Collection<String> topicNames, String name) {
        if (!topicNames.contains(name)) {
            throw new UnknownTopicOrPartitionException("Kafka topic does not exist: " + name);
        }
    }

    static Properties topicStatsConsumerProperties(JsonObject conn) {
        if (conn == null) {
            throw new IllegalStateException("Kafka Agent is not connected");
        }
        Properties props = peekConsumerProperties(conn, 1);
        // The fallback discovers the topic through all-topics metadata, which cannot create
        // a topic, before issuing requests for its concrete partitions.
        props.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, "true");
        return props;
    }

    static Object legacyTopicStatsResult(
        String name,
        List<PartitionInfo> partitions,
        Map<TopicPartition, Long> beginningOffsets,
        Map<TopicPartition, Long> endOffsets
    ) {
        long totalMessages = 0;
        List<Map<String, Object>> partitionStats = new ArrayList<>();
        for (PartitionInfo partition : partitions) {
            TopicPartition topicPartition = new TopicPartition(name, partition.partition());

View on GitHub (pinned to c0390bff16)

Solutions

  1. Call the agent's connect operation with valid Kafka connection config before requesting topic stats.
  2. Verify the connect call succeeded (check for swallowed errors at startup) and that the connection object was actually stored on the agent.
  3. Confirm the connection config source (env vars, config file) is populated in the running environment.
  4. If the connection can be lost, re-establish it or guard calls with an is-connected check before invoking stats operations.

Example fix

// before
Map<String, Object> stats = agent.topicStats("events", 5000); // conn == null
// after
agent.connect(kafkaConfig); // ensure successful connect first
Map<String, Object> stats = agent.topicStats("events", 5000);
Defensive patterns

Strategy: validation

When it happens

Trigger: 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.

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

Related errors

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