Exactly one of Topic or Subscription must be set in pubsubio.go

learning/tour-of-beam/learning-content/io/kafka-io/kafka-write/description.md

@@ -28,7 +28,7 @@ When writing data to Kafka using Apache Beam, it is important to ensure that the
 For detailed [information](https://beam.apache.org/releases/javadoc/2.0.0/org/apache/beam/sdk/io/kafka/KafkaIO.html)
 {{if (eq .Sdk "go")}}
 ```
-data := pubsubio.Read(s, "pubsub-public-data", "taxirides-realtime", nil)
+data := pubsubio.Read(s, "pubsub-public-data", &pubsubio.ReadOptions{Topic: "taxirides-realtime"})
 kvData := beam.ParDo(s, func(elm []byte) ([]byte, []byte) { return []byte(""), elm }, data)
 windowed := beam.WindowInto(s, window.NewFixedWindows(15*time.Second), kvData)
 kafkaio.Write(s, *expansionAddr, *bootstrapServers, *topic, windowed)

learning/tour-of-beam/learning-content/io/kafka-io/kafka-write/go-example/main.go

@@ -76,7 +76,7 @@ func main() {
 
 		// In the main function, the code creates a Beam pipeline, reads from the Pub/Sub source, transforms the data into a key-value pair, applies a windowing function to the data, and writes the windowed data to a Kafka topic.
 
-		data := pubsubio.Read(s, "pubsub-public-data", "taxirides-realtime", nil)
+		data := pubsubio.Read(s, "pubsub-public-data", &pubsubio.ReadOptions{Topic: "taxirides-realtime"})
 		kvData := beam.ParDo(s, func(elm []byte) ([]byte, []byte) { return []byte(""), elm }, data)
 		windowed := beam.WindowInto(s, window.NewFixedWindows(15*time.Second), kvData)
 		kafkaio.Write(s, *expansionAddr, *bootstrapServers, *topic, windowed)

sdks/go/examples/fhirio/read_write_pubsub/read_write_pubsub.go

@@ -143,7 +143,9 @@ func main() {
 		// PubSub notifications will be emitted containing the path of the resource once
 		// it is written to the store. Simultaneously read notifications and resources
 		// from PubSub and store, respectively.
-		resourceNotifications := pubsubio.Read(s, *gcpopts.Project, *pubsubTopic, nil)
+		resourceNotifications := pubsubio.Read(s, *gcpopts.Project, &pubsubio.ReadOptions{
+			Topic: *pubsubTopic,
+		})
 		resourcesInFhirStore, deadLetters := fhirio.Read(s, resourceNotifications)
 
 		// Log the read resources or read errors to the server.

sdks/go/examples/kafka/taxi.go

@@ -155,7 +155,10 @@ func main() {
 	s := p.Root()
 
 	// Read from Pubsub and write to Kafka.
-	data := pubsubio.Read(s, "pubsub-public-data", "taxirides-realtime", nil)
+	opts := &pubsubio.ReadOptions{
+		Topic: "taxirides-realtime",
+	}
+	data := pubsubio.Read(s, "pubsub-public-data", opts)
 	kvData := beam.ParDo(s, func(elm []byte) ([]byte, []byte) { return []byte(""), elm }, data)
 	windowed := beam.WindowInto(s, window.NewFixedWindows(15*time.Second), kvData)
 	kafkaio.Write(s, *expansionAddr, *bootstrapServers, *topic, windowed)

sdks/go/examples/slowly_updating_side_input/slowly_updating_side_input.go

@@ -103,7 +103,9 @@ func main() {
 	_, err = pubsubx.EnsureTopic(ctx, client, inputTopic)
 	fatalf(err, "Failed to ensure topic: %v", err)
 
-	source := pubsubio.Read(s, project, inputTopic, nil)
+	source := pubsubio.Read(s, project, &pubsubio.ReadOptions{
+		Topic: inputTopic,
+	})
 	keyedSource := beam.AddFixedKey(s, source) // simulate keyed data by adding a fixed key
 	mainInput := beam.WindowInto(
 		s,

sdks/go/examples/streaming_wordcap/wordcap.go

@@ -70,7 +70,7 @@ func main() {
 	p := beam.NewPipeline()
 	s := p.Root()
 
-	col := pubsubio.Read(s, project, *input, &pubsubio.ReadOptions{Subscription: sub.ID()})
+	col := pubsubio.Read(s, project, &pubsubio.ReadOptions{Subscription: sub.ID()})
 	str := beam.ParDo(s, func(b []byte) string {
 		return (string)(b)
 	}, col)

sdks/go/pkg/beam/io/pubsubio/pubsubio.go

@@ -52,32 +52,50 @@ func init() {
 
 // ReadOptions represents options for reading from PubSub.
 type ReadOptions struct {
+	Topic              string
 	Subscription       string
 	IDAttribute        string
 	TimestampAttribute string
 	WithAttributes     bool
 }
 
 // Read reads an unbounded number of PubSubMessages from the given
-// pubsub topic. It produces an unbounded PCollecton<*PubSubMessage>,
+// pubsub topic or subscription. It produces an unbounded PCollecton<*PubSubMessage>,
 // if WithAttributes is set, or an unbounded PCollection<[]byte>.
-func Read(s beam.Scope, project, topic string, opts *ReadOptions) beam.PCollection {
+func Read(s beam.Scope, project string, opts *ReadOptions) beam.PCollection {
 	s = s.Scope("pubsubio.Read")
 
-	payload := &pipepb.PubSubReadPayload{
-		Topic: pubsubx.MakeQualifiedTopicName(project, topic),
+	if opts == nil {
+		panic("ReadOptions must not be nil")
 	}
-	if opts != nil {
-		payload.IdAttribute = opts.IDAttribute
-		payload.TimestampAttribute = opts.TimestampAttribute
-		if opts.Subscription != "" {
-			payload.Subscription = pubsubx.MakeQualifiedSubscriptionName(project, opts.Subscription)
-		}
-		payload.WithAttributes = opts.WithAttributes
+
+	// Validate: only one of Topic or Subscription should be set
+	if (opts.Topic == "" && opts.Subscription == "") || (opts.Topic != "" && opts.Subscription != "") {
+		panic("Exactly one of Topic or Subscription must be set in ReadOptions")
 	}
 
-	out := beam.External(s, readURN, protox.MustEncode(payload), nil, []beam.FullType{typex.New(reflectx.ByteSlice)}, false)
-	if opts != nil && opts.WithAttributes {
+	payload := &pipepb.PubSubReadPayload{}
+
+	if opts.Topic != "" {
+		payload.Topic = pubsubx.MakeQualifiedTopicName(project, opts.Topic)
+	} else {
+		payload.Subscription = pubsubx.MakeQualifiedSubscriptionName(project, opts.Subscription)
+	}
+
+	payload.IdAttribute = opts.IDAttribute
+	payload.TimestampAttribute = opts.TimestampAttribute
+	payload.WithAttributes = opts.WithAttributes
+
+	out := beam.External(
+		s,
+		readURN,
+		protox.MustEncode(payload),
+		nil,
+		[]beam.FullType{typex.New(reflectx.ByteSlice)},
+		false,
+	)
+
+	if opts.WithAttributes {
 		return beam.ParDo(s, unmarshalMessageFn, out[0])
 	}
 	return out[0]

sdks/go/pkg/beam/io/pubsubio/pubsubio_test.go

@@ -0,0 +1,71 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package pubsubio
+
+import (
+	"github.com/apache/beam/sdks/v2/go/pkg/beam"
+	"testing"
+)
+
+func TestRead_NilOptionsPanics(t *testing.T) {
+	defer func() {
+		if r := recover(); r == nil {
+			t.Fatal("expected panic when opts is nil")
+		}
+	}()
+
+	beam.Init()
+
+	p := beam.NewPipeline()
+	s := p.Root()
+
+	Read(s, "test-project", nil)
+}
+
+func TestRead_BothTopicAndSubscriptionPanics(t *testing.T) {
+	defer func() {
+		if r := recover(); r == nil {
+			t.Fatal("expected panic when both topic and subscription are set")
+		}
+	}()
+
+	beam.Init()
+
+	p := beam.NewPipeline()
+	s := p.Root()
+
+	opts := &ReadOptions{
+		Topic:        "topic",
+		Subscription: "sub",
+	}
+	Read(s, "test-project", opts)
+}
+
+func TestRead_NeitherTopicNorSubscriptionPanics(t *testing.T) {
+	defer func() {
+		if r := recover(); r == nil {
+			t.Fatal("expected panic when neither topic nor subscription is set")
+		}
+	}()
+
+	beam.Init()
+
+	p := beam.NewPipeline()
+	s := p.Root()
+
+	opts := &ReadOptions{}
+	Read(s, "test-project", opts)
+}