kafkaio.Read requires at least one topic to read from.

Error message

kafkaio.Read requires at least one topic to read from.

What it means

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.

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.

Example fix

// before
kafkaio.Read(s, addr, servers, strings.Split(os.Getenv("TOPICS"), ","))
// after
topics := strings.Split(os.Getenv("TOPICS"), ",")
if len(topics) == 0 { log.Fatalf("TOPICS must specify at least one topic") }
kafkaio.Read(s, addr, servers, topics)

Defensive patterns

Strategy: validation

Validate before calling

if len(topics) == 0 {
    return errors.New("kafkaio.Read needs at least one topic")
}

Prevention

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.

When it happens

Trigger: Calling kafkaio.Read(s, addr, servers, []string{}) or nil, or passing a topics slice built from an empty config/env value.

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

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/bdb25309c8cf9e9a. Report an issue: GitHub.

Appendix: source

Thrown at sdks/go/pkg/beam/io/xlang/kafkaio/kafka.go:120

// service will be automatically started; however this is slower than having a
// persistent expansion service running.
//
// Read also accepts optional parameters as readOptions. All optional parameters
// are predefined in this package as functions that return readOption. To set
// an optional parameter, call the function within Read's function signature.
//
// Example of Read with required and optional parameters:
//
//	expansionAddr := "localhost:1234"
//	bootstrapServer := "bootstrap-server:1234"
//	topic := "topic_name"
//	pcol := kafkaio.Read( s, expansionAddr, bootstrapServer, []string{topic},
//	    kafkaio.MaxNumRecords(100), kafkaio.CommitOffsetInFinalize(true))
func Read(s beam.Scope, addr string, servers string, topics []string, opts ...readOption) beam.PCollection {
	s = s.Scope("kafkaio.Read")

	if len(topics) == 0 {
		panic("kafkaio.Read requires at least one topic to read from.")
	}

	if addr == "" {
		addr = autoStartupAddress
	}

	rpl := readPayload{
		ConsumerConfig:    map[string]string{"bootstrap.servers": servers},
		Topics:            topics,
		KeyDeserializer:   ByteArrayDeserializer,
		ValueDeserializer: ByteArrayDeserializer,
		TimestampPolicy:   string(ProcessingTime),
	}
	rcfg := readConfig{
		pl:  &rpl,
		key: reflectx.ByteSlice,
		val: reflectx.ByteSlice,
	}

View on GitHub (pinned to 12126d8942)