{"record":{"id":"bdb25309c8cf9e9a","repo":"apache/beam","slug":"kafkaio-read-requires-at-least-one-topic-to-read-from","errorCode":null,"errorMessage":"kafkaio.Read requires at least one topic to read from.","messagePattern":"kafkaio\\.Read requires at least one topic to read from\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/io/xlang/kafkaio/kafka.go","lineNumber":120,"sourceCode":"// service will be automatically started; however this is slower than having a\n// persistent expansion service running.\n//\n// Read also accepts optional parameters as readOptions. All optional parameters\n// are predefined in this package as functions that return readOption. To set\n// an optional parameter, call the function within Read's function signature.\n//\n// Example of Read with required and optional parameters:\n//\n//\texpansionAddr := \"localhost:1234\"\n//\tbootstrapServer := \"bootstrap-server:1234\"\n//\ttopic := \"topic_name\"\n//\tpcol := kafkaio.Read( s, expansionAddr, bootstrapServer, []string{topic},\n//\t    kafkaio.MaxNumRecords(100), kafkaio.CommitOffsetInFinalize(true))\nfunc Read(s beam.Scope, addr string, servers string, topics []string, opts ...readOption) beam.PCollection {\n\ts = s.Scope(\"kafkaio.Read\")\n\n\tif len(topics) == 0 {\n\t\tpanic(\"kafkaio.Read requires at least one topic to read from.\")\n\t}\n\n\tif addr == \"\" {\n\t\taddr = autoStartupAddress\n\t}\n\n\trpl := readPayload{\n\t\tConsumerConfig:    map[string]string{\"bootstrap.servers\": servers},\n\t\tTopics:            topics,\n\t\tKeyDeserializer:   ByteArrayDeserializer,\n\t\tValueDeserializer: ByteArrayDeserializer,\n\t\tTimestampPolicy:   string(ProcessingTime),\n\t}\n\trcfg := readConfig{\n\t\tpl:  &rpl,\n\t\tkey: reflectx.ByteSlice,\n\t\tval: reflectx.ByteSlice,\n\t}","sourceCodeStart":102,"sourceCodeEnd":138,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/io/xlang/kafkaio/kafka.go#L102-L138","documentation":"kafkaio.Read in the Beam Go xlang package panics when the topics slice is empty. A Kafka read must subscribe to at least one topic, so an empty topic list is rejected immediately during pipeline construction.","triggerScenarios":"Calling kafkaio.Read(s, addr, servers, []string{}) or nil, or passing a topics slice built from an empty config/env value.","commonSituations":"Topics loaded from a comma-separated env var or flag that was empty; config where the topic key was omitted; building the slice with strings.Split(\"\", \",\") which yields an empty list.","solutions":["Pass at least one topic name: kafkaio.Read(s, addr, servers, []string{\"my-topic\"}).","Guard before the call: if len(topics) == 0 { return error }.","Fix upstream config parsing so an unset topic variable surfaces as a clear config error rather than an empty slice."],"exampleFix":"// before\nkafkaio.Read(s, addr, servers, strings.Split(os.Getenv(\"TOPICS\"), \",\"))\n// after\ntopics := strings.Split(os.Getenv(\"TOPICS\"), \",\")\nif len(topics) == 0 { log.Fatalf(\"TOPICS must specify at least one topic\") }\nkafkaio.Read(s, addr, servers, topics)","handlingStrategy":"validation","validationCode":"if len(topics) == 0 {\n    return errors.New(\"kafkaio.Read needs at least one topic\")\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Validate topic flags/env vars at startup, before pipeline construction.","Remember strings.Split of an empty string yields an empty slice — check length.","Add a startup assertion that required Kafka config (servers and topics) is non-empty."],"tags":["go","apache-beam","kafka","empty-argument","panic"],"backgroundTag":"missing-required-argument","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}