What causes "cannot create a %s reader with an empty topic" in crowdsecurity/crowdsec?

Add the topic key with your Kafka topic name: topic: crowdsec-alerts.

cannot create a %s reader with an empty topic

Error message

cannot create a %s reader with an empty topic

What it means

The kafka source requires a topic to consume from. If 'topic' is empty or missing after the brokers check passes, UnmarshalConfig rejects the config with this message (%s is the source name, 'kafka'). Unlike mode, which defaults to tail, topic has no default because it is environment-specific.

Source

Thrown at pkg/acquisition/modules/kafka/config.go:60

	BatchMaxWait   time.Duration `yaml:"max_wait"`
	BatchQueueSize int           `yaml:"queue_size"`
	CommitInterval time.Duration `yaml:"commit_interval"`
}

func (s *Source) UnmarshalConfig(yamlConfig []byte) error {
	s.Config = Configuration{}

	err := yaml.UnmarshalWithOptions(yamlConfig, &s.Config, yaml.Strict())
	if err != nil {
		return fmt.Errorf("cannot parse: %s", yaml.FormatError(err, false, false))
	}

	if len(s.Config.Brokers) == 0 {
		return fmt.Errorf("cannot create a %s reader with an empty list of broker addresses", s.GetName())
	}

	if s.Config.Topic == "" {
		return fmt.Errorf("cannot create a %s reader with an empty topic", s.GetName())
	}

	if s.Config.Mode == "" {
		s.Config.Mode = configuration.TAIL_MODE
	}

	s.logger.Debugf("successfully parsed kafka configuration : %+v", s.Config)

	return err
}

func (s *Source) Configure(_ context.Context, yamlConfig []byte, logger *log.Entry, metricsLevel metrics.AcquisitionMetricsLevel) error {
	s.logger = logger
	s.metricsLevel = metricsLevel

	s.logger.Debugf("start configuring %s source", s.GetName())

	err := s.UnmarshalConfig(yamlConfig)

View on GitHub (pinned to 909b515798)

Solutions

Add the topic key with your Kafka topic name: topic: crowdsec-alerts.
Verify the topic exists in the cluster (kafka-topics.sh --list) to avoid a later runtime failure.
Check templating/env substitution actually fills the topic value.
Ensure the topic key is not commented out in acquis.yaml.

Example fix

// before
source: kafka
brokers:
  - localhost:9092
# topic: crowdsec
// after
source: kafka
brokers:
  - localhost:9092
topic: crowdsec

Defensive patterns

Strategy: validation

Validate before calling

var kc struct {
    Topic string `yaml:"topic"`
}
if err := yaml.Unmarshal(yamlCfg, &kc); err != nil { return err }
if strings.TrimSpace(kc.Topic) == "" {
    return errors.New("kafka source requires a non-empty topic")
}

Try / catch

if err := source.UnmarshalConfig(cfg); err != nil {
    if strings.Contains(err.Error(), "empty topic") {
        // set topic from env/config and retry
    }
}

Prevention

Always set topic in kafka acquis blocks — it has no default
Confirm the topic exists in the cluster with kafka-topics.sh --list
Check templating actually fills the topic value; avoid leaving the key commented

When it happens

Trigger: Source.UnmarshalConfig receiving a kafka YAML block with no 'topic' key or topic: "" while brokers are correctly set.

Common situations: Following a broker-only example; renaming the topic and leaving the old key commented out; templating that rendered an empty topic value.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

AI-assisted analysis of crowdsecurity/crowdsec@909b515798 (2026-09-06). Data as JSON: /api/errors/a35c17ec722924bc. Report an issue: GitHub.