What causes "topic %s not found" in openimsdk/open-im-server?

Add the topic to the kafka topics configuration so the builder registers it

topic %s not found

Error message

topic %s not found

What it means

kafkaBuilder.GetTopicProducer looks up the logical topic name in its pre-registered logicalTopic map; if the topic was never registered with the builder, it returns 'topic %s not found'. Only topics declared in the Kafka topics config are resolvable.

Source

Thrown at pkg/mqbuild/builder.go:158

func (standaloneBuilder) GetTopicProducer(ctx context.Context, topic string) (mq.Producer, error) {
	return simmq.GetTopicProducer(topic), nil
}

func (standaloneBuilder) GetTopicConsumer(ctx context.Context, topic string) (mq.Consumer, error) {
	return simmq.GetTopicConsumer(topic), nil
}

type kafkaBuilder struct {
	addr         []string
	config       *kafka.Config
	logicalTopic map[string]string
	topicGroupID map[string]string
}

func (x *kafkaBuilder) GetTopicProducer(ctx context.Context, topic string) (mq.Producer, error) {
	realTopic, ok := x.logicalTopic[topic]
	if !ok {
		return nil, fmt.Errorf("topic %s not found", topic)
	}
	return kafka.NewKafkaProducerV2(x.config, x.addr, realTopic)
}

func (x *kafkaBuilder) GetTopicConsumer(ctx context.Context, topic string) (mq.Consumer, error) {
	realTopic, ok := x.logicalTopic[topic]
	if !ok {
		return nil, fmt.Errorf("topic %s not found", topic)
	}
	groupID, ok := x.topicGroupID[realTopic]
	if !ok {
		return nil, fmt.Errorf("topic %s groupID not found", realTopic)
	}
	return kafka.NewMConsumerGroupV2(ctx, x.config, groupID, []string{realTopic}, true)
}

View on GitHub (pinned to 175a7bb067)

Solutions

Add the topic to the kafka topics configuration so the builder registers it
Correct the topic name passed to GetTopicProducer
Rebuild/merge topics (MergeTopics) before constructing producers

Example fix

// before
kafka:
  topics: ["msgToMq"]
// after
kafka:
  topics: ["msgToMq", "newTopic"]

Defensive patterns

Strategy: validation

Validate before calling

func topicRegistered(builder *mqbuild.KafkaBuilderTopics, topic string) bool {
    for _, t := range cfg.Queue.Kafka.Topics {
        if t == topic { return true }
    }
    return false
}
if !topicRegistered(nil, myTopic) {
    return fmt.Errorf("topic %q must be added to kafka.topics config", myTopic)
}

Try / catch

p, err := builder.GetTopicProducer(ctx, topic)
if err != nil && strings.Contains(err.Error(), "not found") {
    // fail fast: topic missing from config
}

Prevention

Keep a single source-of-truth topic constant shared by producer and config
Add a startup assert that all referenced topics exist in config
Review topic lists when adding new producers

When it happens

Trigger: Calling GetTopicProducer with a topic string that MergeTopics/KafkaTopics did not register — typically a topic missing from the kafka.topics config.

Common situations: Code publishes to a new logical topic but ops forgot to add it to config; renamed topic in code without updating config; typo in topic constant.

Related errors

AI-assisted analysis of openimsdk/open-im-server@175a7bb067 (2026-09-04). Data as JSON: /api/errors/a2133ad39a440a82. Report an issue: GitHub.