{"record":{"id":"a35c17ec722924bc","repo":"crowdsecurity/crowdsec","slug":"cannot-create-a-s-reader-with-an-empty-topic","errorCode":null,"errorMessage":"cannot create a %s reader with an empty topic","messagePattern":"cannot create a (.+?) reader with an empty topic","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"pkg/acquisition/modules/kafka/config.go","lineNumber":60,"sourceCode":"\tBatchMaxWait   time.Duration `yaml:\"max_wait\"`\n\tBatchQueueSize int           `yaml:\"queue_size\"`\n\tCommitInterval time.Duration `yaml:\"commit_interval\"`\n}\n\nfunc (s *Source) UnmarshalConfig(yamlConfig []byte) error {\n\ts.Config = Configuration{}\n\n\terr := yaml.UnmarshalWithOptions(yamlConfig, &s.Config, yaml.Strict())\n\tif err != nil {\n\t\treturn fmt.Errorf(\"cannot parse: %s\", yaml.FormatError(err, false, false))\n\t}\n\n\tif len(s.Config.Brokers) == 0 {\n\t\treturn fmt.Errorf(\"cannot create a %s reader with an empty list of broker addresses\", s.GetName())\n\t}\n\n\tif s.Config.Topic == \"\" {\n\t\treturn fmt.Errorf(\"cannot create a %s reader with an empty topic\", s.GetName())\n\t}\n\n\tif s.Config.Mode == \"\" {\n\t\ts.Config.Mode = configuration.TAIL_MODE\n\t}\n\n\ts.logger.Debugf(\"successfully parsed kafka configuration : %+v\", s.Config)\n\n\treturn err\n}\n\nfunc (s *Source) Configure(_ context.Context, yamlConfig []byte, logger *log.Entry, metricsLevel metrics.AcquisitionMetricsLevel) error {\n\ts.logger = logger\n\ts.metricsLevel = metricsLevel\n\n\ts.logger.Debugf(\"start configuring %s source\", s.GetName())\n\n\terr := s.UnmarshalConfig(yamlConfig)","sourceCodeStart":42,"sourceCodeEnd":78,"githubUrl":"https://github.com/crowdsecurity/crowdsec/blob/909b5157986a2b2c2163300fdaef5ed01289f7d2/pkg/acquisition/modules/kafka/config.go#L42-L78","documentation":"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.","triggerScenarios":"Source.UnmarshalConfig receiving a kafka YAML block with no 'topic' key or topic: \"\" while brokers are correctly set.","commonSituations":"Following a broker-only example; renaming the topic and leaving the old key commented out; templating that rendered an empty topic value.","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."],"exampleFix":"// before\nsource: kafka\nbrokers:\n  - localhost:9092\n# topic: crowdsec\n// after\nsource: kafka\nbrokers:\n  - localhost:9092\ntopic: crowdsec","handlingStrategy":"validation","validationCode":"var kc struct {\n    Topic string `yaml:\"topic\"`\n}\nif err := yaml.Unmarshal(yamlCfg, &kc); err != nil { return err }\nif strings.TrimSpace(kc.Topic) == \"\" {\n    return errors.New(\"kafka source requires a non-empty topic\")\n}","typeGuard":null,"tryCatchPattern":"if err := source.UnmarshalConfig(cfg); err != nil {\n    if strings.Contains(err.Error(), \"empty topic\") {\n        // set topic from env/config and retry\n    }\n}","preventionTips":["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"],"tags":["go","kafka","configuration","topic"],"backgroundTag":"missing-required-config-field","analyzedSha":"909b5157986a2b2c2163300fdaef5ed01289f7d2","analyzedAt":"2026-09-06T12:27:26.012Z","contentChangedAt":"2026-09-06T12:27:26.012Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}